generic_input_stream.hpp

Go to the documentation of this file.

#ifndef GENESIS_UTILS_CONTAINERS_GENERIC_INPUT_STREAM_H_
#define GENESIS_UTILS_CONTAINERS_GENERIC_INPUT_STREAM_H_
 
/*
    Genesis - A toolkit for working with phylogenetic data.
    Copyright (C) 2014-2024 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@sund.ku.dk>
    University of Copenhagen, Globe Institute, Section for GeoGenetics
    Oster Voldgade 5-7, 1350 Copenhagen K, Denmark
*/
 
#include "genesis/utils/core/options.hpp"
#include "genesis/utils/threading/thread_pool.hpp"
 
#include <cassert>
#include <functional>
#include <future>
#include <memory>
#include <stdexcept>
#include <string>
#include <utility>
#include <vector>
 
namespace genesis {
namespace utils {
 
// =================================================================================================
//      Helpers
// =================================================================================================
 
struct EmptyGenericInputStreamData
{};
 
// =================================================================================================
//      Generic Input Stream
// =================================================================================================
 
template<class T, class D = EmptyGenericInputStreamData>
class GenericInputStream
{
public:
 
    // -------------------------------------------------------------------------
    //     Member Types
    // -------------------------------------------------------------------------
 
    using self_type         = GenericInputStream;
    using value_type        = T;
    using pointer           = value_type const*;
    using reference         = value_type const&;
    using difference_type   = std::ptrdiff_t;
    using iterator_category = std::input_iterator_tag;
 
    using Data              = D;
 
    static size_t const DEFAULT_BLOCK_SIZE = 0;
 
    // ======================================================================================
    //      Internal Iterator
    // ======================================================================================
 
    class Iterator
    {
    public:
 
        // -------------------------------------------------------------------------
        //     Constructors and Rule of Five
        // -------------------------------------------------------------------------
 
        using self_type         = GenericInputStream<T, D>::Iterator;
        using value_type        = T;
        using pointer           = value_type const*;
        using reference         = value_type const&;
        using difference_type   = std::ptrdiff_t;
        using iterator_category = std::input_iterator_tag;
 
        using Data              = D;
 
        Iterator() = default;
 
    private:
 
        Iterator(
            GenericInputStream const* generator
        )
            : generator_(     generator )
            , current_block_( std::make_shared<std::vector<T>>() )
            , buffer_block_(  std::make_shared<std::vector<T>>() )
            , future_(        std::make_shared<ProactiveFuture<size_t>>() )
        {
            // We use the generator as a check if this Iterator is intended to be a begin()
            // or end() iterator. If its the former, init and get the first element block.
            // We could also just use the default constructor to create end() iterators,
            // this would have the same effect.
            // After we are done iterating the input, we then set the generator_ to nullptr,
            // as a sign that we are done. This allows us also to know if we reached end() without
            // having to store the end() iterator when using this class.
            if( generator_ ) {
                if( ! generator_->get_element_ ) {
                    throw std::invalid_argument(
                        "Cannot use GenericInputStream without a function to get elements."
                    );
                }
 
                // Initialize the current_block_ and buffer_block_,
                // and read the first block(s) of the file.
                begin_iteration_();
            }
        }
 
    public:
 
        ~Iterator() = default;
 
        Iterator( self_type const& ) = default;
        Iterator( self_type&& )      = default;
 
        Iterator& operator= ( self_type const& ) = default;
        Iterator& operator= ( self_type&& )      = default;
 
        friend GenericInputStream;
 
        // -------------------------------------------------------------------------
        //     Accessors
        // -------------------------------------------------------------------------
 
        value_type const * operator->() const
        {
            assert( current_block_ );
            assert( current_pos_ < end_pos_ );
            assert( current_pos_ < current_block_->size() );
            return &((*current_block_)[current_pos_]);
        }
 
        value_type * operator->()
        {
            assert( current_block_ );
            assert( current_pos_ < end_pos_ );
            assert( current_pos_ < current_block_->size() );
            return &((*current_block_)[current_pos_]);
        }
 
        value_type const & operator*() const
        {
            assert( current_block_ );
            assert( current_pos_ < end_pos_ );
            assert( current_pos_ < current_block_->size() );
            return (*current_block_)[current_pos_];
        }
 
        value_type & operator*()
        {
            assert( current_block_ );
            assert( current_pos_ < end_pos_ );
            assert( current_pos_ < current_block_->size() );
            return (*current_block_)[current_pos_];
        }
 
        Data const& data() const
        {
            if( ! generator_ ) {
                throw std::runtime_error(
                    "Cannot access default constructed or past-the-end GenericInputStream content."
                );
            }
            assert( generator_ );
            return generator_->data_;
        }
 
        // We do not offer non-const access at the moment. If we did,
        // the generator pointer stored in this iterator would need to be non-const as well.
 
        // /* *
        //  * @brief Access the data stored in the iterator.
        //  */
        // Data& data()
        // {
        //     if( ! generator_ ) {
        //         throw std::runtime_error(
        //             "Cannot access default constructed or past-the-end GenericInputStream content."
        //         );
        //     }
        //     assert( generator_ );
        //     return generator_->data_;
        // }
 
        // -------------------------------------------------------------------------
        //     Iteration
        // -------------------------------------------------------------------------
 
        self_type& operator ++()
        {
            increment_();
            return *this;
        }
 
        // self_type operator ++(int)
        // {
        //     auto cpy = *this;
        //     increment_();
        //     return cpy;
        // }
 
        operator bool() const
        {
            // We here rely on the fact that we set the generator_ to nullptr if this is either
            // the end() iterator or if the iteration has reached its end after the data is done.
            return generator_ != nullptr;
        }
 
        bool operator==( self_type const& other ) const
        {
            // We compare the generators as a baseline - two past-the-end iterator shall
            // always compare equal. If only one of them is past-the-end, they will compare false.
            // Only if both are valid (not past-the-end) iterators, we compare their current
            // position in the block - two iterators to the same position in the same block
            // (using pointer comparison for the block - not actual data comparison) are equal.
            if( ! generator_ || ! other.generator_ ) {
                // generator_ is used as the indicator whether this is a past-the-end iterator
                // (in which case it is a nullptr), or not.
                return generator_ == other.generator_;
            }
            assert( generator_ && other.generator_ );
            return current_block_ == other.current_block_ && current_pos_ == other.current_pos_;
        }
 
        bool operator!=( self_type const& other ) const
        {
            return !(*this == other);
        }
 
        // -------------------------------------------------------------------------
        //     Internal Members
        // -------------------------------------------------------------------------
 
    private:
 
        void begin_iteration_()
        {
            // Check that they are set up from the constructor.
            assert( generator_ );
            assert( current_block_ );
            assert( buffer_block_ );
 
            // Before starting to init anything, call the callbacks.
            execute_begin_callbacks_();
 
            // Run the correct init function depending on buffering.
            if( generator_->block_size_ == 0 ) {
                begin_iteration_without_buffer_();
            } else {
                begin_iteration_with_buffer_();
            }
        }
 
        void begin_iteration_without_buffer_()
        {
            // Edge case: no buffering.
            // Block size zero indicates to use no buffering, so we just use a single element.
            // Make space for it, read the first element, and then we are done here.
            assert( generator_ );
            assert( generator_->block_size_ == 0 );
            assert( current_pos_ == 0 );
            current_block_->resize( 1 );
            end_pos_ = 1;
            increment_without_buffer_();
        }
 
        void begin_iteration_with_buffer_()
        {
            // The main class makes sure that we have a thread pool when we want to use the buffer
            assert( generator_->thread_pool_ );
 
            // Init the records and create empty VcfRecord to read into.
            // The blocks have been initialized in the contructor already; assert this.
            current_block_->resize( generator_->block_size_ );
            buffer_block_->resize(  generator_->block_size_ );
            assert( current_block_->size() == generator_->block_size_ );
            assert( buffer_block_->size()  == generator_->block_size_ );
 
            // Read the first block synchronously,
            // so that there is data initialized to be dereferenced.
            end_pos_ = read_block_( generator_, current_block_, generator_->block_size_ );
            assert( current_pos_ == 0 );
 
            // Check how much data we got, and whether we want to start the background worker.
            // If there is less data than the block size, the file is already done.
            // However, if the first block was fully read, we start the async worker thread
            // to fill the buffer with the next block of data.
            if( end_pos_ == generator_->block_size_ ) {
                enqueue_filling_of_buffer_block_();
            } else if( end_pos_ == 0 ) {
                // Edge case: zero elements read. We are already done then.
                end_iteration_();
                return;
            }
            assert( end_pos_ > 0 );
            assert( end_pos_ <= generator_->block_size_ );
 
            // Now we have an element, which will be the first one of the iteration,
            // and so we execute the observers for it.
            assert( current_pos_ == 0 );
            execute_on_enter_observers_( (*current_block_)[current_pos_] );
        }
 
        void increment_()
        {
            // Make sure that all things are still in place.
            assert( generator_ );
            assert( current_block_ && current_block_->size() > 0 );
 
            // Make sure that sizes have not been changed in the parent class.
            assert(
                current_block_->size() == generator_->block_size_ ||
                (
                    generator_->block_size_ == 0 &&
                    current_block_->size()  == 1
                )
            );
            assert( buffer_block_->size() == generator_->block_size_ );
 
            // We are moving to the next element. This is the only time that we need to call
            // the leaving observers, as this happens both during the normal iteration,
            // as well as at its end. We first call this function, and then do the actual increment.
            // What that incrementation finds that we are at the end of the data, we have already
            // called our leaving observer, so all good, no extra case needed for that.
            assert( current_pos_ < current_block_->size() );
            execute_on_leave_observers_( (*current_block_)[current_pos_] );
 
            // Run the actual increment implementation.
            if( generator_->block_size_ == 0 ) {
                increment_without_buffer_();
            } else {
                increment_with_buffer_();
            }
        }
 
        void increment_without_buffer_()
        {
            // Edge case: no buffering.
            // Read the next element. If there is none, we are done.
            assert( current_pos_ == 0 );
            assert( current_block_ );
            assert( current_block_->size() == 1 );
            if( get_next_element_( generator_, (*current_block_)[0] )) {
                execute_on_enter_observers_( (*current_block_)[0] );
            } else {
                end_iteration_();
            }
        }
 
        void increment_with_buffer_()
        {
            // Finish the reading (potentially waiting if not yet finished in the worker thread).
            // The future returns how much data there was to be read, which we use as our status.
            // After that, swap the buffer and start a new reading operation in the worker thread.
 
            // Move to the next element in the vector.
            ++current_pos_;
            assert( current_pos_ <= end_pos_ );
 
            // If we are at the end of the data, we are either done iterating,
            // or reached the end of the current buffer block.
            if( current_pos_ == end_pos_ ) {
 
                // If we did not get a full block size when reading, we are done iterating.
                // Indicate this by unsetting the generator_ pointer.
                if( end_pos_ < generator_->block_size_ ) {
                    end_iteration_();
                    return;
                }
 
                // If we are at the end of the record vector, and if that vector was full
                // (complete block size), there is more data, so start reading.
 
                assert( end_pos_ == generator_->block_size_ );
                assert( future_ );
                assert( future_->valid() );
 
                // Get how many records were read into the buffer, which also waits for the reading
                // if necessary. If we did not read any now, that means that the number of total
                // elements is a multiple of the block size, which we handle as a special case here.
                // There is probably some better way to restructure the code to avoid this edge
                // case... but good enough for now.
                end_pos_ = future_->get();
                if( end_pos_ == 0 ) {
                    end_iteration_();
                    return;
                }
 
                // Here, we know that there is more data buffered, so we can swap the buffer.
                // If that last read returned with less than a full block of items, we have reached
                // the end of the input, and do not want to start reading more (that condition was
                // missing before - nasty bug that only showed up in consequetive concurrent tests,
                // so that a thread was started even after there was no more input. That thread
                // would then access data from the next test case, which of course happened to
                // occupy the same stack space. This never showed up in read data, as that
                // stray thread would not communicate with anything, and somehow did not throw
                // or segfault either. Super nasty to find.
                // Anyway, if we had read a full block before, we need to start reading again.
                // Finally, set or internal current location to the first element of the vector again.
                assert( current_block_ );
                assert( buffer_block_ );
                assert( end_pos_ > 0 && end_pos_ <= generator_->block_size_ );
                std::swap( buffer_block_, current_block_ );
                if( end_pos_ == generator_->block_size_ ) {
                    assert( !future_->valid() );
                    enqueue_filling_of_buffer_block_();
                    assert( future_->valid() );
                }
                current_pos_ = 0;
            }
 
            // Now we have moved to the next element, and potentially the next block,
            // so we are ready to call the observers for that element.
            assert( current_pos_ < end_pos_ );
            execute_on_enter_observers_( (*current_block_)[current_pos_] );
        }
 
        void enqueue_filling_of_buffer_block_()
        {
            // Those shared pointers have been initialized in the constructor; let's assert this.
            assert( generator_ );
            assert( generator_->thread_pool_ );
            assert( future_ );
 
            // This function is only every called after we finished any previous operations,
            // so let's assert that the thread pool and future are in the states that we expect.
            // assert( generator_->thread_pool_->load() == 0 );
            assert( ! future_->valid() );
 
            // Make sure that sizes have not been changed in the parent class.
            assert(
                current_block_->size() == generator_->block_size_ ||
                (
                    generator_->block_size_ == 0 &&
                    current_block_->size()  == 1
                )
            );
            assert( buffer_block_ );
            assert( buffer_block_->size() == generator_->block_size_ );
 
            // In order to use lambda captures by copy for class member variables in C++11, we first
            // have to make local copies, and then capture those. Capturing the class members
            // directly was only introduced later. Bit cumbersome, but gets the job done.
            auto generator    = generator_;
            auto buffer_block = buffer_block_;
            auto block_size   = generator_->block_size_;
 
            // The lambda returns the result of read_block_ call, that is, the number of records that
            // have been read, and which we later (in the future_) use to see how much data we got.
            // It is of course important that the input is read in the correct order of elements.
            // Despite its parallel nature, we can use a thread pool here, as we are only ever
            // submitting a single read task to it, so there cannot be two reads of the same lambda
            // iterator in the pool.
            *future_ = generator_->thread_pool_->enqueue_and_retrieve(
                [ generator, buffer_block, block_size ](){
                    return read_block_( generator, buffer_block, block_size );
                }
            );
        }
 
        static size_t read_block_(
            GenericInputStream const* generator,
            std::shared_ptr<std::vector<T>> buffer_block,
            size_t block_size
        ) {
            // This is a static function that does not depend on the class member data, so that
            // we can use it from the future lambda in the thread pool above without having to worry
            // about lambda captures of `this` going extinct... which was an absolutely nasty bug to
            // find! Such a rookie mistake! For that reason, we also take all arguments as shared
            // pointers, so that they are kept alive while the thread pool is working.
            // However, once its done with its work, the function (the one that we give to the thread
            // pool with a lambda) is popped from the thread queue, so that the shared pointer can
            // be freed again - that is, we do not need to worry about the lambda keeping the shared
            // pointer from freeing its memory indefinitely.
 
            assert( generator );
            assert( buffer_block );
            assert( buffer_block->size() == block_size );
 
            // Read as long as there is data. Return number of read records.
            size_t i = 0;
            for( ; i < block_size; ++i ) {
                if( ! get_next_element_( generator, (*buffer_block)[i] )) {
                    break;
                }
            }
            return i;
        }
 
        static bool get_next_element_(
            GenericInputStream const* generator,
            T& target
        ) {
            assert( generator );
            bool usable_element = false;
            while( true ) {
                // Get the next element from the input source.
                bool const got_element = generator->get_element_( target );
 
                if( got_element ) {
                    // If this is an element (not yet at the end of the data),
                    // apply all transforms and filters, and get out of here if all of them
                    // return `true`, that is, if none of them wants to filter out the element.
                    // If however one of them returns `false`, we need to find another element.
                    usable_element = true;
                    for( auto const& tra_fil : generator->transforms_and_filters_ ) {
                        usable_element = tra_fil( target );
                        if( ! usable_element ) {
                            // If one of the transforms/filters does not want us to continue,
                            // we do not call the others. Break out of the inner loop.
                            break;
                        }
                    }
 
                    // If we got out of the above loop without any failing filters,
                    // we have fonud an element, so let's move it to our target.
                    if( usable_element ) {
                        // We already checked that got_element holds data, so no need to repeat,
                        // just assert it. Done here, break out of the while loop.
                        assert( got_element );
                        break;
                    }
                    // Else (usable_element == false): Loop again, try the next element.
 
                } else {
                    // If this is not a valid element, we reached the end of the input.
                    // We note this in the return value, then break out of the while loop.
                    assert( ! got_element );
                    usable_element = false;
                    break;
                }
            }
            return usable_element;
        }
 
        void end_iteration_()
        {
            // This function is called at any point when we reach the end of the input source.
            // We first need to call all end callbacks, before unsetting the generator pointer.
            // Doing so is our signal that we have reached the end, and is used for example
            // in the equality comparison of this iterator to test whether an iterator is valid
            // or not (or equal to the past-the-end iterator).
            assert( generator_ );
            assert( ! future_->valid() );
            execute_end_callbacks_();
            generator_ = nullptr;
        }
 
        void execute_on_enter_observers_( T const& element ) const
        {
            assert( generator_ );
            for( auto const& observer : generator_->on_enter_observers_ ) {
                observer( element );
            }
        }
 
        void execute_on_leave_observers_( T const& element ) const
        {
            assert( generator_ );
            for( auto const& observer : generator_->on_leave_observers_ ) {
                observer( element );
            }
        }
 
        void execute_begin_callbacks_() const
        {
            assert( generator_ );
            for( auto const& cb : generator_->begin_callbacks_ ) {
                cb( *generator_ );
            }
        }
 
        void execute_end_callbacks_() const
        {
            assert( generator_ );
            for( auto const& cb : generator_->end_callbacks_ ) {
                cb( *generator_ );
            }
        }
 
    private:
 
        // Parent.
        GenericInputStream const* generator_ = nullptr;
 
        // Buffer buffering data, and positions in it.
        std::shared_ptr<std::vector<T>> current_block_;
        std::shared_ptr<std::vector<T>> buffer_block_;
        size_t current_pos_ = 0;
        size_t end_pos_ = 0;
 
        // Store the future_ used to keep track of the background task. It returns the number of
        // elements that have been read into the buffer (block_size_, or less at the end of the file).
        std::shared_ptr<ProactiveFuture<size_t>> future_;
 
    };
 
    // ======================================================================================
    //      Main Class
    // ======================================================================================
 
    // -------------------------------------------------------------------------
    //     Constructors and Rule of Five
    // -------------------------------------------------------------------------
 
    GenericInputStream() = default;
 
    GenericInputStream(
        std::function<bool(value_type&)> get_element,
        std::shared_ptr<utils::ThreadPool> thread_pool = nullptr,
        size_t block_size = DEFAULT_BLOCK_SIZE
    )
        : get_element_(get_element)
        , block_size_( block_size )
    {
        // We only need to set the thread pool if we are going to use buffering.
        if( block_size > 0 ) {
            thread_pool_ = thread_pool ? thread_pool : Options::get().global_thread_pool();
        }
    }
 
    GenericInputStream(
        std::function<bool(value_type&)> get_element,
        Data const& data,
        std::shared_ptr<utils::ThreadPool> thread_pool = nullptr,
        size_t block_size = DEFAULT_BLOCK_SIZE
    )
        : GenericInputStream( get_element, thread_pool, block_size )
    {
        data_ = data;
    }
 
    GenericInputStream(
        std::function<bool(value_type&)> get_element,
        Data&& data,
        std::shared_ptr<utils::ThreadPool> thread_pool = nullptr,
        size_t block_size = DEFAULT_BLOCK_SIZE
    )
        : GenericInputStream( get_element, thread_pool, block_size )
    {
        data_ = std::move( data );
    }
 
    ~GenericInputStream() = default;
 
    GenericInputStream( self_type const& ) = default;
    GenericInputStream( self_type&& )      = default;
 
    self_type& operator= ( self_type const& ) = default;
    self_type& operator= ( self_type&& )      = default;
 
    friend Iterator;
 
    // -------------------------------------------------------------------------
    //     Iteration
    // -------------------------------------------------------------------------
 
    Iterator begin()
    {
        if( has_started_ ) {
            throw std::runtime_error( "GenericInputStream: Cannot call begin() multiple times." );
        }
        has_started_ = true;
        return Iterator( this );
    }
 
    Iterator end()
    {
        return Iterator( nullptr );
    }
 
    operator bool() const
    {
        return static_cast<bool>( get_element_ );
    }
 
    Data const& data() const
    {
        return data_;
    }
 
    Data& data()
    {
        return data_;
    }
 
    // -------------------------------------------------------------------------
    //     Filters and Transformations
    // -------------------------------------------------------------------------
 
    self_type& add_transform( std::function<void(T&)> const& transform )
    {
        return add_transform_filter(
            [transform]( T& element ){
                transform( element );
                return true;
            }
        );
    }
 
    self_type& add_filter( std::function<bool(T const&)> const& filter )
    {
        return add_transform_filter(
            [filter]( T& element ){
                return filter( element );
            }
        );
    }
 
    self_type& add_transform_filter( std::function<bool(T&)> const& filter )
    {
        // Sanity check.
        if( has_started_ ) {
            throw std::runtime_error(
                "GenericInputStream: Cannot change filters/transformations after iteration has started."
            );
        }
        transforms_and_filters_.push_back( filter );
        return *this;
    }
 
    self_type& clear_filters_and_transformations()
    {
        if( has_started_ ) {
            throw std::runtime_error(
                "GenericInputStream: Cannot change filters/transformations after iteration has started."
            );
        }
        transforms_and_filters_.clear();
        return *this;
    }
 
    // -------------------------------------------------------------------------
    //     Observers and Callbacks
    // -------------------------------------------------------------------------
 
    self_type& add_on_enter_observer( std::function<void(T const&)> const& observer )
    {
        if( has_started_ ) {
            throw std::runtime_error(
                "GenericInputStream: Cannot change observers after iteration has started."
            );
        }
        on_enter_observers_.push_back( observer );
        return *this;
    }
 
    self_type& add_on_leave_observer( std::function<void(T const&)> const& observer )
    {
        if( has_started_ ) {
            throw std::runtime_error(
                "GenericInputStream: Cannot change observers after iteration has started."
            );
        }
        on_leave_observers_.push_back( observer );
        return *this;
    }
 
    self_type& clear_observers()
    {
        if( has_started_ ) {
            throw std::runtime_error(
                "GenericInputStream: Cannot change observers after iteration has started."
            );
        }
        on_enter_observers_.clear();
        on_leave_observers_.clear();
        return *this;
    }
 
    self_type& add_begin_callback( std::function<void(GenericInputStream const&)> const& callback )
    {
        if( has_started_ ) {
            throw std::runtime_error(
                "GenericInputStream: Cannot change callbacks after iteration has started."
            );
        }
        begin_callbacks_.push_back( callback );
        return *this;
    }
 
    self_type& add_end_callback( std::function<void(GenericInputStream const&)> const& callback )
    {
        if( has_started_ ) {
            throw std::runtime_error(
                "GenericInputStream: Cannot change callbacks after iteration has started."
            );
        }
        end_callbacks_.push_back( callback );
        return *this;
    }
 
    self_type& clear_callbacks()
    {
        if( has_started_ ) {
            throw std::runtime_error(
                "GenericInputStream: Cannot change callbacks after iteration has started."
            );
        }
        begin_callbacks_.clear();
        end_callbacks_.clear();
        return *this;
    }
 
    // -------------------------------------------------------------------------
    //     Thread Settings
    // -------------------------------------------------------------------------
 
    std::shared_ptr<utils::ThreadPool> thread_pool() const
    {
        return thread_pool_;
    }
 
    self_type& thread_pool( std::shared_ptr<utils::ThreadPool> value )
    {
        if( has_started_ ) {
            throw std::runtime_error(
                "GenericInputStream: Cannot change thread pool after iteration has started."
            );
        }
        if( !value ) {
            throw std::runtime_error(
                "GenericInputStream: Cannot change thread pool to empty object."
            );
        }
        thread_pool_ = value;
        return *this;
    }
 
    size_t block_size() const
    {
        return block_size_;
    }
 
    self_type& block_size( size_t value )
    {
        if( has_started_ ) {
            throw std::runtime_error(
                "GenericInputStream: Cannot change thread pool block size after iteration has started."
            );
        }
        block_size_ = value;
        return *this;
    }
 
    // -------------------------------------------------------------------------
    //     Data Members
    // -------------------------------------------------------------------------
 
private:
 
    // We have two different types of functions that we accept to operate on the data:
    // transforms and filters, which are both executed when filling the buffers.
    // To keep it simple and a bit faster, we just store them as the same type here,
    // which can do both of their functionality at once.
    std::vector<std::function<bool(T&)>> transforms_and_filters_;
 
    // We also offer observers that are executed once the iteration
    // reaches or leaves the respective element.
    std::vector<std::function<void(T const&)>> on_enter_observers_;
    std::vector<std::function<void(T const&)>> on_leave_observers_;
 
    // We furthermore allow callbacks for the beginning and and of the iteration.
    std::vector<std::function<void(self_type const&)>> begin_callbacks_;
    std::vector<std::function<void(self_type const&)>> end_callbacks_;
 
    // Underlying iterator and associated data.
    std::function<bool( value_type& )> get_element_;
    Data data_;
 
    // Thread pool to run the buffering in the background.
    std::shared_ptr<utils::ThreadPool> thread_pool_;
 
    // Block buffering settings.
    size_t block_size_ = DEFAULT_BLOCK_SIZE;
 
    // Flag to avoid accidentally calling begin() twice, or calling threading settings
    // after begin() has already been called.
    bool has_started_ = false;
 
};
 
} // namespace utils
} // namespace genesis
 
#endif // include guard