#include <DefaultSkipListRequestHandler.hpp>

Inheritance diagram for dunedaq::datahandlinglibs::DefaultSkipListRequestHandler< T >:

Collaboration diagram for dunedaq::datahandlinglibs::DefaultSkipListRequestHandler< T >:

Public Types
using	inherited = datahandlinglibs::DefaultRequestHandlerModel<T, datahandlinglibs::SkipListLatencyBufferModel<T>>
using	SkipListAcc = typename folly::ConcurrentSkipList<T>::Accessor
using	SkipListSkip = typename folly::ConcurrentSkipList<T>::Skipper
Public Types inherited from dunedaq::datahandlinglibs::DefaultRequestHandlerModel< T, datahandlinglibs::SkipListLatencyBufferModel< T > >
using	RDT
using	LBT
using	RequestResult
using	ResultCode
Public Types inherited from dunedaq::opmonlib::MonitorableObject
using	NodePtr = std::weak_ptr<MonitorableObject>
using	NewNodePtr = std::shared_ptr<MonitorableObject>
using	ElementId = std::string

Public Member Functions
	DefaultSkipListRequestHandler (std::shared_ptr< datahandlinglibs::SkipListLatencyBufferModel< T > > &latency_buffer, std::unique_ptr< datahandlinglibs::FrameErrorRegistry > &error_registry)
Public Member Functions inherited from dunedaq::datahandlinglibs::DefaultRequestHandlerModel< T, datahandlinglibs::SkipListLatencyBufferModel< T > >
	DefaultRequestHandlerModel (std::shared_ptr< datahandlinglibs::SkipListLatencyBufferModel< T > > &latency_buffer, std::unique_ptr< FrameErrorRegistry > &error_registry)
void	conf (const dunedaq::appmodel::DataHandlerModule *)
void	scrap (const appfwk::DAQModule::CommandData_t &) override
void	start (const appfwk::DAQModule::CommandData_t &)
void	stop (const appfwk::DAQModule::CommandData_t &)
void	record (const appfwk::DAQModule::CommandData_t &args) override
void	cleanup_check () override
	Check if cleanup is necessary and execute it if necessary.
virtual void	periodic_data_transmission () override
	Periodic data transmission - relevant for trigger in particular.
void	issue_request (dfmessages::DataRequest datarequest, bool is_retry=false) override
	Issue a data request to the request handler.
virtual dunedaq::daqdataformats::timestamp_t	get_cutoff_timestamp ()
virtual bool	supports_cutoff_timestamp ()
void	reset_oldest_time ()
std::uint64_t	get_oldest_time () override
	Get oldest timestamp in the buffer.
Public Member Functions inherited from dunedaq::datahandlinglibs::RequestHandlerConcept< ReadoutType, LatencyBufferType >
	RequestHandlerConcept ()
virtual	~RequestHandlerConcept ()
	RequestHandlerConcept (const RequestHandlerConcept &)=delete
	RequestHandlerConcept is not copy-constructible.
RequestHandlerConcept &	operator= (const RequestHandlerConcept &)=delete
	RequestHandlerConcept is not copy-assginable.
	RequestHandlerConcept (RequestHandlerConcept &&)=delete
	RequestHandlerConcept is not move-constructible.
RequestHandlerConcept &	operator= (RequestHandlerConcept &&)=delete
	RequestHandlerConcept is not move-assignable.
Public Member Functions inherited from dunedaq::opmonlib::MonitorableObject
	MonitorableObject (const MonitorableObject &)=delete
MonitorableObject &	operator= (const MonitorableObject &)=delete
	MonitorableObject (MonitorableObject &&)=delete
MonitorableObject &	operator= (MonitorableObject &&)=delete
virtual	~MonitorableObject ()=default
auto	get_opmon_id () const noexcept
auto	get_opmon_level () const noexcept

Protected Member Functions
void	cleanup () override
void	skip_list_cleanup_request ()
Protected Member Functions inherited from dunedaq::datahandlinglibs::DefaultRequestHandlerModel< T, datahandlinglibs::SkipListLatencyBufferModel< T > >
daqdataformats::FragmentHeader	create_fragment_header (const dfmessages::DataRequest &dr)
std::unique_ptr< daqdataformats::Fragment >	create_empty_fragment (const dfmessages::DataRequest &dr)
void	dump_to_buffer (const void data, std::size_t size, void buffer, uint32_t buffer_pos, const std::size_t &buffer_size)
void	periodic_cleanups ()
void	periodic_data_transmissions ()
void	check_waiting_requests ()
std::vector< std::pair< void *, size_t > >	get_fragment_pieces (uint64_t start_win_ts, uint64_t end_win_ts, RequestResult &rres)
RequestResult	data_request (dfmessages::DataRequest dr) override
virtual void	generate_opmon_data () override
Protected Member Functions inherited from dunedaq::datahandlinglibs::RequestHandlerConcept< ReadoutType, LatencyBufferType >
const std::string &	resultCodeAsString (ResultCode rc)
Protected Member Functions inherited from dunedaq::opmonlib::MonitorableObject
	MonitorableObject ()=default
void	register_node (ElementId name, NewNodePtr)
void	publish (google::protobuf::Message &&, CustomOrigin &&co={}, OpMonLevel l=to_level(EntryOpMonLevel::kDefault)) const noexcept

Private Attributes
std::atomic< int >	m_found_requested_count { 0 }
std::atomic< int >	m_bad_requested_count { 0 }

Static Private Attributes
static const constexpr uint64_t	m_max_ts_diff = 625000000

Additional Inherited Members
Static Public Member Functions inherited from dunedaq::opmonlib::MonitorableObject
static bool	publishable_metric (OpMonLevel entry, OpMonLevel system) noexcept
Protected Types inherited from dunedaq::datahandlinglibs::RequestHandlerConcept< ReadoutType, LatencyBufferType >
enum	ResultCode { kFound = 0 , kNotFound , kTooOld , kNotYet , kPartial , kPartiallyOld , kCleanup , kUnknown }
Protected Attributes inherited from dunedaq::datahandlinglibs::DefaultRequestHandlerModel< T, datahandlinglibs::SkipListLatencyBufferModel< T > >
std::shared_ptr< datahandlinglibs::SkipListLatencyBufferModel< T > > &	m_latency_buffer
BufferedFileWriter	m_buffered_writer
utilities::ReusableThread	m_recording_thread
utilities::ReusableThread	m_cleanup_thread
utilities::ReusableThread	m_periodic_transmission_thread
std::map< dfmessages::DataRequest, int >	m_request_counter
std::mutex	m_cv_mutex
std::condition_variable	m_cv
std::atomic< bool >	m_cleanup_requested
std::atomic< int >	m_requests_running
std::vector< RequestElement >	m_waiting_requests
std::mutex	m_waiting_requests_lock
std::unique_ptr< boost::asio::thread_pool >	m_request_handler_thread_pool
size_t	m_num_request_handling_threads
std::unique_ptr< FrameErrorRegistry > &	m_error_registry
std::chrono::time_point< std::chrono::high_resolution_clock >	m_t0
std::atomic< bool >	m_run_marker
std::thread	m_waiting_queue_thread
std::atomic< bool >	m_recording
std::atomic< uint64_t >	m_next_timestamp_to_record
bool	m_configured
float	m_pop_limit_pct
float	m_pop_size_pct
unsigned	m_pop_limit_size
size_t	m_buffer_capacity
daqdataformats::SourceID	m_sourceid
uint16_t	m_detid
std::string	m_output_file
size_t	m_stream_buffer_size
bool	m_recording_configured
bool	m_warn_on_timeout
bool	m_warn_about_empty_buffer
uint32_t	m_periodic_data_transmission_ms
std::vector< std::string >	m_frag_out_conn_ids
std::atomic< int >	m_pop_counter
std::atomic< int >	m_num_buffer_cleanups
std::atomic< int >	m_pop_reqs
std::atomic< int >	m_pops_count
std::atomic< int >	m_occupancy
std::atomic< int >	m_num_requests_found
std::atomic< int >	m_num_requests_bad
std::atomic< int >	m_num_requests_old_window
std::atomic< int >	m_num_requests_delayed
std::atomic< int >	m_num_requests_uncategorized
std::atomic< int >	m_num_requests_timed_out
std::atomic< int >	m_handled_requests
std::atomic< int >	m_response_time_acc
std::atomic< int >	m_response_time_min
std::atomic< int >	m_response_time_max
std::atomic< int >	m_payloads_written
std::atomic< int >	m_bytes_written
std::atomic< uint64_t >	m_num_periodic_sent
std::atomic< uint64_t >	m_num_periodic_send_failed
std::atomic< uint64_t >	m_oldest_timestamp
int	m_fragment_send_timeout_ms
Protected Attributes inherited from dunedaq::datahandlinglibs::RequestHandlerConcept< ReadoutType, LatencyBufferType >
std::map< ResultCode, std::string >	ResultCodeStrings

Detailed Description

template<class T>
class dunedaq::datahandlinglibs::DefaultSkipListRequestHandler< T >

Definition at line 30 of file DefaultSkipListRequestHandler.hpp.

Member Typedef Documentation

◆ inherited

template<class T>

using dunedaq::datahandlinglibs::DefaultSkipListRequestHandler< T >::inherited = datahandlinglibs::DefaultRequestHandlerModel<T, datahandlinglibs::SkipListLatencyBufferModel<T>>

Definition at line 34 of file DefaultSkipListRequestHandler.hpp.

◆ SkipListAcc

template<class T>

using dunedaq::datahandlinglibs::DefaultSkipListRequestHandler< T >::SkipListAcc = typename folly::ConcurrentSkipList<T>::Accessor

Definition at line 35 of file DefaultSkipListRequestHandler.hpp.

◆ SkipListSkip

template<class T>

using dunedaq::datahandlinglibs::DefaultSkipListRequestHandler< T >::SkipListSkip = typename folly::ConcurrentSkipList<T>::Skipper

Definition at line 36 of file DefaultSkipListRequestHandler.hpp.

Constructor & Destructor Documentation

◆ DefaultSkipListRequestHandler()

template<class T>

dunedaq::datahandlinglibs::DefaultSkipListRequestHandler< T >::DefaultSkipListRequestHandler	(	std::shared_ptr< datahandlinglibs::SkipListLatencyBufferModel< T > > &	latency_buffer,
		std::unique_ptr< datahandlinglibs::FrameErrorRegistry > &	error_registry )

inline

Definition at line 39 of file DefaultSkipListRequestHandler.hpp.

    : datahandlinglibs::DefaultRequestHandlerModel<T, datahandlinglibs::SkipListLatencyBufferModel<T>>(
        latency_buffer,
        error_registry)
  {
    TLOG_DEBUG(TLVL_WORK_STEPS) << "DefaultSkipListRequestHandler created...";
  }

Member Function Documentation

◆ cleanup()

template<class T>

void dunedaq::datahandlinglibs::DefaultSkipListRequestHandler< T >::cleanup ( )

inlineoverrideprotectedvirtual

Reimplemented from dunedaq::datahandlinglibs::DefaultRequestHandlerModel< T, datahandlinglibs::SkipListLatencyBufferModel< T > >.

Definition at line 50 of file DefaultSkipListRequestHandler.hpp.

50{ skip_list_cleanup_request(); }

dunedaq::datahandlinglibs::DefaultSkipListRequestHandler::skip_list_cleanup_request

void skip_list_cleanup_request()

Definition DefaultSkipListRequestHandler.hxx:8

◆ skip_list_cleanup_request()

template<class T>

void dunedaq::datahandlinglibs::DefaultSkipListRequestHandler< T >::skip_list_cleanup_request ( )

protected

Definition at line 8 of file DefaultSkipListRequestHandler.hxx.

{
  size_t removed_ctr = 0;
  uint64_t tailts = 0; // oldest // NOLINT(build/unsigned)
  uint64_t headts = 0; // newest // NOLINT(build/unsigned)
  {
    SkipListAcc acc(inherited::m_latency_buffer->get_skip_list());
    auto tail = acc.last();
    auto head = acc.first();
    if (tail && head) {
      tailts = (*tail).get_timestamp(); // tailptr->get_timestamp();
      headts = (*head).get_timestamp(); // headptr->get_timestamp();
      TLOG_DEBUG(TLVL_WORK_STEPS) << "Cleanup REQUEST with "
                                  << "Oldest stored TS=" << headts << " "
                                  << "Newest stored TS=" << tailts;
      if (tailts - headts > m_max_ts_diff) { // ts differnce exceeds maximum
        ++(inherited::m_pop_reqs);
        uint64_t timediff = m_max_ts_diff; // NOLINT(build/unsigned)
        while (timediff >= m_max_ts_diff) {
          bool removed = acc.remove(*head);
          if (!removed) {
            TLOG_DEBUG(TLVL_WORK_STEPS) << "Unsuccesfull remove from SKL during cleanup: " << removed;
          } else {
            ++removed_ctr;
          }
          head = acc.first();
          headts = (*head).get_timestamp(); 
          timediff = tailts - headts;
        }
        inherited::m_pops_count += removed_ctr;
    
        // Update hte oldest timestamp monitorable
        inherited::m_oldest_timestamp = headts;
      }
    } else {
      TLOG_DEBUG(TLVL_WORK_STEPS) << "Didn't manage to get SKL head and tail!";
    }
  }
  inherited::m_num_buffer_cleanups++;
}

Member Data Documentation

◆ m_bad_requested_count

template<class T>

std::atomic<int> dunedaq::datahandlinglibs::DefaultSkipListRequestHandler< T >::m_bad_requested_count { 0 }

private

Definition at line 61 of file DefaultSkipListRequestHandler.hpp.

61{ 0 };

◆ m_found_requested_count

template<class T>

std::atomic<int> dunedaq::datahandlinglibs::DefaultSkipListRequestHandler< T >::m_found_requested_count { 0 }

private

Definition at line 60 of file DefaultSkipListRequestHandler.hpp.

60{ 0 };

◆ m_max_ts_diff

template<class T>

const constexpr uint64_t dunedaq::datahandlinglibs::DefaultSkipListRequestHandler< T >::m_max_ts_diff = 625000000

staticconstexprprivate

Definition at line 57 of file DefaultSkipListRequestHandler.hpp.

The documentation for this class was generated from the following files:

/github/workspace/dunedaq/sourcecode/datahandlinglibs/include/datahandlinglibs/models/DefaultSkipListRequestHandler.hpp
/github/workspace/dunedaq/sourcecode/datahandlinglibs/include/datahandlinglibs/models/detail/DefaultSkipListRequestHandler.hxx

Public Types

Public Member Functions

Protected Member Functions

Private Attributes

Static Private Attributes

Additional Inherited Members

Detailed Description

Member Typedef Documentation

◆ inherited

◆ SkipListAcc

◆ SkipListSkip

Constructor & Destructor Documentation

◆ DefaultSkipListRequestHandler()

Member Function Documentation

◆ cleanup()

◆ skip_list_cleanup_request()

Member Data Documentation

◆ m_bad_requested_count

◆ m_found_requested_count

◆ m_max_ts_diff