input_stream.hpp

Go to the documentation of this file.

#ifndef GENESIS_UTILS_IO_INPUT_STREAM_H_
#define GENESIS_UTILS_IO_INPUT_STREAM_H_

/*
    Genesis - A toolkit for working with phylogenetic data.
    Copyright (C) 2014-2017 Lucas Czech

    This program is free software: you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
    the Free Software Foundation, either version 3 of the License, or
    (at your option) any later version.

    This program is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU General Public License for more details.

    You should have received a copy of the GNU General Public License
    along with this program.  If not, see <http://www.gnu.org/licenses/>.

    Contact:
    Lucas Czech <lucas.czech@h-its.org>
    Exelixis Lab, Heidelberg Institute for Theoretical Studies
    Schloss-Wolfsbrunnenweg 35, D-69118 Heidelberg, Germany
*/

#include "genesis/utils/io/input_source.hpp"
#include "genesis/utils/io/input_reader.hpp"

#include <assert.h>
#include <stdexcept>
#include <string>
#include <utility>

namespace genesis {
namespace utils {

// =================================================================================================
//     Input Stream
// =================================================================================================

class InputStream
{
public:

    // -------------------------------------------------------------
    //     Member Types
    // -------------------------------------------------------------

    static const size_t BlockLength = 1 << 24;

    using self_type         = InputStream;
    using value_type        = char;

    // -------------------------------------------------------------
    //     Constructors and Rule of Five
    // -------------------------------------------------------------

    InputStream()
        : source_name_( "invalid source" )
        , buffer_(   nullptr )
        , data_pos_( 0 )
        , data_end_( 0 )
        , current_( '\0' )
        , line_(     0 )
        , column_(   0 )
    {}

    explicit InputStream( std::unique_ptr<BaseInputSource> input_source )
        : line_(   1 )
        , column_( 1 )
    {
        init_( std::move( input_source ));
    }

    ~InputStream()
    {
        delete[] buffer_;
        buffer_ = nullptr;
    }

    InputStream(self_type const&) = delete;
    InputStream(self_type&&)      = delete;

    self_type& operator= (self_type const&) = delete;
    self_type& operator= (self_type&&)      = delete;

    // -------------------------------------------------------------
    //     Char Operations
    // -------------------------------------------------------------

    char operator * () const
    {
        return current_;
    }

    char current() const
    {
        if( data_pos_ >= data_end_ ) {
            throw std::runtime_error(
                "Unexpected end of " + source_name() + " at " + at() + "."
            );
        }
        if( current_ < 0 ) {
            throw std::domain_error(
                "Invalid input char in " + source_name() + " at " + at() + "."
            );
        }
        return current_;
    }

    self_type& advance()
    {
        // If we were already at the end, set counter so zero.
        if( data_pos_ >= data_end_ ) {
            reset_();
            return *this;
        }

        // Read data if necessary.
        update_blocks_();

        // In case we are moving to a new line, set the counters accordingly.
        if( current_ == '\n' ) {
            ++line_;
            column_ = 1;
        } else {
            ++column_;
        }

        // Next position.
        ++data_pos_;

        // Set the char.
        set_current_char_();
        return *this;
    }

    self_type& operator ++ ()
    {
        advance();
        return *this;
    }

    char get_char()
    {
        char ret = current_;
        advance();
        return ret;
    }

    // -------------------------------------------------------------
    //     Line Operations
    // -------------------------------------------------------------

    std::pair< char*, size_t > get_line()
    {
        if( data_pos_ >= data_end_ ) {
            return { nullptr, 0 };
        }

        // Read data if necessary.
        update_blocks_();

        // Find the end of the line.
        size_t line_end = data_pos_;
        while( line_end != data_end_       &&
               buffer_[ line_end ] != '\n' &&
               buffer_[ line_end ] != '\r'
        ) {
            ++line_end;
        }

        // If the line is too long, throw.
        if( line_end - data_pos_ + 1 > BlockLength ) {
            throw std::runtime_error( "Input line too long in " + source_name() + " at " + at() );
        }

        // Set the end of the line to \0, so that downstream parses can work with it.
        if( buffer_[ line_end ] == '\n' ) {
            buffer_[ line_end ] = '\0';

        } else if( buffer_[ line_end ] == '\r' ) {
            buffer_[ line_end ] = '\0';

            // Treat stupid Windows \r\n lines breaks.
            if( line_end + 1 < data_end_ && buffer_[ line_end + 1 ] == '\n' ) {
                ++line_end;
                buffer_[ line_end ] = '\0';
            }
        } else {

            // Files might be missing the line break at the end of the last line. Add it.
            assert( line_end == data_end_ );
            ++data_end_;
            buffer_[ line_end ] = '\0';
        }

        // Get pointer to beginning of the line, and length of the line, for returning it.
        char*  ret_ptr = buffer_ + data_pos_;
        size_t ret_len = line_end - data_pos_;

        // Move to the first char of the next line, so that future calls for reading a line or
        // char start at the right position.
        data_pos_ = line_end + 1;
        set_current_char_();

        // Set counters.
        ++line_;
        column_ = 1;

        return { ret_ptr, ret_len };
    }

    // -------------------------------------------------------------
    //     State
    // -------------------------------------------------------------

    size_t line() const
    {
        return line_;
    }

    size_t column() const
    {
        return column_;
    }

    std::string at() const
    {
        return std::to_string( line_ ) + ":" + std::to_string( column_ );
    }

    bool good() const
    {
        return data_pos_ < data_end_;
    }

    explicit operator bool() const
    {
        return data_pos_ < data_end_;
    }

    bool eof() const
    {
        return data_pos_ >= data_end_;
    }

    std::string source_name() const
    {
        return source_name_;
    }

    // -------------------------------------------------------------
    //     Internal Members
    // -------------------------------------------------------------

private:

    void reset_()
    {
        line_    = 0;
        column_  = 0;
        current_ = '\0';
    }

    void update_blocks_()
    {
        // This function is only called from the read char and line methods, which both beforehand
        // check the following condition. So, if it breaks, this function is invalidly called
        // from somewhere else.
        assert( data_pos_ <  data_end_ );

        // If this assertion breaks, someone tempered with our internal invariants.
        assert( data_end_ <= BlockLength * 2 );

        // If we are past the first block, we need to load more data into the blocks.
        if( data_pos_ >= BlockLength ) {

            // Move the second to the first block.
            std::memcpy( buffer_, buffer_ + BlockLength, BlockLength );
            data_pos_ -= BlockLength;
            data_end_ -= BlockLength;

            // If we are not yet at the end of the data, start the reader again:
            // Copy the third block to the second, and read into the third one.
            if( input_reader_.valid() ) {
                data_end_ += input_reader_.finish_reading();
                std::memcpy( buffer_ + BlockLength, buffer_ + 2 * BlockLength, BlockLength );
                input_reader_.start_reading( buffer_ + 2 * BlockLength, BlockLength );
            }
        }
    }

    void set_current_char_()
    {
        if( data_pos_ >= data_end_ ) {
            // If we just reached the end, do not fully reset the line and column counters.
            // They might be needed in some parser.
            current_ = '\0';
            return;
        }

        // Treat stupid Windows and Mac lines breaks. Set them to \n, so that downstream parsers
        // don't have to deal with this.
        if( buffer_[ data_pos_ ] == '\r' ) {
            buffer_[ data_pos_ ] = '\n';

            // If this is a Win line break \r\n, skip one of them, so that only a single \n
            // is visible to the outside.
            if( data_pos_ + 1 < data_end_ && buffer_[ data_pos_ + 1 ] == '\n' ) {
                ++data_pos_;
            }
        }

        // If this is the last char of the data, but there is no closing \n, add one.
        if( data_pos_ + 1 == data_end_ && buffer_[ data_pos_ ] != '\n' ) {
            ++data_end_;
            buffer_[ data_pos_ + 1 ] = '\n';
        }

        // Set the char.
        current_ = buffer_[ data_pos_ ];
    }

    void init_( std::unique_ptr<BaseInputSource> input_source )
    {
        // Set to empty defaults if there is no input.
        if( input_source == nullptr ) {
            source_name_ = "invalid source";

            buffer_   = nullptr;
            data_pos_ = 0;
            data_end_ = 0;

            current_ = '\0';
            line_    = 0;
            column_  = 0;
            return;
        }

        // We use three buffer blocks: one and two for the current line. The max line length is
        // one buffer length, so the beginning of the line is always in the first block, while its
        // end can reach into the second block, but never exeed it.
        // The third block is for the async reading.
        buffer_ = new char[ 3 * BlockLength ];

        try {
            // Set source name.
            source_name_ = input_source->source_name();

            // Read up to two blocks.
            data_pos_ = 0;
            data_end_ = input_source->read( buffer_, 2 * BlockLength );

            // Skip UTF-8 BOM, if found.
            if( data_end_  >= 3      &&
                buffer_[0] == '\xEF' &&
                buffer_[1] == '\xBB' &&
                buffer_[2] == '\xBF'
            ) {
                data_pos_ = 3;
            }

            // If there was no data, set to "empty" values.
            if( data_pos_ == data_end_ ) {
                reset_();

            // If there is data, set char value.
            } else {
                set_current_char_();
            }

            // If there is more data after the two blocks that we just read, start the
            // reading process (possibly async, if pthreads is available).
            if( data_end_ == 2 * BlockLength ) {
                input_reader_.init( std::move( input_source ));
                input_reader_.start_reading( buffer_ + 2 * BlockLength, BlockLength );
            }

        } catch( ... ) {
            delete[] buffer_;
            throw;
        }
    }

    // -------------------------------------------------------------
    //     Data Members
    // -------------------------------------------------------------

private:

    // Input data comes from here...
    InputReader input_reader_;
    std::string source_name_;

    // ...and is buffered here.
    char*  buffer_;
    size_t data_pos_;
    size_t data_end_;

    // Also, keep track of the current char and position in the data.
    char   current_;
    size_t line_;
    size_t column_;
};

} // namespace utils
} // namespace genesis

#endif // include guard