dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType > Class Template Reference

#include <DataHandlingModel.hpp>

Inheritance diagram for dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >:

Collaboration diagram for dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >:

Public Types
using	RDT = ReadoutType
using	RHT = RequestHandlerType
using	LBT = LatencyBufferType
using	RPT = RawDataProcessorType
using	IDT = InputDataType
using	timestamp_t = std::uint64_t
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
	DataHandlingModel (std::atomic< bool > &run_marker)
virtual	~DataHandlingModel ()=default
void	init (const appmodel::DataHandlerModule *modconf)
	Forward calls from the appfwk.
void	conf (const nlohmann::json &args)
void	scrap (const nlohmann::json &args)
void	start (const nlohmann::json &args)
void	stop (const nlohmann::json &args)
void	record (const nlohmann::json &args) override
void	consume_payload (RDT &&payload)
Public Member Functions inherited from dunedaq::datahandlinglibs::DataHandlingConcept
	DataHandlingConcept ()
virtual	~DataHandlingConcept ()
	DataHandlingConcept (const DataHandlingConcept &)=delete
	DataHandlingConcept is not copy-constructible.
DataHandlingConcept &	operator= (const DataHandlingConcept &)=delete
	DataHandlingConcept is not copy-assginable.
	DataHandlingConcept (DataHandlingConcept &&)=delete
	DataHandlingConcept is not move-constructible.
DataHandlingConcept &	operator= (DataHandlingConcept &&)=delete
	DataHandlingConcept 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

Public Attributes
std::function< void(RDT &&)>	m_consume_callback

Static Public Attributes
static constexpr timestamp_t	ns = 1
static constexpr timestamp_t	us = 1000 * ns
static constexpr timestamp_t	ms = 1000 * us
static constexpr timestamp_t	s = 1000 * ms

Protected Types
using	metric_t = dunedaq::datahandlinglibs::opmon::DataHandlerInfo
using	num_payload_t = std::remove_const<std::invoke_result<decltype(&metric_t::num_payloads),metric_t>::type>::type
using	sum_payload_t = std::remove_const<std::invoke_result<decltype(&metric_t::sum_payloads),metric_t>::type>::type
using	num_request_t = std::remove_const<std::invoke_result<decltype(&metric_t::num_requests),metric_t>::type>::type
using	sum_request_t = std::remove_const<std::invoke_result<decltype(&metric_t::sum_requests),metric_t>::type>::type
using	rawq_timeout_count_t = std::remove_const<std::invoke_result<decltype(&metric_t::num_data_input_timeouts),metric_t>::type>::type
using	num_lb_insert_failures_t = std::remove_const<std::invoke_result<decltype(&metric_t::num_lb_insert_failures),metric_t>::type>::type
using	raw_receiver_ct = iomanager::ReceiverConcept<InputDataType>
using	request_receiver_ct = iomanager::ReceiverConcept<dfmessages::DataRequest>
using	timesync_sender_ct = iomanager::SenderConcept<dfmessages::TimeSync>

Protected Member Functions
void	process_item (RDT &payload)
void	run_consume ()
	Function that will be run in its own thread to read the raw packets from the connection and add them to the LB.
void	run_timesync ()
	Function that will be run in its own thread and sends periodic timesync messages by pushing them to the connection.
void	dispatch_requests (dfmessages::DataRequest &data_request)
virtual std::vector< RDT >	transform_payload (IDT &original) const
virtual void	generate_opmon_data () override
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

Protected Attributes
std::atomic< bool > &	m_run_marker
bool	m_callback_mode
bool	m_fake_trigger
bool	m_generate_timesync = false
int	m_current_fake_trigger_id
daqdataformats::SourceID	m_sourceid
daqdataformats::run_number_t	m_run_number
uint64_t	m_processing_delay_ticks
std::atomic< num_payload_t >	m_num_payloads { 0 }
std::atomic< sum_payload_t >	m_sum_payloads { 0 }
std::atomic< num_request_t >	m_num_requests { 0 }
std::atomic< sum_request_t >	m_sum_requests { 0 }
std::atomic< rawq_timeout_count_t >	m_rawq_timeout_count { 0 }
std::atomic< num_lb_insert_failures_t >	m_num_lb_insert_failures { 0 }
std::atomic< int >	m_stats_packet_count { 0 }
utilities::ReusableThread	m_consumer_thread
std::chrono::milliseconds	m_raw_receiver_timeout_ms
std::chrono::microseconds	m_raw_receiver_sleep_us
std::shared_ptr< raw_receiver_ct >	m_raw_data_receiver
std::string	m_raw_data_receiver_connection_name
std::shared_ptr< request_receiver_ct >	m_data_request_receiver
std::shared_ptr< timesync_sender_ct >	m_timesync_sender
utilities::ReusableThread	m_timesync_thread
std::string	m_timesync_connection_name
uint32_t	m_pid_of_current_process
std::shared_ptr< LatencyBufferType >	m_latency_buffer_impl
std::shared_ptr< RawDataProcessorType >	m_raw_processor_impl
std::shared_ptr< RequestHandlerType >	m_request_handler_impl
bool	m_request_handler_supports_cutoff_timestamp
std::unique_ptr< FrameErrorRegistry >	m_error_registry
std::chrono::time_point< std::chrono::high_resolution_clock >	m_t0

Additional Inherited Members
Static Public Member Functions inherited from dunedaq::opmonlib::MonitorableObject
static bool	publishable_metric (OpMonLevel entry, OpMonLevel system) noexcept

Detailed Description

template<class ReadoutType, class RequestHandlerType, class LatencyBufferType, class RawDataProcessorType, class InputDataType = ReadoutType>
class dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >

Definition at line 63 of file DataHandlingModel.hpp.

Member Typedef Documentation

IDT

template<class ReadoutType , class RequestHandlerType , class LatencyBufferType , class RawDataProcessorType , class InputDataType = ReadoutType>

using dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::IDT = InputDataType

Definition at line 71 of file DataHandlingModel.hpp.

LBT

template<class ReadoutType , class RequestHandlerType , class LatencyBufferType , class RawDataProcessorType , class InputDataType = ReadoutType>

using dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::LBT = LatencyBufferType

Definition at line 69 of file DataHandlingModel.hpp.

metric_t

template<class ReadoutType , class RequestHandlerType , class LatencyBufferType , class RawDataProcessorType , class InputDataType = ReadoutType>

using dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::metric_t = dunedaq::datahandlinglibs::opmon::DataHandlerInfo

protected

Definition at line 171 of file DataHandlingModel.hpp.

num_lb_insert_failures_t

template<class ReadoutType , class RequestHandlerType , class LatencyBufferType , class RawDataProcessorType , class InputDataType = ReadoutType>

using dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::num_lb_insert_failures_t = std::remove_const<std::invoke_result<decltype(&metric_t::num_lb_insert_failures),metric_t>::type>::type

protected

Definition at line 177 of file DataHandlingModel.hpp.

num_payload_t

template<class ReadoutType , class RequestHandlerType , class LatencyBufferType , class RawDataProcessorType , class InputDataType = ReadoutType>

using dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::num_payload_t = std::remove_const<std::invoke_result<decltype(&metric_t::num_payloads),metric_t>::type>::type

protected

Definition at line 172 of file DataHandlingModel.hpp.

num_request_t

template<class ReadoutType , class RequestHandlerType , class LatencyBufferType , class RawDataProcessorType , class InputDataType = ReadoutType>

using dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::num_request_t = std::remove_const<std::invoke_result<decltype(&metric_t::num_requests),metric_t>::type>::type

protected

Definition at line 174 of file DataHandlingModel.hpp.

raw_receiver_ct

template<class ReadoutType , class RequestHandlerType , class LatencyBufferType , class RawDataProcessorType , class InputDataType = ReadoutType>

using dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::raw_receiver_ct = iomanager::ReceiverConcept<InputDataType>

protected

Definition at line 193 of file DataHandlingModel.hpp.

rawq_timeout_count_t

template<class ReadoutType , class RequestHandlerType , class LatencyBufferType , class RawDataProcessorType , class InputDataType = ReadoutType>

using dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::rawq_timeout_count_t = std::remove_const<std::invoke_result<decltype(&metric_t::num_data_input_timeouts),metric_t>::type>::type

protected

Definition at line 176 of file DataHandlingModel.hpp.

RDT

template<class ReadoutType , class RequestHandlerType , class LatencyBufferType , class RawDataProcessorType , class InputDataType = ReadoutType>

using dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::RDT = ReadoutType

Definition at line 67 of file DataHandlingModel.hpp.

request_receiver_ct

template<class ReadoutType , class RequestHandlerType , class LatencyBufferType , class RawDataProcessorType , class InputDataType = ReadoutType>

using dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::request_receiver_ct = iomanager::ReceiverConcept<dfmessages::DataRequest>

protected

Definition at line 198 of file DataHandlingModel.hpp.

RHT

template<class ReadoutType , class RequestHandlerType , class LatencyBufferType , class RawDataProcessorType , class InputDataType = ReadoutType>

using dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::RHT = RequestHandlerType

Definition at line 68 of file DataHandlingModel.hpp.

RPT

template<class ReadoutType , class RequestHandlerType , class LatencyBufferType , class RawDataProcessorType , class InputDataType = ReadoutType>

using dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::RPT = RawDataProcessorType

Definition at line 70 of file DataHandlingModel.hpp.

sum_payload_t

template<class ReadoutType , class RequestHandlerType , class LatencyBufferType , class RawDataProcessorType , class InputDataType = ReadoutType>

using dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::sum_payload_t = std::remove_const<std::invoke_result<decltype(&metric_t::sum_payloads),metric_t>::type>::type

protected

Definition at line 173 of file DataHandlingModel.hpp.

sum_request_t

template<class ReadoutType , class RequestHandlerType , class LatencyBufferType , class RawDataProcessorType , class InputDataType = ReadoutType>

using dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::sum_request_t = std::remove_const<std::invoke_result<decltype(&metric_t::sum_requests),metric_t>::type>::type

protected

Definition at line 175 of file DataHandlingModel.hpp.

timestamp_t

template<class ReadoutType , class RequestHandlerType , class LatencyBufferType , class RawDataProcessorType , class InputDataType = ReadoutType>

using dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::timestamp_t = std::uint64_t

Definition at line 74 of file DataHandlingModel.hpp.

timesync_sender_ct

template<class ReadoutType , class RequestHandlerType , class LatencyBufferType , class RawDataProcessorType , class InputDataType = ReadoutType>

using dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::timesync_sender_ct = iomanager::SenderConcept<dfmessages::TimeSync>

protected

Definition at line 207 of file DataHandlingModel.hpp.

Constructor & Destructor Documentation

DataHandlingModel()

template<class ReadoutType , class RequestHandlerType , class LatencyBufferType , class RawDataProcessorType , class InputDataType = ReadoutType>

dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::DataHandlingModel ( std::atomic< bool > & run_marker )

inlineexplicit

Definition at line 81 of file DataHandlingModel.hpp.

    : m_run_marker(run_marker)
    , m_callback_mode(false)
    , m_fake_trigger(false)
    , m_current_fake_trigger_id(0)
    , m_consumer_thread(0)
    , m_raw_receiver_timeout_ms(0)
    , m_raw_receiver_sleep_us(0)
    , m_raw_data_receiver(nullptr)
    , m_timesync_thread(0)
    , m_latency_buffer_impl(nullptr)
    , m_raw_processor_impl(nullptr)
  {
    m_pid_of_current_process = getpid();
  }

~DataHandlingModel()

template<class ReadoutType , class RequestHandlerType , class LatencyBufferType , class RawDataProcessorType , class InputDataType = ReadoutType>

virtual dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::~DataHandlingModel ( )

virtualdefault

Member Function Documentation

conf()

template<class RDT , class RHT , class LBT , class RPT , class IDT >

void dunedaq::datahandlinglibs::DataHandlingModel< RDT, RHT, LBT, RPT, IDT >::conf ( const nlohmann::json & args )

virtual

Implements dunedaq::datahandlinglibs::DataHandlingConcept.

Definition at line 98 of file DataHandlingModel.hxx.

{
  // Register callbacks if operating in that mode.
  if (m_callback_mode) {
    // Configure and register consume callback
    m_consume_callback = std::bind(&DataHandlingModel<RDT, RHT, LBT, RPT, IDT>::consume_payload, this, std::placeholders::_1);
 
    // Register callback
    auto dmcbr = DataMoveCallbackRegistry::get();
    dmcbr->register_callback<RDT>(m_raw_data_receiver_connection_name, m_consume_callback);
  }
 
  // Configure threads:
  m_consumer_thread.set_name("consumer", m_sourceid.id);
  if (m_generate_timesync)
    m_timesync_thread.set_name("timesync", m_sourceid.id);
}

consume_payload()

template<class RDT , class RHT , class LBT , class RPT , class IDT >

void dunedaq::datahandlinglibs::DataHandlingModel< RDT, RHT, LBT, RPT, IDT >::consume_payload

(

RDT &&

payload

)

Definition at line 313 of file DataHandlingModel.hxx.

{
 //m_rawq_timeout_count = 0;
 //m_num_payloads = 0;
 //m_sum_payloads = 0;
 //m_stats_packet_count = 0;
  m_raw_processor_impl->preprocess_item(&payload);
  if (m_request_handler_supports_cutoff_timestamp) {
    int64_t diff1 = payload.get_timestamp() - m_request_handler_impl->get_cutoff_timestamp();
    if (diff1 <= 0) {
      //m_request_handler_impl->increment_tardy_tp_count();
      ers::warning(DataPacketArrivedTooLate(ERS_HERE, m_run_number, payload.get_timestamp(),
                                            m_request_handler_impl->get_cutoff_timestamp(), diff1,
                                            (static_cast<double>(diff1)/62500.0)));
    }
  }
  if (!m_latency_buffer_impl->write(std::move(payload))) {
    TLOG_DEBUG(TLVL_TAKE_NOTE) << "***ERROR: Latency buffer is full and data was overwritten!";
    m_num_lb_insert_failures++;
  }
#warning RS FIXME: Post-processing delay feature is not implemented in callback consume!
  m_raw_processor_impl->postprocess_item(m_latency_buffer_impl->back());
  ++m_num_payloads;
  ++m_sum_payloads;
  ++m_stats_packet_count;
}

dispatch_requests()

template<class RDT , class RHT , class LBT , class RPT , class IDT >

void dunedaq::datahandlinglibs::DataHandlingModel< RDT, RHT, LBT, RPT, IDT >::dispatch_requests ( dfmessages::DataRequest & data_request )

protected

Definition at line 423 of file DataHandlingModel.hxx.

{
  if (data_request.request_information.component != m_sourceid) {
     ers::error(RequestSourceIDMismatch(ERS_HERE, m_sourceid, data_request.request_information.component));
     return;
  }
  TLOG_DEBUG(TLVL_QUEUE_POP) << "Received DataRequest" 
    << " for trig/seq_number " << data_request.trigger_number << "." << data_request.sequence_number
    << ", runno " << data_request.run_number
    << ", trig timestamp " << data_request.trigger_timestamp
    << ", SourceID: " << data_request.request_information.component
    << ", window begin/end " << data_request.request_information.window_begin
    << "/" << data_request.request_information.window_end
    << ", dest: " << data_request.data_destination;
  m_request_handler_impl->issue_request(data_request);
  ++m_num_requests;
  ++m_sum_requests;
}

generate_opmon_data()

template<class RDT , class RHT , class LBT , class RPT , class IDT >

void dunedaq::datahandlinglibs::DataHandlingModel< RDT, RHT, LBT, RPT, IDT >::generate_opmon_data ( )

overrideprotectedvirtual

Hook for customisable pubblication. The function can throw, exception will be caught by the monitoring thread

Reimplemented from dunedaq::opmonlib::MonitorableObject.

Definition at line 174 of file DataHandlingModel.hxx.

 {
   opmon::DataHandlerInfo ri;
   ri.set_sum_payloads(m_sum_payloads.load());
   ri.set_num_payloads(m_num_payloads.exchange(0));
 
   ri.set_num_data_input_timeouts(m_rawq_timeout_count.exchange(0));
 
   auto now = std::chrono::high_resolution_clock::now();
   int new_packets = m_stats_packet_count.exchange(0);
   double seconds = std::chrono::duration_cast<std::chrono::microseconds>(now - m_t0).count() / 1000000.;
   m_t0 = now;
 
   // 08-May-2025, KAB: added a message to warn users when latency buffer inserts are failing.
   int local_num_lb_insert_failures = m_num_lb_insert_failures.exchange(0);
   if (local_num_lb_insert_failures != 0) {
     ers::warning(NonZeroLatencyBufferInsertFailures(ERS_HERE, local_num_lb_insert_failures, ri.num_payloads()));
   }
 
   ri.set_rate_payloads_consumed(new_packets / seconds / 1000.);
   ri.set_num_lb_insert_failures(local_num_lb_insert_failures);
   ri.set_sum_requests(m_sum_requests.load());
   ri.set_num_requests(m_num_requests.exchange(0));
   ri.set_last_daq_timestamp(m_raw_processor_impl->get_last_daq_time());
 
   this->publish(std::move(ri));
 }

init()

template<class RDT , class RHT , class LBT , class RPT , class IDT >

void dunedaq::datahandlinglibs::DataHandlingModel< RDT, RHT, LBT, RPT, IDT >::init ( const appmodel::DataHandlerModule * mcfg )

virtual

Forward calls from the appfwk.

Implements dunedaq::datahandlinglibs::DataHandlingConcept.

Definition at line 10 of file DataHandlingModel.hxx.

{
  // Setup request queues
  //setup_request_queues(mcfg);
  try {
    for (auto input : mcfg->get_inputs()) {
      if (input->get_data_type() == "DataRequest") {
        m_data_request_receiver = get_iom_receiver<dfmessages::DataRequest>(input->UID()) ;
      }
      else {
    m_raw_data_receiver_connection_name = input->UID();
        // Parse for prefix
        std::string conn_name = input->UID(); 
        const char delim = '_';
        std::vector<std::string> words;
        std::size_t start;
        std::size_t end = 0;
        while ((start = conn_name.find_first_not_of(delim, end)) != std::string::npos) {
          end = conn_name.find(delim, start);
          words.push_back(conn_name.substr(start, end - start));
        }
 
    TLOG_DEBUG(TLVL_WORK_STEPS) << "Initialize connection based on uid: " 
          << m_raw_data_receiver_connection_name << " front word: " << words.front();
 
    std::string cb_prefix("cb");
        if (words.front() == cb_prefix) {
          m_callback_mode = true;
        }
 
        if (!m_callback_mode) {
          m_raw_data_receiver = get_iom_receiver<IDT>(m_raw_data_receiver_connection_name);
          m_raw_receiver_timeout_ms = std::chrono::milliseconds(input->get_recv_timeout_ms());
        }
      }
    }
    for (auto output : mcfg->get_outputs()) {
      if (output->get_data_type() == "TimeSync") {
        m_generate_timesync = true;
        m_timesync_sender = get_iom_sender<dfmessages::TimeSync>(output->UID()) ;
        m_timesync_connection_name = output->UID();
        break;
      }
    }
  } catch (const ers::Issue& excpt) {
    throw ResourceQueueError(ERS_HERE, "raw_input or frag_output", "DataHandlingModel", excpt);
  }
 
  // Raw input connection sensibility check
  if (!m_callback_mode && m_raw_data_receiver == nullptr) {
    ers::error(ConfigurationError(ERS_HERE, m_sourceid, "Non callback mode, and receiver is unset!"));
  }
 
  // Instantiate functionalities
  m_error_registry.reset(new FrameErrorRegistry());
  m_error_registry->set_ers_metadata("DLH of SourceID[" + std::to_string(mcfg->get_source_id()) + "] ");
  m_latency_buffer_impl.reset(new LBT());
  m_raw_processor_impl.reset(new RPT(m_error_registry, mcfg->get_post_processing_enabled()));
  m_request_handler_impl.reset(new RHT(m_latency_buffer_impl, m_error_registry));
 
  register_node(mcfg->get_module_configuration()->get_latency_buffer()->UID(), m_latency_buffer_impl);
  register_node(mcfg->get_module_configuration()->get_data_processor()->UID(), m_raw_processor_impl);
  register_node(mcfg->get_module_configuration()->get_request_handler()->UID(), m_request_handler_impl);
 
  //m_request_handler_impl->init(args);
  //m_raw_processor_impl->init(args);
  m_request_handler_supports_cutoff_timestamp = m_request_handler_impl->supports_cutoff_timestamp();
  m_fake_trigger = false;
  m_raw_receiver_sleep_us = std::chrono::microseconds::zero();
  m_sourceid.id = mcfg->get_source_id();
  m_sourceid.subsystem = RDT::subsystem;
  m_processing_delay_ticks = mcfg->get_module_configuration()->get_post_processing_delay_ticks();
  
 
  // Configure implementations:
  m_raw_processor_impl->conf(mcfg);
  // Configure the latency buffer before the request handler so the request handler can check for alignment
  // restrictions
  try {
    m_latency_buffer_impl->conf(mcfg->get_module_configuration()->get_latency_buffer());
  } catch (const std::bad_alloc& be) {
    ers::error(ConfigurationError(ERS_HERE, m_sourceid, "Latency Buffer can't be allocated with size!"));
  }
  m_request_handler_impl->conf(mcfg);
}

process_item()

template<class RDT , class RHT , class LBT , class RPT , class IDT >

void dunedaq::datahandlinglibs::DataHandlingModel< RDT, RHT, LBT, RPT, IDT >::process_item ( RDT & payload )

protected

Definition at line 204 of file DataHandlingModel.hxx.

{
  m_raw_processor_impl->preprocess_item(&payload);
  if (m_request_handler_supports_cutoff_timestamp) {
    int64_t diff1 = payload.get_timestamp() - m_request_handler_impl->get_cutoff_timestamp();
    if (diff1 <= 0) {
      //m_request_handler_impl->increment_tardy_tp_count();
      ers::warning(DataPacketArrivedTooLate(ERS_HERE, m_run_number, payload.get_timestamp(),
                                            m_request_handler_impl->get_cutoff_timestamp(), diff1,
                                            (static_cast<double>(diff1)/62500.0)));
    }
  }
  if (!m_latency_buffer_impl->write(std::move(payload))) {
    //TLOG_DEBUG(TLVL_TAKE_NOTE) << "***ERROR: Latency buffer is full and data was overwritten!";
    m_num_lb_insert_failures++;
  }
  if (m_processing_delay_ticks ==0) {
    m_raw_processor_impl->postprocess_item(m_latency_buffer_impl->back());
    ++m_num_payloads;
    ++m_sum_payloads;
    ++m_stats_packet_count;
  }  
}

record()

template<class ReadoutType , class RequestHandlerType , class LatencyBufferType , class RawDataProcessorType , class InputDataType = ReadoutType>

void dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::record ( const nlohmann::json & args )

inlineoverridevirtual

Implements dunedaq::datahandlinglibs::DataHandlingConcept.

Definition at line 120 of file DataHandlingModel.hpp.

  { 
    m_request_handler_impl->record(args); 
  }

run_consume()

template<class RDT , class RHT , class LBT , class RPT , class IDT >

void dunedaq::datahandlinglibs::DataHandlingModel< RDT, RHT, LBT, RPT, IDT >::run_consume ( )

protectedvirtual

Function that will be run in its own thread to read the raw packets from the connection and add them to the LB.

Implements dunedaq::datahandlinglibs::DataHandlingConcept.

Definition at line 230 of file DataHandlingModel.hxx.

{
 
  TLOG_DEBUG(TLVL_WORK_STEPS) << "Consumer thread started...";
  m_rawq_timeout_count = 0;
  m_num_payloads = 0;
  m_sum_payloads = 0;
  m_stats_packet_count = 0;
 
  // Variables for book-keeping of delayed post-processing
  //timestamp_t oldest_ts=0;
  timestamp_t newest_ts=0;
  timestamp_t end_win_ts=0;
  bool first_cycle = true;
  auto last_post_proc_time = std::chrono::system_clock::now();
  auto now = last_post_proc_time;
  std::chrono::milliseconds milliseconds;
  RDT processed_element;
 
  while (m_run_marker.load()) {
    // Try to acquire data
 
    auto opt_payload = m_raw_data_receiver->try_receive(m_raw_receiver_timeout_ms);
 
    if (opt_payload) {
 
      IDT& original = opt_payload.value();
      
      if constexpr (std::is_same_v<IDT, RDT>) {
        process_item(original);
      } else {
        auto transformed = transform_payload(original);
        for (auto& i : transformed) {
          process_item(i);
        }   
      }
    } else {
      ++m_rawq_timeout_count;
      // Protection against a zero sleep becoming a yield
      if ( m_raw_receiver_sleep_us != std::chrono::microseconds::zero())
        std::this_thread::sleep_for(m_raw_receiver_sleep_us);
    }
    
    // Add here a possible deferral of the post processing, to allow elements being reordered in the LB
    // Basically, find data older than a certain timestamp and process all data since the last post-processed element up to that value
    if (m_processing_delay_ticks !=0 && m_latency_buffer_impl->occupancy() > 0) {
      now = std::chrono::system_clock::now();
      milliseconds = std::chrono::duration_cast<std::chrono::milliseconds>(now - last_post_proc_time);
 
      if (milliseconds.count() > 1) {
        last_post_proc_time = now;
    // Get the LB boundaries
    auto tail = m_latency_buffer_impl->back();
        newest_ts = tail->get_timestamp();
        
        if (first_cycle) {
      auto head = m_latency_buffer_impl->front();
          processed_element.set_timestamp(head->get_timestamp()); 
          first_cycle = false;
      TLOG() << "***** First pass post processing *****" ;
        }
        
    if (newest_ts - processed_element.get_timestamp() > m_processing_delay_ticks) {
          end_win_ts = newest_ts - m_processing_delay_ticks; 
      auto start_iter=m_latency_buffer_impl->lower_bound(processed_element, false);
      processed_element.set_timestamp(end_win_ts);
      auto end_iter=m_latency_buffer_impl->lower_bound(processed_element, false);
 
      for (auto it = start_iter; it!= end_iter; ++it) { 
              m_raw_processor_impl->postprocess_item(&(*it));
              ++m_num_payloads;
              ++m_sum_payloads;
              ++m_stats_packet_count;
      }
     }
      }
    }
  }
  TLOG_DEBUG(TLVL_WORK_STEPS) << "Consumer thread joins... ";
}

run_timesync()

template<class RDT , class RHT , class LBT , class RPT , class IDT >

void dunedaq::datahandlinglibs::DataHandlingModel< RDT, RHT, LBT, RPT, IDT >::run_timesync ( )

protectedvirtual

Function that will be run in its own thread and sends periodic timesync messages by pushing them to the connection.

Implements dunedaq::datahandlinglibs::DataHandlingConcept.

Definition at line 342 of file DataHandlingModel.hxx.

{
  TLOG_DEBUG(TLVL_WORK_STEPS) << "TimeSync thread started...";
  m_num_requests = 0;
  m_sum_requests = 0;
  uint64_t msg_seqno = 0;
  timestamp_t prev_timestamp = 0;
  auto once_per_run = true;
  size_t zero_timestamp_count = 0;
  size_t duplicate_timestamp_count = 0;
  size_t total_timestamp_count = 0;
  while (m_run_marker.load()) {
    try {
      auto timesyncmsg = dfmessages::TimeSync(m_raw_processor_impl->get_last_daq_time());
      ++total_timestamp_count;
      // daq_time is zero for the first received timesync, and may
      // be the same as the previous daq_time if the data has
      // stopped flowing. In both cases we don't send the TimeSync
      if (timesyncmsg.daq_time != 0 && timesyncmsg.daq_time != prev_timestamp) {
        prev_timestamp = timesyncmsg.daq_time;
        timesyncmsg.run_number = m_run_number;
        timesyncmsg.sequence_number = ++msg_seqno;
        timesyncmsg.source_pid = m_pid_of_current_process;
        TLOG_DEBUG(TLVL_TIME_SYNCS) << "New timesync: daq=" << timesyncmsg.daq_time
          << " wall=" << timesyncmsg.system_time << " run=" << timesyncmsg.run_number
          << " seqno=" << timesyncmsg.sequence_number << " pid=" << timesyncmsg.source_pid;
        try {
            dfmessages::TimeSync timesyncmsg_copy(timesyncmsg);
          m_timesync_sender->send(std::move(timesyncmsg_copy), std::chrono::milliseconds(500));
        } catch (ers::Issue& excpt) {
          ers::warning(
            TimeSyncTransmissionFailed(ERS_HERE, m_sourceid, m_timesync_connection_name, excpt));
        }
          
        if (m_fake_trigger) {
          dfmessages::DataRequest dr;
          ++m_current_fake_trigger_id;
          dr.trigger_number = m_current_fake_trigger_id;
          dr.trigger_timestamp = timesyncmsg.daq_time > 500 * us ? timesyncmsg.daq_time - 500 * us : 0;
          auto width = 300000;
          uint offset = 100;
          dr.request_information.window_begin = dr.trigger_timestamp > offset ? dr.trigger_timestamp - offset : 0;
          dr.request_information.window_end = dr.request_information.window_begin + width;
          dr.request_information.component = m_sourceid;
          dr.data_destination = "data_fragments_q";
          TLOG_DEBUG(TLVL_WORK_STEPS) << "Issuing fake trigger based on timesync. "
            << " ts=" << dr.trigger_timestamp 
            << " window_begin=" << dr.request_information.window_begin
            << " window_end=" << dr.request_information.window_end;
          m_request_handler_impl->issue_request(dr);
 
          ++m_num_requests;
          ++m_sum_requests;
        }
      } else {
        if (timesyncmsg.daq_time == 0) {++zero_timestamp_count;}
        if (timesyncmsg.daq_time == prev_timestamp) {++duplicate_timestamp_count;}
        if (once_per_run) {
          TLOG() << "Timesync with DAQ time 0 won't be sent out as it's an invalid sync.";
          once_per_run = false;
        }
      }
    } catch (const iomanager::TimeoutExpired& excpt) {
      // ++m_timesyncqueue_timeout;
    }
    // Split up the 100ms sleep into 10 sleeps of 10ms, so we respond to "stop" quicker
    for (size_t i=0; i<10; ++i) {
      std::this_thread::sleep_for(std::chrono::milliseconds(10));
      if (!m_run_marker.load()) {
        break;
      }
    }
  }
  once_per_run = true;
  TLOG_DEBUG(TLVL_WORK_STEPS) << "TimeSync thread joins... (timestamp count, zero/same/total  = "
                              << zero_timestamp_count << "/" << duplicate_timestamp_count << "/"
                              << total_timestamp_count << ")";
}

scrap()

template<class ReadoutType , class RequestHandlerType , class LatencyBufferType , class RawDataProcessorType , class InputDataType = ReadoutType>

void dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::scrap ( const nlohmann::json & args )

inlinevirtual

Implements dunedaq::datahandlinglibs::DataHandlingConcept.

Definition at line 106 of file DataHandlingModel.hpp.

  {
    m_request_handler_impl->scrap(args);
    m_latency_buffer_impl->scrap(args);
    m_raw_processor_impl->scrap(args);
  }

start()

template<class RDT , class RHT , class LBT , class RPT , class IDT >

void dunedaq::datahandlinglibs::DataHandlingModel< RDT, RHT, LBT, RPT, IDT >::start ( const nlohmann::json & args )

virtual

Implements dunedaq::datahandlinglibs::DataHandlingConcept.

Definition at line 119 of file DataHandlingModel.hxx.

{
  // Reset opmon variables
  m_sum_payloads = 0;
  m_num_payloads = 0;
  m_sum_requests = 0;
  m_num_requests = 0;
  m_num_lb_insert_failures = 0;
  m_stats_packet_count = 0;
  m_rawq_timeout_count = 0;
 
  m_t0 = std::chrono::high_resolution_clock::now();
 
  m_run_number = args.value<dunedaq::daqdataformats::run_number_t>("run", 1);
 
  TLOG_DEBUG(TLVL_WORK_STEPS) << "Starting threads...";
  m_raw_processor_impl->start(args);
  m_request_handler_impl->start(args);
  if (!m_callback_mode) {
    m_consumer_thread.set_work(&DataHandlingModel<RDT, RHT, LBT, RPT, IDT>::run_consume, this);
  }
  if (m_generate_timesync) m_timesync_thread.set_work(&DataHandlingModel<RDT, RHT, LBT, RPT, IDT>::run_timesync, this);
  // Register callback to receive and dispatch data requests
  m_data_request_receiver->add_callback(
    std::bind(&DataHandlingModel<RDT, RHT, LBT, RPT, IDT>::dispatch_requests, this, std::placeholders::_1));
}

stop()

template<class RDT , class RHT , class LBT , class RPT , class IDT >

void dunedaq::datahandlinglibs::DataHandlingModel< RDT, RHT, LBT, RPT, IDT >::stop ( const nlohmann::json & args )

virtual

Implements dunedaq::datahandlinglibs::DataHandlingConcept.

Definition at line 148 of file DataHandlingModel.hxx.

{
  TLOG_DEBUG(TLVL_WORK_STEPS) << "Stoppping threads...";
 
  // Stop receiving data requests as first thing
  m_data_request_receiver->remove_callback();
  // Stop the other threads
  m_request_handler_impl->stop(args);
  if (m_generate_timesync) {
    while (!m_timesync_thread.get_readiness()) {
      std::this_thread::sleep_for(std::chrono::milliseconds(10));
    }
  }
  if (!m_callback_mode) {
    while (!m_consumer_thread.get_readiness()) {
      std::this_thread::sleep_for(std::chrono::milliseconds(10));
    }
  }
  TLOG_DEBUG(TLVL_WORK_STEPS) << "Flushing latency buffer with occupancy: " << m_latency_buffer_impl->occupancy();
  m_latency_buffer_impl->flush();
  m_raw_processor_impl->stop(args);
  m_raw_processor_impl->reset_last_daq_time();
}

transform_payload()

template<class ReadoutType , class RequestHandlerType , class LatencyBufferType , class RawDataProcessorType , class InputDataType = ReadoutType>

virtual std::vector< RDT > dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::transform_payload ( IDT & original ) const

inlineprotectedvirtual

Definition at line 149 of file DataHandlingModel.hpp.

  {
    return { reinterpret_cast<RDT&>(original) };
  }

Member Data Documentation

m_callback_mode

template<class ReadoutType , class RequestHandlerType , class LatencyBufferType , class RawDataProcessorType , class InputDataType = ReadoutType>

bool dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::m_callback_mode

protected

Definition at line 162 of file DataHandlingModel.hpp.

m_consume_callback

template<class ReadoutType , class RequestHandlerType , class LatencyBufferType , class RawDataProcessorType , class InputDataType = ReadoutType>

std::function<void(RDT&&)> dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::m_consume_callback

Definition at line 132 of file DataHandlingModel.hpp.

m_consumer_thread

template<class ReadoutType , class RequestHandlerType , class LatencyBufferType , class RawDataProcessorType , class InputDataType = ReadoutType>

utilities::ReusableThread dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::m_consumer_thread

protected

Definition at line 188 of file DataHandlingModel.hpp.

m_current_fake_trigger_id

template<class ReadoutType , class RequestHandlerType , class LatencyBufferType , class RawDataProcessorType , class InputDataType = ReadoutType>

int dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::m_current_fake_trigger_id

protected

Definition at line 165 of file DataHandlingModel.hpp.

m_data_request_receiver

template<class ReadoutType , class RequestHandlerType , class LatencyBufferType , class RawDataProcessorType , class InputDataType = ReadoutType>

std::shared_ptr<request_receiver_ct> dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::m_data_request_receiver

protected

Definition at line 199 of file DataHandlingModel.hpp.

m_error_registry

template<class ReadoutType , class RequestHandlerType , class LatencyBufferType , class RawDataProcessorType , class InputDataType = ReadoutType>

std::unique_ptr<FrameErrorRegistry> dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::m_error_registry

protected

Definition at line 224 of file DataHandlingModel.hpp.

m_fake_trigger

template<class ReadoutType , class RequestHandlerType , class LatencyBufferType , class RawDataProcessorType , class InputDataType = ReadoutType>

bool dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::m_fake_trigger

protected

Definition at line 163 of file DataHandlingModel.hpp.

m_generate_timesync

template<class ReadoutType , class RequestHandlerType , class LatencyBufferType , class RawDataProcessorType , class InputDataType = ReadoutType>

bool dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::m_generate_timesync = false

protected

Definition at line 164 of file DataHandlingModel.hpp.

m_latency_buffer_impl

template<class ReadoutType , class RequestHandlerType , class LatencyBufferType , class RawDataProcessorType , class InputDataType = ReadoutType>

std::shared_ptr<LatencyBufferType> dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::m_latency_buffer_impl

protected

Definition at line 214 of file DataHandlingModel.hpp.

m_num_lb_insert_failures

template<class ReadoutType , class RequestHandlerType , class LatencyBufferType , class RawDataProcessorType , class InputDataType = ReadoutType>

std::atomic<num_lb_insert_failures_t> dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::m_num_lb_insert_failures { 0 }

protected

Definition at line 184 of file DataHandlingModel.hpp.

{ 0 };

m_num_payloads

template<class ReadoutType , class RequestHandlerType , class LatencyBufferType , class RawDataProcessorType , class InputDataType = ReadoutType>

std::atomic<num_payload_t> dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::m_num_payloads { 0 }

protected

Definition at line 179 of file DataHandlingModel.hpp.

{ 0 };

m_num_requests

template<class ReadoutType , class RequestHandlerType , class LatencyBufferType , class RawDataProcessorType , class InputDataType = ReadoutType>

std::atomic<num_request_t> dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::m_num_requests { 0 }

protected

Definition at line 181 of file DataHandlingModel.hpp.

{ 0 };

m_pid_of_current_process

template<class ReadoutType , class RequestHandlerType , class LatencyBufferType , class RawDataProcessorType , class InputDataType = ReadoutType>

uint32_t dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::m_pid_of_current_process

protected

Definition at line 211 of file DataHandlingModel.hpp.

m_processing_delay_ticks

template<class ReadoutType , class RequestHandlerType , class LatencyBufferType , class RawDataProcessorType , class InputDataType = ReadoutType>

uint64_t dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::m_processing_delay_ticks

protected

Definition at line 168 of file DataHandlingModel.hpp.

m_raw_data_receiver

template<class ReadoutType , class RequestHandlerType , class LatencyBufferType , class RawDataProcessorType , class InputDataType = ReadoutType>

std::shared_ptr<raw_receiver_ct> dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::m_raw_data_receiver

protected

Definition at line 194 of file DataHandlingModel.hpp.

m_raw_data_receiver_connection_name

template<class ReadoutType , class RequestHandlerType , class LatencyBufferType , class RawDataProcessorType , class InputDataType = ReadoutType>

std::string dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::m_raw_data_receiver_connection_name

protected

Definition at line 195 of file DataHandlingModel.hpp.

m_raw_processor_impl

template<class ReadoutType , class RequestHandlerType , class LatencyBufferType , class RawDataProcessorType , class InputDataType = ReadoutType>

std::shared_ptr<RawDataProcessorType> dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::m_raw_processor_impl

protected

Definition at line 217 of file DataHandlingModel.hpp.

m_raw_receiver_sleep_us

template<class ReadoutType , class RequestHandlerType , class LatencyBufferType , class RawDataProcessorType , class InputDataType = ReadoutType>

std::chrono::microseconds dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::m_raw_receiver_sleep_us

protected

Definition at line 192 of file DataHandlingModel.hpp.

m_raw_receiver_timeout_ms

template<class ReadoutType , class RequestHandlerType , class LatencyBufferType , class RawDataProcessorType , class InputDataType = ReadoutType>

std::chrono::milliseconds dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::m_raw_receiver_timeout_ms

protected

Definition at line 191 of file DataHandlingModel.hpp.

m_rawq_timeout_count

template<class ReadoutType , class RequestHandlerType , class LatencyBufferType , class RawDataProcessorType , class InputDataType = ReadoutType>

std::atomic<rawq_timeout_count_t> dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::m_rawq_timeout_count { 0 }

protected

Definition at line 183 of file DataHandlingModel.hpp.

{ 0 };

m_request_handler_impl

template<class ReadoutType , class RequestHandlerType , class LatencyBufferType , class RawDataProcessorType , class InputDataType = ReadoutType>

std::shared_ptr<RequestHandlerType> dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::m_request_handler_impl

protected

Definition at line 220 of file DataHandlingModel.hpp.

m_request_handler_supports_cutoff_timestamp

template<class ReadoutType , class RequestHandlerType , class LatencyBufferType , class RawDataProcessorType , class InputDataType = ReadoutType>

bool dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::m_request_handler_supports_cutoff_timestamp

protected

Definition at line 221 of file DataHandlingModel.hpp.

m_run_marker

template<class ReadoutType , class RequestHandlerType , class LatencyBufferType , class RawDataProcessorType , class InputDataType = ReadoutType>

std::atomic<bool>& dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::m_run_marker

protected

Definition at line 158 of file DataHandlingModel.hpp.

m_run_number

template<class ReadoutType , class RequestHandlerType , class LatencyBufferType , class RawDataProcessorType , class InputDataType = ReadoutType>

daqdataformats::run_number_t dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::m_run_number

protected

Definition at line 167 of file DataHandlingModel.hpp.

m_sourceid

template<class ReadoutType , class RequestHandlerType , class LatencyBufferType , class RawDataProcessorType , class InputDataType = ReadoutType>

daqdataformats::SourceID dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::m_sourceid

protected

Definition at line 166 of file DataHandlingModel.hpp.

m_stats_packet_count

template<class ReadoutType , class RequestHandlerType , class LatencyBufferType , class RawDataProcessorType , class InputDataType = ReadoutType>

std::atomic<int> dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::m_stats_packet_count { 0 }

protected

Definition at line 185 of file DataHandlingModel.hpp.

{ 0 };

m_sum_payloads

template<class ReadoutType , class RequestHandlerType , class LatencyBufferType , class RawDataProcessorType , class InputDataType = ReadoutType>

std::atomic<sum_payload_t> dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::m_sum_payloads { 0 }

protected

Definition at line 180 of file DataHandlingModel.hpp.

{ 0 };

m_sum_requests

template<class ReadoutType , class RequestHandlerType , class LatencyBufferType , class RawDataProcessorType , class InputDataType = ReadoutType>

std::atomic<sum_request_t> dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::m_sum_requests { 0 }

protected

Definition at line 182 of file DataHandlingModel.hpp.

{ 0 };

m_t0

template<class ReadoutType , class RequestHandlerType , class LatencyBufferType , class RawDataProcessorType , class InputDataType = ReadoutType>

std::chrono::time_point<std::chrono::high_resolution_clock> dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::m_t0

protected

Definition at line 227 of file DataHandlingModel.hpp.

m_timesync_connection_name

template<class ReadoutType , class RequestHandlerType , class LatencyBufferType , class RawDataProcessorType , class InputDataType = ReadoutType>

std::string dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::m_timesync_connection_name

protected

Definition at line 210 of file DataHandlingModel.hpp.

m_timesync_sender

template<class ReadoutType , class RequestHandlerType , class LatencyBufferType , class RawDataProcessorType , class InputDataType = ReadoutType>

std::shared_ptr<timesync_sender_ct> dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::m_timesync_sender

protected

Definition at line 208 of file DataHandlingModel.hpp.

m_timesync_thread

template<class ReadoutType , class RequestHandlerType , class LatencyBufferType , class RawDataProcessorType , class InputDataType = ReadoutType>

utilities::ReusableThread dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::m_timesync_thread

protected

Definition at line 209 of file DataHandlingModel.hpp.

ms

template<class ReadoutType , class RequestHandlerType , class LatencyBufferType , class RawDataProcessorType , class InputDataType = ReadoutType>

timestamp_t dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::ms = 1000 * us

inlinestaticconstexpr

Definition at line 77 of file DataHandlingModel.hpp.

ns

template<class ReadoutType , class RequestHandlerType , class LatencyBufferType , class RawDataProcessorType , class InputDataType = ReadoutType>

timestamp_t dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::ns = 1

inlinestaticconstexpr

Definition at line 75 of file DataHandlingModel.hpp.

s

template<class ReadoutType , class RequestHandlerType , class LatencyBufferType , class RawDataProcessorType , class InputDataType = ReadoutType>

timestamp_t dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::s = 1000 * ms

inlinestaticconstexpr

Definition at line 78 of file DataHandlingModel.hpp.

us

template<class ReadoutType , class RequestHandlerType , class LatencyBufferType , class RawDataProcessorType , class InputDataType = ReadoutType>

timestamp_t dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::us = 1000 * ns

inlinestaticconstexpr

Definition at line 76 of file DataHandlingModel.hpp.

---

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

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