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
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(IDT &&)>	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	num_post_processing_delay_max_waits_t = std::remove_const<std::invoke_result<decltype(&metric_t::num_post_processing_delay_max_waits),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	transform_and_process (IDT &&payload)
void	consume_callback (IDT &&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	run_postprocess_scheduler ()
folly::coro::Task< void >	postprocess_schedule ()
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
uint64_t	m_post_processing_delay_min_wait
uint64_t	m_post_processing_delay_max_wait
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< num_post_processing_delay_max_waits_t >	m_num_post_processing_delay_max_waits { 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
utilities::ReusableThread	m_postprocess_scheduler_thread
folly::coro::Baton	m_baton
std::unique_ptr< folly::ThreadWheelTimekeeper >	m_timekeeper
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 67 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 75 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 73 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 186 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 192 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 187 of file DataHandlingModel.hpp.

num_post_processing_delay_max_waits_t

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

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

protected

Definition at line 193 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 189 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 210 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 191 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 71 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 215 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 72 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 74 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 188 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 190 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 78 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 224 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 85 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 103 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_callback, this, std::placeholders::_1);
 
    // Register callback
    auto dmcbr = DataMoveCallbackRegistry::get();
    dmcbr->register_callback<IDT>(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);
  }
  if (m_processing_delay_ticks) {
    m_postprocess_scheduler_thread.set_name("pprocsched", m_sourceid.id);
    m_timekeeper = std::make_unique<folly::ThreadWheelTimekeeper>();
  }  
}

consume_callback()

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

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

protected

Definition at line 240 of file DataHandlingModel.hxx.

                                                                          {
  transform_and_process(std::move(payload));
}

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 458 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 196 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_num_post_processing_delay_max_waits(m_num_post_processing_delay_max_waits.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 13 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();
  m_post_processing_delay_min_wait = mcfg->get_module_configuration()->get_post_processing_delay_min_wait();
  m_post_processing_delay_max_wait = mcfg->get_module_configuration()->get_post_processing_delay_max_wait();
  
 
  // 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);
}

postprocess_schedule()

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

folly::coro::Task< void > dunedaq::datahandlinglibs::DataHandlingModel< RDT, RHT, LBT, RPT, IDT >::postprocess_schedule ( )

protected

Definition at line 311 of file DataHandlingModel.hxx.

                                                                 {  
 
  TLOG_DEBUG(TLVL_WORK_STEPS) << "Postprocess schedule coroutine started...";
  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;
 
  // 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  
  while (m_run_marker.load()) {
    try {
      co_await folly::coro::timeout(
        m_baton.operator co_await(),
        std::chrono::milliseconds{m_post_processing_delay_max_wait},
        m_timekeeper.get());
      m_baton.reset();
    } catch (const folly::FutureTimeout&) {
      ++m_num_post_processing_delay_max_waits;
    }
 
    if (m_latency_buffer_impl->occupancy() == 0) {
      continue;
    }
 
    now = std::chrono::system_clock::now();
    milliseconds = std::chrono::duration_cast<std::chrono::milliseconds>(now - last_post_proc_time);
 
    if (milliseconds.count() <= m_post_processing_delay_min_wait) {
      continue;
    }
 
    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;
      }
    }
  }
}

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 246 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 insert failed! (Payload timestamp=" << payload.get_timestamp() << ")";
    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;
  } else {
    m_baton.post();
  }
}

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 124 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 281 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;
  m_num_post_processing_delay_max_waits = 0;
 
  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& payload = opt_payload.value();
      transform_and_process(std::move(payload));
    } 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);
    }
  }
  TLOG_DEBUG(TLVL_WORK_STEPS) << "Consumer thread joins... ";
}

run_postprocess_scheduler()

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

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

protected

Definition at line 274 of file DataHandlingModel.hxx.

{
  folly::coro::blockingWait(postprocess_schedule());
}

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 377 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 110 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 129 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_num_post_processing_delay_max_waits = 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);
  }
  if (m_processing_delay_ticks) {
    m_postprocess_scheduler_thread.set_work(&DataHandlingModel<RDT, RHT, LBT, RPT, IDT>::run_postprocess_scheduler, 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 164 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));
    }
  }
  if (m_processing_delay_ticks) {
    m_baton.post(); // In case the coroutine is still waiting when the consumer has stopped
    while (!m_postprocess_scheduler_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_and_process()

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

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

protected

Definition at line 227 of file DataHandlingModel.hxx.

                                                                               {
  if constexpr (std::is_same_v<IDT, RDT>) {
    process_item(std::move(payload));
  } else {
    auto transformed = transform_payload(payload);
    for (auto& i : transformed) {
      process_item(std::move(i));
    }   
  }  
}

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 162 of file DataHandlingModel.hpp.

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

Member Data Documentation

m_baton

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

folly::coro::Baton dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::m_baton

protected

Definition at line 232 of file DataHandlingModel.hpp.

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 175 of file DataHandlingModel.hpp.

m_consume_callback

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

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

Definition at line 133 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 205 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 178 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 216 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 246 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 176 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 177 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 236 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 200 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 195 of file DataHandlingModel.hpp.

{ 0 };

m_num_post_processing_delay_max_waits

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

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

protected

Definition at line 201 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 197 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 228 of file DataHandlingModel.hpp.

m_post_processing_delay_max_wait

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

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

protected

Definition at line 183 of file DataHandlingModel.hpp.

m_post_processing_delay_min_wait

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

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

protected

Definition at line 182 of file DataHandlingModel.hpp.

m_postprocess_scheduler_thread

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

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

protected

Definition at line 231 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 181 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 211 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 212 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 239 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 209 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 208 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 199 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 242 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 243 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 171 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 180 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 179 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 202 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 196 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 198 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 249 of file DataHandlingModel.hpp.

m_timekeeper

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

std::unique_ptr<folly::ThreadWheelTimekeeper> dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::m_timekeeper

protected

Definition at line 233 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 227 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 225 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 226 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 81 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 79 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 82 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 80 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