nds_iterate_handlers.hh

Go to the documentation of this file.

//
// Created by jonathan.hanks on 5/5/17.
//

#ifndef NDS2_CLIENT_NDS_ITERATE_HANDLERS_HH
#define NDS2_CLIENT_NDS_ITERATE_HANDLERS_HH

#include <algorithm>
#include <limits>
#include <memory>

#include "nds_iterate_handler.hh"

namespace NDS
{
    namespace detail
    {
        class iterate_fast_handler : public detail::iterate_handler
        {
        private:
            buffer::gps_second_type cur_gps_;
            buffer::gps_second_type gps_start_;
            buffer::gps_second_type gps_stop_;
            buffer::gps_second_type stride_;
            bool                    online_;
            buffers_type            next_entry_;

            template < typename T >
            T
            safe_add( T val1, T val2 )
            {
                static const T max_val = std::numeric_limits< T >::max( );
                if ( val2 >= max_val - val1 )
                {
                    return max_val;
                }
                return val1 + val2;
            }

            void
            get_next_block( buffers_type& output )
            {
                auto parent = conn( );
                if ( !parent )
                {
                    throw std::runtime_error( "Connection object is null" );
                }
                parent->next_raw_buffer( output );
                if ( gps_start_ == 0 )
                {
                    gps_start_ = output.front( ).Start( );
                    gps_stop_ = safe_add< NDS::buffer::gps_second_type >(
                        gps_start_,
                        ( gps_stop_ == 0 ? 1893024016 : gps_stop_ ) );
                    parent->request_end_time( gps_stop_ );
                }
                if ( !output.empty( ) )
                {
                    cur_gps_ = output.front( ).Stop( );
                }
                if ( cur_gps_ >= gps_stop_ && online_ )
                {
                    parent->cycle_nds1_connection( );
                }
            }

        public:
            iterate_fast_handler(
                buffer::gps_second_type                       gps_start,
                buffer::gps_second_type                       gps_stop,
                buffer::gps_second_type                       stride,
                const connection::channel_names_type&         channel_names,
                std::shared_ptr< NDS::detail::conn_p_type >&& parent )
                : iterate_handler( std::move( parent ) ), cur_gps_( 0 ),
                  gps_start_( gps_start ), gps_stop_( gps_stop ),
                  stride_( stride ), online_( gps_start == 0 )
            {
                buffer::gps_second_type stop_time =
                    ( parent->protocol == NDS::connection::PROTOCOL_ONE &&
                              gps_start == 0
                          ? 0
                          : gps_stop_ );
                parent->issue_iterate(
                    gps_start_, stop_time, stride, channel_names );
            }
            ~iterate_fast_handler( ) override = default;

            bool
            has_next( ) override
            {
                if ( next_entry_.size( ) > 0 )
                {
                    return true;
                }
                if ( cur_gps_ >= gps_stop_ && gps_start_ != 0 )
                {
                    return false;
                }
                try
                {
                    get_next_block( next_entry_ );
                }
                catch ( ... )
                {
                    return false;
                }
                return true;
            }

            void
            next( buffers_type& output ) override
            {
                while ( has_next( ) )
                {
                    if ( next_entry_.size( ) > 0 )
                    {
                        output.swap( next_entry_ );
                        next_entry_.clear( );
                        return;
                    }
                    get_next_block( next_entry_ );
                }
            }
        };

        class iterate_handler_with_simple_gaps : public detail::iterate_handler
        {
            typedef std::vector< NDS::channel >::iterator ch_vec_iter;
            typedef std::vector< NDS::channel >::const_iterator
                ch_vec_const_iter;

            struct indexed_buffers
            {

                buffers_type            bufs_;
                buffer::gps_second_type cur_;
                buffer::gps_second_type stride_;
                std::vector<
                    std::unique_ptr< NDS::detail::delayed_gap_handler > >
                    delay_;

                indexed_buffers( )
                    : bufs_( ), cur_( 0 ), stride_( 0 ), delay_( )
                {
                }

                void
                initialize( const std::vector< NDS::channel >& channel_list,
                            buffer::gps_second_type            gps_time,
                            buffer::gps_second_type            stride )
                {
                    if ( bufs_.size( ) != channel_list.size( ) )
                    {
                        bufs_.resize( channel_list.size( ) );
                    }

                    for ( int i = 0; i < channel_list.size( ); ++i )
                    {
                        const NDS::channel& cur_ch = channel_list[ i ];
                        NDS::buffer&        cur_buf = bufs_[ i ];

                        cur_buf.reset_channel_info( cur_ch, gps_time, 0 );
                        if ( cur_ch.Type( ) &
                             NDS::channel::CHANNEL_TYPE_MTREND )
                        {
                            cur_buf.resize(
                                stride /
                                static_cast< buffer::size_type >( 60 ) );
                        }
                        else
                        {
                            cur_buf.resize( stride *
                                            static_cast< buffer::size_type >(
                                                cur_ch.SampleRate( ) ) );
                        }
                    }
                    cur_ = gps_time;
                    stride_ = stride;
                    delay_.clear( );
                }

                bool
                valid( ) const
                {
                    if ( bufs_.size( ) == 0 )
                    {
                        return false;
                    }
                    if ( cur_ >= bufs_.front( ).Start( ) + stride_ )
                    {
                        return false;
                    }
                    return true;
                }

                void
                reset( buffers_type& other )
                {
                    bufs_.swap( other );
                    cur_ = 0;
                    stride_ = 0;
                    if ( bufs_.size( ) > 0 )
                    {
                        cur_ = bufs_.front( ).Start( );
                        stride_ = bufs_.front( ).Stop( ) - cur_;
                    }
                    delay_.clear( );
                }

                buffers_type&
                bufs( )
                {
                    return bufs_;
                }

                buffer::gps_second_type
                cur( ) const
                {
                    return cur_;
                }

                buffer::gps_second_type
                start( ) const
                {
                    if ( bufs_.size( ) == 0 )
                    {
                        return 0;
                    }
                    return bufs_.front( ).Start( );
                }

                buffer::gps_second_type
                remaining( ) const
                {
                    if ( bufs_.size( ) == 0 )
                    {
                        return 0;
                    }
                    return start( ) + stride_ - cur_;
                }

                void
                advance( buffer::gps_second_type delta )
                {
                    cur_ += delta;
                }

                buffer::gps_second_type
                append_data_from( indexed_buffers& other )
                {
                    if ( other.bufs_.size( ) != bufs_.size( ) )
                    {
                        return 0;
                    }
                    if ( other.cur( ) != cur( ) )
                    {
                        return 0;
                    }
                    buffer::gps_second_type secs_appended =
                        std::min< buffer::gps_second_type >(
                            remaining( ), other.remaining( ) );

                    buffer::gps_second_type src_offset_sec =
                        other.cur( ) - other.start( );
                    buffer::gps_second_type dest_offset_sec = cur( ) - start( );

                    for ( buffers_type::size_type i = 0; i < bufs_.size( );
                          ++i )
                    {
                        NDS::buffer& src_buf = other.bufs_[ i ];
                        NDS::buffer& dest_buf = bufs_[ i ];

                        auto src = const_cast< char* >(
                            reinterpret_cast< const char* >(
                                src_buf.cbegin< void >( ) ) );
                        src += src_buf.samples_to_bytes(
                            src_buf.seconds_to_samples( src_offset_sec, 0 ) );
                        auto dest = const_cast< char* >(
                            reinterpret_cast< const char* >(
                                dest_buf.cbegin< void >( ) ) );
                        dest += dest_buf.samples_to_bytes(
                            dest_buf.seconds_to_samples( dest_offset_sec, 0 ) );

                        buffer::size_type copy_bytes = src_buf.samples_to_bytes(
                            src_buf.seconds_to_samples( secs_appended, 0 ) );
                        std::copy( src, src + copy_bytes, dest );
                    }
                    advance( secs_appended );
                    other.advance( secs_appended );
                    return secs_appended;
                }

                void
                apply_gap_handler( detail::gap_handler&    handler,
                                   buffer::gps_second_type gps_stop )
                {
                    if ( gps_stop <= cur_ )
                    {
                        return;
                    }
                    buffer::gps_second_type start_time = start( );
                    if ( gps_stop > start_time + stride_ )
                    {
                        gps_stop = start_time + stride_;
                    }

                    for ( buffers_type::iterator cur_buf = bufs_.begin( );
                          cur_buf != bufs_.end( );
                          ++cur_buf )
                    {
                        std::unique_ptr< detail::delayed_gap_handler >
                            delayed_handler( handler.fill_gap(
                                *cur_buf,
                                cur_buf->seconds_to_samples( cur_ -
                                                             start_time ),
                                cur_buf->seconds_to_samples( gps_stop -
                                                             start_time ) ) );
                        if ( delayed_handler )
                        {
                            delay_.push_back( std::move( delayed_handler ) );
                        }
                    }
                    advance( gps_stop - cur_ );
                }

                void
                apply_delayed_handlers( )
                {
                    for ( std::vector< std::unique_ptr<
                              detail::delayed_gap_handler > >::iterator cur =
                              delay_.begin( );
                          cur != delay_.end( );
                          ++cur )
                    {
                        if ( *cur )
                        {
                            ( **cur )( );
                        }
                    }
                    delay_.clear( );
                }
            };

            buffer::gps_second_type                cur_gps_;
            buffer::gps_second_type                gps_start_;
            buffer::gps_second_type                gps_stop_;
            buffer::gps_second_type                stride_;
            std::unique_ptr< detail::gap_handler > gap_handler_;
            std::vector< NDS::channel >            channel_list_;

            indexed_buffers current_buffer_;
            indexed_buffers pending_buffer_;

            bool is_more_data_available_;

        public:
            iterate_handler_with_simple_gaps(
                buffer::gps_second_type                       gps_start,
                buffer::gps_second_type                       gps_stop,
                buffer::gps_second_type                       stride,
                const connection::channel_names_type&         channel_names,
                std::shared_ptr< NDS::detail::conn_p_type >&& parent,
                std::unique_ptr< detail::gap_handler >        ghandler )
                : iterate_handler( std::move( parent ) ), cur_gps_( gps_start ),
                  gps_start_( gps_start ), gps_stop_( gps_stop ),
                  stride_( stride ), gap_handler_( std::move( ghandler ) ),
                  channel_list_( ), current_buffer_( ), pending_buffer_( ),
                  is_more_data_available_( true )
            {
                channel_list_.reserve( channel_names.size( ) );
                parent->issue_iterate( gps_start_,
                                       gps_stop_,
                                       stride_,
                                       channel_names,
                                       &channel_list_ );
                if ( stride_ == 0 )
                {
                    stride_ = parent->calculate_stride(
                        gps_start_, gps_stop_, channel_list_ );
                }
                current_buffer_.initialize( channel_list_, cur_gps_, stride_ );
            }
            ~iterate_handler_with_simple_gaps( ) override = default;

            bool
            has_next( ) override
            {
                return conn( ) && ( cur_gps_ < gps_stop_ );
            }

            void
            next( buffers_type& output ) override
            {
                auto parent = conn( );
                if ( !has_next( ) || !parent )
                {
                    throw std::out_of_range( "No Next" );
                }

                buffer::gps_second_type segment_end = cur_gps_ + stride_;
                if ( current_buffer_.cur( ) > segment_end )
                {
                    throw std::runtime_error(
                        "Impossible condition triggered, gap "
                        "handled iterator went beyond "
                        "bounds" );
                }
                while ( current_buffer_.cur( ) < segment_end )
                {
                    /* loop doing one of:
                     * 1. take from the current pending_buffer_ if it is
                     * adjacent
                     * and valid
                     * 2. if pending is valid gap fill up to the end of the
                     * segment
                     * or the pending
                     *    buffer, which ever is fist
                     * 3. if pending is not valid and there is more data fill
                     * pending
                     * 4. gap fill the remainder of the segment
                     */

                    if ( pending_buffer_.valid( ) )
                    {
                        if ( current_buffer_.cur( ) == pending_buffer_.cur( ) )
                        {
                            current_buffer_.append_data_from( pending_buffer_ );
                        }
                        else if ( current_buffer_.cur( ) <
                                  pending_buffer_.cur( ) )
                        {
                            current_buffer_.apply_gap_handler(
                                *gap_handler_,
                                std::min( pending_buffer_.cur( ),
                                          segment_end ) );
                        }
                    }
                    else if ( is_more_data_available_ )
                    {
                        try
                        {
                            buffers_type tmp;
                            parent->next_raw_buffer( tmp );
                            pending_buffer_.reset( tmp );
                            if ( pending_buffer_.bufs( ).size( ) == 0 )
                            {
                                is_more_data_available_ = false;
                            }
                        }
                        catch ( ... )
                        {
                            is_more_data_available_ = false;
                        }
                    }
                    else
                    {
                        current_buffer_.apply_gap_handler( *gap_handler_,
                                                           segment_end );
                    }
                }
                current_buffer_.apply_delayed_handlers( );
                output.swap( current_buffer_.bufs( ) );
                cur_gps_ += stride_;
                if ( cur_gps_ + stride_ > gps_stop_ )
                {
                    stride_ = gps_stop_ - cur_gps_;
                }
                current_buffer_.initialize( channel_list_, cur_gps_, stride_ );
            }
        };

        class iterate_available_handler : public detail::iterate_handler
        {
            detail::request_fragments_type fragment_list_;
            buffer::size_type              cur_segment_;
            buffer::gps_second_type        cur_gps_;
            buffer::gps_second_type        gps_start_;
            buffer::gps_second_type        gps_stop_;
            buffer::gps_second_type        max_stride_;
            channel::channel_names_type    names_;
            buffers_type                   next_entry_;

            void
            setup_next_step( NDS::detail::conn_p_type& parent )
            {
                NDS::detail::dout( ) << "setup_next_iterate_step()"
                                     << std::endl;
                if ( cur_gps_ == 0 )
                {
                    NDS::detail::dout( ) << "Finding first segment"
                                         << std::endl;
                    cur_segment_ = 0;
                    cur_gps_ = gps_start_;
                }

                if ( fragment_list_[ 0 ].time_spans.empty( ) )
                {
                    cur_segment_++;
                    return;
                }
                simple_segment_list_type::value_type cur_segment =
                    ( fragment_list_[ 0 ].time_spans )[ cur_segment_ ];

                NDS::detail::dout( ) << "status " << cur_gps_ << " "
                                     << cur_segment.gps_start << "-"
                                     << cur_segment.gps_stop << std::endl;
                if ( cur_gps_ <= cur_segment.gps_start )
                {
                    NDS::detail::dout( ) << "Starting a segment " << std::endl;
                    buffer::gps_second_type delta =
                        cur_segment.gps_stop - cur_segment.gps_start;
                    buffer::gps_second_type stride =
                        ( max_stride_ > delta ? delta : max_stride_ );
                    NDS::detail::dout( ) << cur_segment.gps_start << " "
                                         << cur_segment.gps_stop << " "
                                         << stride << std::endl;
                    parent.issue_iterate( cur_segment.gps_start,
                                          cur_segment.gps_stop,
                                          stride,
                                          names_ );
                }
                else if ( cur_gps_ == cur_segment.gps_stop )
                {
                    NDS::detail::dout( ) << "Ending a segment " << std::endl;
                    cur_segment_++;
                    if ( cur_segment_ >=
                         fragment_list_[ 0 ].time_spans.size( ) )
                    {
                        return;
                    }
                    simple_segment_list_type::value_type new_segment =
                        ( fragment_list_[ 0 ].time_spans )[ cur_segment_ ];
                    cur_gps_ = new_segment.gps_start;
                    buffer::gps_second_type delta =
                        new_segment.gps_stop - new_segment.gps_start;
                    buffer::gps_second_type stride =
                        ( max_stride_ > delta ? delta : max_stride_ );
                    NDS::detail::dout( ) << new_segment.gps_start << " "
                                         << new_segment.gps_stop << " "
                                         << stride;
                    parent.issue_iterate( new_segment.gps_start,
                                          new_segment.gps_stop,
                                          stride,
                                          names_ );
                }
                else
                {
                    NDS::detail::dout( ) << "Mid segment" << std::endl;
                }
            }

        public:
            iterate_available_handler(
                buffer::gps_second_type                       gps_start,
                buffer::gps_second_type                       gps_stop,
                buffer::gps_second_type                       stride,
                const channel::channel_names_type&            channel_names,
                std::shared_ptr< NDS::detail::conn_p_type >&& parent )
                : iterate_handler( std::move( parent ) ), fragment_list_( ),
                  cur_segment_( 0 ), cur_gps_( 0 ), gps_start_( gps_start ),
                  gps_stop_( gps_stop ), max_stride_( stride ),
                  names_( channel_names )
            {
                if ( parent->protocol == connection::PROTOCOL_ONE ||
                     gps_start == 0 )
                {
                    NDS::detail::dout( ) << "Fast path" << std::endl;
                    // NDS1 and online data do not get special treatement
                    parent->issue_iterate(
                        gps_start, gps_stop, stride, channel_names );
                    return;
                }
                NDS::detail::dout( ) << "Planning fetches" << std::endl;

                buffers_type retval;
                retval.resize( channel_names.size( ) );
                parent->plan_fetches(
                    gps_start_, gps_stop_, names_, retval, fragment_list_ );

                // do not know how to deal with gaps that do not line up
                if ( fragment_list_.size( ) != 1 )
                {
                    throw connection::daq_error(
                        DAQD_NOT_FOUND,
                        "The requested channels have "
                        "different/non-identical gaps." );
                }

                setup_next_step( *( parent.get( ) ) );
            };

            ~iterate_available_handler( ) override = default;

            bool
            has_next( ) override
            {
                if ( next_entry_.size( ) > 0 )
                {
                    return true;
                }
                auto parent = conn( );
                if ( cur_gps_ >= gps_stop_ || !parent )
                {
                    return false;
                }
                try
                {
                    parent->next_raw_buffer( next_entry_ );
                    if ( next_entry_.size( ) == 0 )
                    {
                        return false;
                    }
                    cur_gps_ = next_entry_[ 0 ].Stop( );
                    setup_next_step( *parent );
                }
                catch ( ... )
                {
                    return false;
                }
                return true;
            }

            void
            next( buffers_type& output ) override
            {
                buffers_type retval;

                if ( next_entry_.size( ) > 0 )
                {
                    output.swap( next_entry_ );
                    next_entry_.clear( );
                    return;
                }
                auto parent = conn( );
                if ( !parent )
                {
                    throw( std::out_of_range( "No Next" ) );
                }
                parent->next_raw_buffer( retval );
                if ( retval.empty( ) )
                {
                    throw( std::out_of_range( "No Next" ) );
                }
                else
                {
                    cur_gps_ = retval[ 0 ].Stop( );
                    setup_next_step( *parent );
                }
                output.swap( retval );
            }
        };

        class iterate_full_handler : public detail::iterate_handler
        {
            buffer::gps_second_type     gps_start_;
            buffer::gps_second_type     gps_stop_;
            buffer::gps_second_type     gps_stride_;
            buffer::gps_second_type     cur_gps_;
            channel::channel_names_type names_;
            channels_type               channels_;
            epoch                       prev_epoch_;

        public:
            iterate_full_handler(
                buffer::gps_second_type                       gps_start,
                buffer::gps_second_type                       gps_stop,
                buffer::gps_second_type                       stride,
                const channel::channel_names_type&            channel_names,
                epoch                                         prev_epoch,
                const channels_type&                          channel_list,
                std::shared_ptr< NDS::detail::conn_p_type >&& parent )
                : iterate_handler( std::move( parent ) ),
                  gps_start_( gps_start ), gps_stop_( gps_stop ),
                  gps_stride_( stride ), cur_gps_( gps_start ),
                  names_( channel_names ), prev_epoch_( prev_epoch ),
                  channels_( channel_list ){

                  };

            ~iterate_full_handler( ) override = default;

            void
            advance( )
            {
                cur_gps_ += gps_stride_;
            };
            bool
            done( ) override
            {
                return ( gps_stride_ == 0 || cur_gps_ >= gps_stop_ );
            };

            bool
            has_next( ) override
            {
                if ( !conn( ) )
                {
                    return false;
                }
                if ( gps_start_ == 0 && gps_stop_ == 0 )
                {
                    return true;
                }
                return !done( );
            }

            void
            next( buffers_type& output ) override
            {
                buffers_type retval;

                NDS::detail::dout( ) << "next_full" << std::endl;

                auto parent = conn( );
                if ( !parent )
                {
                    throw std::out_of_range( "No next buffer" );
                }
                // fast path (online data)
                if ( gps_start_ == 0 )
                {
                    NDS::detail::dout( ) << "fast path" << std::endl;
                    parent->next_raw_buffer( retval );
                }
                else
                {
                    NDS::detail::dout( )
                        << "Disabling request in progress in order to fetch"
                        << std::endl;
                    parent->request_in_progress( false );

                    buffer::gps_second_type start = cur_gps_;
                    buffer::gps_second_type end = gps_stop_;
                    buffer::gps_second_type stride = gps_stride_;
                    buffer::gps_second_type stop = start + stride;

                    if ( stop > end )
                    {
                        stop = end;
                        stride = stop - start;
                    }

                    NDS::detail::dout( ) << "Issuing fetch( " << start << ", "
                                         << stop << " ... ) " << std::endl;
                    retval =
                        parent->fetch( start, stop, names_, &( channels_ ) );

                    advance( );
                    if ( done( ) )
                    {
                        NDS::detail::dout( ) << "iterate_full is complete"
                                             << std::endl;
                        parent->set_epoch( prev_epoch_.gps_start,
                                           prev_epoch_.gps_stop );
                    }
                    else
                    {
                        NDS::detail::dout( ) << "There is more to iterate, "
                                                "request_in_progres enabled"
                                             << std::endl;
                        parent->request_in_progress( true );
                    }
                }
                output.swap( retval );
            }
        };
    }
}

#endif // NDS2_CLIENT_NDS_ITERATE_HANDLERS_HH