lambda_iterator.hpp

Go to the documentation of this file.

#ifndef GENESIS_UTILS_CONTAINERS_LAMBDA_ITERATOR_H_
#define GENESIS_UTILS_CONTAINERS_LAMBDA_ITERATOR_H_
 
/*
    Genesis - A toolkit for working with phylogenetic data.
    Copyright (C) 2014-2022 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 <lczech@carnegiescience.edu>
    Department of Plant Biology, Carnegie Institution For Science
    260 Panama Street, Stanford, CA 94305, USA
*/
 
#include "genesis/utils/core/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 EmptyLambdaIteratorData
{};
 
// =================================================================================================
//      Lambda Iterator
// =================================================================================================
 
template<class T, class D = EmptyLambdaIteratorData>
class LambdaIterator
{
public:
 
    // -------------------------------------------------------------------------
    //     Member Types
    // -------------------------------------------------------------------------
 
    using self_type         = LambdaIterator;
    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         = LambdaIterator<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(
            LambdaIterator const* generator
        )
            : generator_(     generator )
            , current_block_( std::make_shared<std::vector<T>>() )
            , buffer_block_(  std::make_shared<std::vector<T>>() )
            , thread_pool_(   std::make_shared<utils::ThreadPool>( 1 ))
            , future_(        std::make_shared<std::future<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 LambdaIterator without a function to get elements."
                    );
                }
 
                // Initialize the current_block_ and buffer_block_,
                // and read the first block(s) of the file.
                init_();
            }
        }
 
    public:
 
        ~Iterator() = default;
 
        Iterator( self_type const& ) = default;
        Iterator( self_type&& )      = default;
 
        Iterator& operator= ( self_type const& ) = default;
        Iterator& operator= ( self_type&& )      = default;
 
        friend LambdaIterator;
 
        // -------------------------------------------------------------------------
        //     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 LambdaIterator 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 LambdaIterator 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 init_()
        {
            // Check that they are set up from the constructor.
            assert( generator_ );
            assert( current_block_ );
            assert( buffer_block_ );
 
            // Edge case: no buffering.
            // Block size zero indicates to use no buffering, so we just use a single element.
            if( generator_->block_size_ == 0 ) {
                // Make space for it, read the first element, and then we are done here.
                current_block_->resize( 1 );
                end_pos_ = 1;
                increment_();
                return;
            }
 
            // 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 );
 
            // If there is less data than the block size, the file is already done.
            // No need to start the async buffering, we can just get out of here.
            if( end_pos_ < generator_->block_size_ ) {
 
                // Edge case: zero elements read. We are already done then.
                if( end_pos_ == 0 ) {
                    generator_ = nullptr;
                }
                return;
            }
 
            // If we are here, the first block was fully read,
            // so we start the async worker thread to fill the buffer.
            assert( end_pos_ == generator_->block_size_ );
            fill_buffer_block_();
        }
 
        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_ );
 
            // Edge case: no buffering.
            // Read the next element. If there is none, we are done.
            if( generator_->block_size_ == 0 ) {
                assert( current_block_->size() == 1 );
                if( ! get_next_element_( generator_, (*current_block_)[0] )) {
                    generator_ = nullptr;
                }
                return;
            }
 
            // 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_ ) {
                    generator_ = nullptr;
                    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 ) {
                    generator_ = nullptr;
                    return;
                }
 
                // Here, we know that there is more data, so we can swap the buffer,
                // start reading again, and set or internal current location
                // to the first element of the vector again.
                assert( end_pos_ > 0 && end_pos_ <= generator_->block_size_ );
                std::swap( buffer_block_, current_block_ );
                fill_buffer_block_();
                current_pos_ = 0;
            }
        }
 
        void fill_buffer_block_()
        {
            // Those shared pointers have been initialized in the constructor; let's assert this.
            assert( generator_ );
            assert( 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( 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_->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 direclty
            // 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.
            *future_ = thread_pool_->enqueue(
                [ generator, buffer_block, block_size ](){
                    return read_block_( generator, buffer_block, block_size );
                }
            );
        }
 
        static size_t read_block_(
            LambdaIterator 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->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_(
            LambdaIterator 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;
        }
 
    private:
 
        // Parent.
        LambdaIterator 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;
 
        // Thread pool to run the buffering in the background.
        // Also, 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<utils::ThreadPool> thread_pool_;
        std::shared_ptr<std::future<size_t>> future_;
 
    };
 
    // ======================================================================================
    //      Main Class
    // ======================================================================================
 
    // -------------------------------------------------------------------------
    //     Constructors and Rule of Five
    // -------------------------------------------------------------------------
 
    LambdaIterator() = default;
 
    LambdaIterator(
        std::function<bool(value_type&)> get_element,
        size_t block_size = DEFAULT_BLOCK_SIZE
    )
        : get_element_(get_element)
        , block_size_( block_size )
    {}
 
    LambdaIterator(
        std::function<bool(value_type&)> get_element,
        Data const& data,
        size_t block_size = DEFAULT_BLOCK_SIZE
    )
        : get_element_(get_element)
        , data_(data)
        , block_size_( block_size )
    {}
 
    LambdaIterator(
        std::function<bool(value_type&)> get_element,
        Data&& data,
        size_t block_size = DEFAULT_BLOCK_SIZE
    )
        : get_element_(get_element)
        , data_( std::move( data ))
        , block_size_( block_size )
    {}
 
    ~LambdaIterator() = default;
 
    LambdaIterator( self_type const& ) = default;
    LambdaIterator( self_type&& )      = default;
 
    self_type& operator= ( self_type const& ) = default;
    self_type& operator= ( self_type&& )      = default;
 
    friend Iterator;
 
    // -------------------------------------------------------------------------
    //     Iteration
    // -------------------------------------------------------------------------
 
    Iterator begin()
    {
        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&)> transform )
    {
        transforms_and_filters_.push_back(
            [transform]( T& element ){
                transform( element );
                return true;
            }
        );
        return *this;
    }
 
    self_type& add_filter( std::function<bool(T const&)> filter )
    {
        transforms_and_filters_.push_back(
            [filter]( T& element ){
                return filter( element );
            }
        );
        return *this;
    }
 
    self_type& add_transform_filter( std::function<bool(T&)> filter )
    {
        transforms_and_filters_.push_back(
            [filter]( T& element ){
                return filter( element );
            }
        );
        return *this;
    }
 
    self_type& clear_filters_and_transformations()
    {
        transforms_and_filters_.clear();
    }
 
    // -------------------------------------------------------------------------
    //     Other Settings
    // -------------------------------------------------------------------------
 
    size_t block_size() const
    {
        return block_size_;
    }
 
    self_type& block_size( size_t value )
    {
        block_size_ = value;
        return *this;
    }
 
    // -------------------------------------------------------------------------
    //     Data Members
    // -------------------------------------------------------------------------
 
private:
 
    // std::vector<std::function<void(T&)>> transforms_;
    // std::vector<std::function<bool(T const&)>> filters_;
    std::vector<std::function<bool(T&)>> transforms_and_filters_;
 
    // Underlying iterator and associated data.
    std::function<bool( value_type& )> get_element_;
    Data data_;
 
    // Block buffering settings.
    size_t block_size_ = DEFAULT_BLOCK_SIZE;
 
};
 
} // namespace utils
} // namespace genesis
 
#endif // include guard