nds_connection_ptype.hh

Go to the documentation of this file.

#ifndef NDS_CONNECTION_DETAIL_PTYPE_HH
#define NDS_CONNECTION_DETAIL_PTYPE_HH

#include <cerrno>
#include <map>
#include <memory>
#include <sstream>
#include <vector>

#include "debug_stream.hh"

#include "daq_config.h"
#include "daqc.h"
#include "daqc_internal.h"
#include "daqc_response.h"

#include "nds_channel.hh"
#include "nds_buffer.hh"

#include "nds_gap_handler.hh"
#include "nds_request_fragment.hh"
#include "nds_iterate_handler.hh"

#include "nds_parameter_block.hh"

#include "nds_channel_cache.hh"

namespace NDS
{
    namespace detail
    {

        class basic_channel_filter
        {
        public:
            basic_channel_filter( channel::data_type        data_type_mask,
                                  channel::sample_rate_type min_sample_rate,
                                  channel::sample_rate_type max_sample_rate )
                : data_type_mask_( data_type_mask ),
                  min_sample_rate_( min_sample_rate ),
                  max_sample_rate_( max_sample_rate )
            {
            }
            bool
            operator( )( const channel& ch )
            {
                return (
                    ( ( ( ch.DataType( ) & data_type_mask_ ) != 0 ) ||
                      data_type_mask_ == NDS::channel::DEFAULT_DATA_MASK ) &&
                    ( ch.SampleRate( ) <= max_sample_rate_ ) &&
                    ( ch.SampleRate( ) >= min_sample_rate_ ) );
            }

        private:
            channel::data_type        data_type_mask_;
            channel::sample_rate_type min_sample_rate_;
            channel::sample_rate_type max_sample_rate_;
        };

        class push_back_channel
        {
        public:
            push_back_channel( std::vector< channel >& buffer )
                : buffer_( buffer )
            {
            }
            void
            operator( )( const channel& ch )
            {
                buffer_.push_back( ch );
            }

        private:
            std::vector< channel >& buffer_;
        };

        class count_channels
        {
        public:
            count_channels( ) : count_( 0 )
            {
            }
            void
            operator( )( const channel& ch )
            {
                ++count_;
            }

            size_t
            count( ) const
            {
                return count_;
            }

        private:
            size_t count_;
        };
    }

    namespace detail
    {

        class buffer_initializer
        {
        public:
            DLL_EXPORT
            buffer_initializer( buffer::gps_second_type gps_start,
                                buffer::gps_second_type gps_stop )
                : gps_start( gps_start ), gps_stop( gps_stop ){};

            DLL_EXPORT
            void reset_buffer( buffer*        cur_buffer,
                               const channel& channel_info ) const;

        private:
            buffer::gps_second_type gps_start;
            buffer::gps_second_type gps_stop;
        };

        class daq_accessor
        {
        public:
            daq_accessor( daq_t& handle ) : handle_( handle ){};

            daq_t*
            operator( )( )
            {
                return &handle_;
            };

        private:
            daq_t& handle_;
        };

        //---------------------------------------------------------------------
        // conn_p_type
        //---------------------------------------------------------------------
        struct conn_p_type : public std::enable_shared_from_this< conn_p_type >
        {
            enum class iterate_finalize_reason
            {
                FINISHED,
                ABORTED
            };

            typedef channel::channel_type     channel_type;
            typedef channel::data_type        data_type;
            typedef channel::sample_rate_type sample_rate_type;

            typedef std::map< std::string, daq_channel_t >
                                                    channel_mem_cache_type;
            typedef NDS::detail::channel_cache_nds1 channel_cache_type;
            typedef long                            time_type;

            typedef nds_socket_type socket_t;

            connection::host_type     host;
            connection::port_type     port;
            connection::protocol_type protocol;
            daq_t                     handle;
            daq_accessor              accesor_;
            bool                      connected;
            channel_cache_type        channel_cache_;
            channel_mem_cache_type    channel_mem_cache_;

            time_type request_start_time_;
            time_type request_end_time_;
            bool      request_in_progress_;

            NDS::parameters parameters_;

            DLL_EXPORT
            explicit conn_p_type( const NDS::parameters& params );

            DLL_EXPORT
            ~conn_p_type( );

            void connect( );

            availability_list_type get_availability(
                const epoch&                          time_span,
                const connection::channel_names_type& channel_names );
            availability_list_type get_availability(
                buffer::gps_second_type               gps_start,
                buffer::gps_second_type               gps_stop,
                const connection::channel_names_type& channel_names );

            bool check( buffer::gps_second_type               gps_start,
                        buffer::gps_second_type               gps_stop,
                        const connection::channel_names_type& channel_names );

            bool
            has_gaps( buffer::gps_second_type               gps_start,
                      buffer::gps_second_type               gps_stop,
                      const connection::channel_names_type& channel_names );

            buffers_type
            fetch( buffer::gps_second_type               gps_start,
                   buffer::gps_second_type               gps_stop,
                   const connection::channel_names_type& channel_names,
                   channels_type* reference_channels = nullptr );

            size_t count_channels_nds1(
                std::string  channel_glob,
                channel_type channel_type_mask = channel::DEFAULT_CHANNEL_MASK,
                data_type    data_type_mask = channel::DEFAULT_DATA_MASK,
                sample_rate_type min_sample_rate = channel::MIN_SAMPLE_RATE,
                sample_rate_type max_sample_rate = channel::MAX_SAMPLE_RATE );

            size_t count_channels_nds2(
                std::string  channel_glob,
                channel_type channel_type_mask = channel::DEFAULT_CHANNEL_MASK,
                data_type    data_type_mask = channel::DEFAULT_DATA_MASK,
                sample_rate_type min_sample_rate = channel::MIN_SAMPLE_RATE,
                sample_rate_type max_sample_rate = channel::MAX_SAMPLE_RATE );

            size_t
            count_channels(
                std::string  channel_glob,
                channel_type channel_type_mask = channel::DEFAULT_CHANNEL_MASK,
                data_type    data_type_mask = channel::DEFAULT_DATA_MASK,
                sample_rate_type min_sample_rate = channel::MIN_SAMPLE_RATE,
                sample_rate_type max_sample_rate = channel::MAX_SAMPLE_RATE )
            {
                if ( protocol == connection::protocol_type::PROTOCOL_ONE )
                {
                    return count_channels_nds1( channel_glob,
                                                channel_type_mask,
                                                data_type_mask,
                                                min_sample_rate,
                                                max_sample_rate );
                }
                return count_channels_nds2( channel_glob,
                                            channel_type_mask,
                                            data_type_mask,
                                            min_sample_rate,
                                            max_sample_rate );
            }

            void find_channels( channels_type&                       output,
                                const NDS::channel_predicate_object& pred );

            void find_channels_nds1(
                channels_type& output,
                std::string    channel_glob,
                channel_type channel_type_mask = channel::DEFAULT_CHANNEL_MASK,
                data_type    data_type_mask = channel::DEFAULT_DATA_MASK,
                sample_rate_type min_sample_rate = channel::MIN_SAMPLE_RATE,
                sample_rate_type max_sample_rate = channel::MAX_SAMPLE_RATE );

            void find_channels_nds2(
                channels_type& output,
                std::string    channel_glob,
                channel_type channel_type_mask = channel::DEFAULT_CHANNEL_MASK,
                data_type    data_type_mask = channel::DEFAULT_DATA_MASK,
                sample_rate_type min_sample_rate = channel::MIN_SAMPLE_RATE,
                sample_rate_type max_sample_rate = channel::MAX_SAMPLE_RATE );

            epochs_type get_epochs( );

            bool set_epoch_if_changed( const epoch& time_span );
            bool set_epoch( const std::string& epoch_name );

            bool set_epoch( buffer::gps_second_type gps_start,
                            buffer::gps_second_type gps_stop );

            epoch current_epoch( ) const;

            const channel::hash_type& hash( ) const;

            std::shared_ptr< detail::iterate_handler > dispatch_iterate(
                buffer::gps_second_type               gps_start,
                buffer::gps_second_type               gps_stop,
                buffer::gps_second_type               stride,
                const connection::channel_names_type& channel_names );

            void issue_iterate(
                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::vector< NDS::channel >* final_channel_list = nullptr );

            void finalize_iterate( detail::iterate_handler* handler,
                                   iterate_finalize_reason  reason );

            //            bool has_next();
            //
            //            buffers_type next();

            void next_raw_buffer( buffers_type& output );

            void shutdown( );

            void fill_gap( data_type          DataType,
                           channel::size_type DataSizeType,
                           unsigned char*     start,
                           unsigned char*     end );

            inline time_type
            request_start_time( ) const
            {
                return request_start_time_;
            }

            inline void
            request_start_time( time_type Value )
            {
                request_start_time_ = Value;
            }

            inline time_type
            request_end_time( ) const
            {
                return request_end_time_;
            }

            inline void
            request_end_time( time_type Value )
            {
                request_end_time_ = Value;
            }

            inline bool
            request_in_progress( ) const
            {
                return request_in_progress_;
            }

            inline void
            request_in_progress( bool Value )
            {
                request_in_progress_ = Value;
            }

            NDS::buffer::gps_second_type
            calculate_stride( NDS::buffer::gps_second_type gps_start,
                              NDS::buffer::gps_second_type gps_stop,
                              NDS::channels_type& selected_channels ) const;

            inline void
            termination_block( )
            {
                //-----------------------------------------------------------------
                // NDS1 transfers end with a 'termination block', an empty block
                // that is indistinguisable from a 'data not found' condition.
                // If this is an NDS1 connection, we must digest the termination
                // block.
                //-----------------------------------------------------------------
                if ( ( request_in_progress( ) ) &&
                     ( protocol == connection::protocol_type::PROTOCOL_ONE ) )
                {
                    int rc = daq_recv_next( &handle );

                    if ( rc != DAQD_NOT_FOUND )
                    {
                        //-----------------------------------------------------
                        // Unexpected error
                        //-----------------------------------------------------
                        throw connection::daq_error( rc );
                    }
                }
            }

            NDS::buffer::gps_second_type cur_nds1_gpstime( );

            void validate( ) const;

            void validate_daq( int RetCode ) const;

            void infer_daq_channel_info( const std::string& channel_name,
                                         daq_channel_t&     channel,
                                         time_type          gps );

            void channel_mask_to_query_type_strings(
                channel_type                channel_type_mask,
                std::vector< std::string >& queryTypes );

            // Helper functions for error messages
            std::string get_last_message( ) const throw( );

            std::string err_msg_unexpected_no_data_found(
                buffer::gps_second_type            gps_start,
                buffer::gps_second_type            gps_stop,
                const channel::channel_names_type& names );

            void
            plan_fetches( buffer::gps_second_type               gps_start,
                          buffer::gps_second_type               gps_stop,
                          const connection::channel_names_type& channel_names,
                          buffers_type&                         dest_buffers,
                          request_fragments_type&               retval );

            void
            sync_parameters( )
            {
                detail::parameter_accessor paccess( parameters_ );
                if ( handle.conceal )
                {
                    handle.conceal->max_command_count = static_cast< size_t >(
                        paccess( ).max_nds1_command_count( ) );
                }
            }

            void cycle_nds1_connection( );

        protected:
            epoch resolve_epoch( const std::string& epoch_name );
            void setup_daq_chanlist(
                buffer::gps_second_type               gps_start,
                const connection::channel_names_type& channel_names,
                bool&                                 have_minute_trends,
                double&                               bytes_per_sample,
                std::vector< NDS::channel >* final_channel_list = nullptr );

            void process_check_data_result( int result, bool gaps_ok );

            void fetch_fragment( request_fragment&         fragment,
                                 const buffer_initializer& initializer,
                                 bool buffers_initialized );

            std::vector< buffers_type > fetch_available(
                buffer::gps_second_type               gps_start,
                buffer::gps_second_type               gps_stop,
                const connection::channel_names_type& channel_names );

            std::weak_ptr< detail::iterate_handler > iterate_handler_;

            epochs_type        epoch_cache_;
            epoch              current_epoch_;
            channel::hash_type hash_;

            void load_epochs_to_cache( );

            channel _parse_nds2_get_channel_line( char* buffer );

            bool _read_uint4( socket_t fd, uint4_type* dest );

            bool _read_buffer( socket_t fd, void* dest, size_t size );

            template < typename Filter, typename Function >
            void retreive_channels_from_nds2(
                const std::vector< std::string >& types,
                const std::string&                channel_glob,
                Filter&                           filt,
                Function&                         fn );

            std::shared_ptr< detail::iterate_handler > iterate_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< detail::iterate_handler >
            iterate_fast( 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< detail::iterate_handler > iterate_available(
                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< detail::iterate_handler >
            iterate_full( buffer::gps_second_type               gps_start,
                          buffer::gps_second_type               gps_stop,
                          buffer::gps_second_type               stride,
                          const connection::channel_names_type& channel_names );

        private:
            static bool initialized;
        };

        template < typename Filter, typename Function >
        void
        conn_p_type::retreive_channels_from_nds2(
            const std::vector< std::string >& types,
            const std::string&                channel_glob,
            Filter&                           filter,
            Function&                         fn )
        {
            std::vector< char > lineBuffer;
            lineBuffer.resize( 512 );

            for ( std::vector< std::string >::const_iterator cur =
                      types.begin( );
                  cur != types.end( );
                  ++cur )
            {
                NDS::detail::dout( ) << "Retreiving channels for type " << *cur
                                     << std::endl;
                {
                    std::ostringstream cmdBuf;
                    // cmdBuf << "` " << current_epoch_.gps_start << " " <<
                    // cur_type;
                    if ( current_epoch_.gps_stop ==
                         current_epoch_.gps_start + 1 )
                    {
                        cmdBuf << "get-channels " << current_epoch_.gps_start
                               << " ";
                    }
                    else
                    {
                        cmdBuf << "get-channels 0 ";
                    }
                    cmdBuf << *cur;
                    if ( channel_glob.compare( "*" ) != 0 )
                    {
                        cmdBuf << " {" << channel_glob << "}";
                    }
                    cmdBuf << ";\n";
                    validate_daq(
                        daq_send( &( handle ), cmdBuf.str( ).c_str( ) ) );
                    NDS::detail::dout( ) << "Sent command '" << cmdBuf.str( )
                                         << "'" << std::endl;
                }

                socket_t   fd = handle.conceal->sockfd;
                uint4_type count = 0;
                NDS::detail::dout( ) << "Reading channel count" << std::endl;
                if ( !_read_uint4( fd, &count ) )
                {
                    throw connection::unexpected_channels_received_error( );
                }
                // result.reserve(count);

                NDS::detail::dout( ) << "Expecting " << count << "channels"
                                     << std::endl;
                for ( uint4_type i = 0; i < count; ++i )
                {
                    uint4_type line_size = 0;
                    if ( !_read_uint4( handle.conceal->sockfd, &line_size ) )
                    {
                        throw connection::unexpected_channels_received_error( );
                    }
                    if ( line_size < lineBuffer.size( ) )
                    {
                        lineBuffer.resize( line_size );
                    }
                    if ( !_read_buffer( handle.conceal->sockfd,
                                        &( lineBuffer[ 0 ] ),
                                        line_size ) )
                    {
                        throw connection::unexpected_channels_received_error( );
                    }
                    channel curCh =
                        _parse_nds2_get_channel_line( &( lineBuffer[ 0 ] ) );
                    if ( filter( curCh ) )
                    {
                        fn( curCh );
                    }
                }
            }
        }
    } // namespace detail

} // namespace NDS

#endif // NDS_CONNECTION_DETAIL_PTYPE_HH