#include <DataHandlingModel.hpp>

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

[legend]

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

[legend]

Classes
class	PostprocessScheduleAlgorithm

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 appfwk::DAQModule::CommandData_t &args)

void	scrap (const appfwk::DAQModule::CommandData_t &args)

void	start (const appfwk::DAQModule::CommandData_t &args)

void	stop (const appfwk::DAQModule::CommandData_t &args)

void	record (const appfwk::DAQModule::CommandData_t &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	invoke_postprocess_schedule_timeout_policy () 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_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

const appmodel::DataMoveCallbackConf *	m_raw_data_callback_conf

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

utilities::ReusableThread	m_postprocess_scheduler_thread

folly::coro::Baton	m_baton

std::unique_ptr< folly::Timekeeper >	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 69 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 77 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 75 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 335 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 341 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 336 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 342 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 338 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 359 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 340 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 73 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 365 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 74 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 76 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 337 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 339 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 80 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 374 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 87 of file DataHandlingModel.hpp.

    : m_run_marker(run_marker)
    , 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)
  {
  }

◆ ~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 appfwk::DAQModule::CommandData_t & args )

virtual

Implements dunedaq::datahandlinglibs::DataHandlingConcept.

Definition at line 104 of file DataHandlingModel.hxx.

{
  // Register callbacks if operating in that mode.
  if (m_raw_data_callback_conf != nullptr) {
    // 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_callback_conf, 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 246 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 198 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, m_sourceid, 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());
  ri.set_newest_timestamp(m_raw_processor_impl->get_last_daq_time());
  ri.set_oldest_timestamp(m_request_handler_impl->get_oldest_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 16 of file DataHandlingModel.hxx.

{
  // Check if a callback is defined (TPs use IOManager Queues instead)
  m_raw_data_callback_conf = mcfg->get_raw_data_callback();
  if (m_raw_data_callback_conf != nullptr) {
    m_raw_data_callback_conf = mcfg->get_raw_data_callback();
    TLOG_DEBUG(TLVL_WORK_STEPS) << "DataHandlingModel operating in callback mode.";
  } else {
    TLOG_DEBUG(TLVL_WORK_STEPS) << "DataHandlingModel operating in message polling mode.";
  }
 
  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();
 
        if (m_raw_data_callback_conf == nullptr) {
          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_raw_data_callback_conf == nullptr && m_raw_data_receiver == nullptr) {
    ers::error(ConfigurationError(ERS_HERE, m_sourceid, "No callback configuration, 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();
 
  if (m_processing_delay_ticks) {
    if constexpr (!SupportsDelayedPostprocessing<LBT>) {
      ers::error(ConfigurationError(
        ERS_HERE,
        m_sourceid,
        "Delayed postprocessing (post_processing_delay_ticks > 0) requires a sorted buffer (SkipList). "
        "Queue buffers (FixedRateQueue, BinarySearchQueue) expect in-order data and must use "
        "post_processing_delay_ticks = 0."));
    }
  }
 
  // 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);
}

◆ invoke_postprocess_schedule_timeout_policy()

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

virtual void dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::invoke_postprocess_schedule_timeout_policy ( ) const

inlineprotectedvirtual

Definition at line 312 of file DataHandlingModel.hpp.

  {
    return; // No-op for this class
  }

◆ 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 325 of file DataHandlingModel.hxx.

{
 
  TLOG_DEBUG(TLVL_WORK_STEPS) << "Postprocess schedule coroutine started...";
  TLOG() << "***** Starting post-process coroutine with timout " << m_post_processing_delay_max_wait << " *****";
 
  PostprocessScheduleAlgorithm sched_algo{ *m_latency_buffer_impl,
                                           *m_raw_processor_impl,
                                           m_processing_delay_ticks,
                                           m_post_processing_delay_min_wait,
                                           m_post_processing_delay_max_wait };
 
  const auto wait_data = [this]() -> folly::coro::Task<void> {
    // folly::coro::timeout cancels the task on timeout.
    // Baton is not cancellable, so we attach a callback to resume the coroutine.
    auto token = co_await folly::coro::co_current_cancellation_token;
    folly::CancellationCallback cb(token, [this] { m_baton.post(); });
    co_await m_baton; // Wait data
  };
 
  while (m_run_marker.load()) {
    bool timeout = false;
 
    if (m_post_processing_delay_max_wait > 0) {
      try {
        co_await folly::coro::timeout(
          wait_data(), std::chrono::milliseconds{ m_post_processing_delay_max_wait }, m_timekeeper.get());
 
      } catch (const folly::FutureTimeout&) {
        timeout = true;
        ++m_num_post_processing_delay_max_waits;
      }
    } else {
      co_await m_baton;
    }
 
    m_baton.reset();
 
    if (auto processed = sched_algo.run(timeout); processed > 0) {
      m_num_payloads += processed;
      m_sum_payloads += processed;
      m_stats_packet_count += processed;
    }
  }
}

◆ 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 253 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_sourceid,
                                            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++;
    return;
  }
 
  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 appfwk::DAQModule::CommandData_t & 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 295 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 288 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 373 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_id = m_sourceid.id;
        TLOG_DEBUG(TLVL_TIME_SYNCS) << "New timesync: daq=" << timesyncmsg.daq_time
                                    << " wall=" << timesyncmsg.system_time << " run=" << timesyncmsg.run_number
                                    << " seqno=" << timesyncmsg.sequence_number
                                    << " source_id=" << timesyncmsg.source_id;
        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 appfwk::DAQModule::CommandData_t & 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 appfwk::DAQModule::CommandData_t & args )

virtual

Implements dunedaq::datahandlinglibs::DataHandlingConcept.

Definition at line 130 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_raw_data_callback_conf == nullptr) {
    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 appfwk::DAQModule::CommandData_t & args )

virtual

Implements dunedaq::datahandlinglibs::DataHandlingConcept.

Definition at line 166 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_raw_data_callback_conf == nullptr) {
    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 232 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 306 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 381 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 354 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 327 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 366 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 395 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 325 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 326 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 385 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 349 of file DataHandlingModel.hpp.

349{ 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 344 of file DataHandlingModel.hpp.

344{ 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 350 of file DataHandlingModel.hpp.

350{ 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 346 of file DataHandlingModel.hpp.

346{ 0 };

◆ 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 332 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 331 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 380 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 330 of file DataHandlingModel.hpp.

◆ m_raw_data_callback_conf

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

const appmodel::DataMoveCallbackConf* dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::m_raw_data_callback_conf

protected

Definition at line 362 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 360 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 361 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 388 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 358 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 357 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 348 of file DataHandlingModel.hpp.

348{ 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 391 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 392 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 321 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 329 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 328 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 351 of file DataHandlingModel.hpp.

351{ 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 345 of file DataHandlingModel.hpp.

345{ 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 347 of file DataHandlingModel.hpp.

347{ 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 398 of file DataHandlingModel.hpp.

◆ m_timekeeper

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

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

protected

Definition at line 382 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 377 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 375 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 376 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 83 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 81 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 84 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 82 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

Classes

Public Types

Public Member Functions

Public Attributes

Static Public Attributes

Protected Types

Protected Member Functions

Protected Attributes

Additional Inherited Members

Detailed Description

Member Typedef Documentation

◆ IDT

◆ LBT

◆ metric_t

◆ num_lb_insert_failures_t

◆ num_payload_t

◆ num_post_processing_delay_max_waits_t

◆ num_request_t

◆ raw_receiver_ct

◆ rawq_timeout_count_t

◆ RDT

◆ request_receiver_ct

◆ RHT

◆ RPT

◆ sum_payload_t

◆ sum_request_t

◆ timestamp_t

◆ timesync_sender_ct

Constructor & Destructor Documentation

◆ DataHandlingModel()

◆ ~DataHandlingModel()

Member Function Documentation

◆ conf()

◆ consume_callback()

◆ dispatch_requests()

◆ generate_opmon_data()

◆ init()

◆ invoke_postprocess_schedule_timeout_policy()

◆ postprocess_schedule()

◆ process_item()

◆ record()

◆ run_consume()

◆ run_postprocess_scheduler()

◆ run_timesync()

◆ scrap()

◆ start()

◆ stop()

◆ transform_and_process()

◆ transform_payload()

Member Data Documentation

◆ m_baton

◆ m_consume_callback

◆ m_consumer_thread

◆ m_current_fake_trigger_id

◆ m_data_request_receiver

◆ m_error_registry

◆ m_fake_trigger

◆ m_generate_timesync

◆ m_latency_buffer_impl

◆ m_num_lb_insert_failures

◆ m_num_payloads

◆ m_num_post_processing_delay_max_waits

◆ m_num_requests

◆ m_post_processing_delay_max_wait

◆ m_post_processing_delay_min_wait

◆ m_postprocess_scheduler_thread

◆ m_processing_delay_ticks

◆ m_raw_data_callback_conf

◆ m_raw_data_receiver

◆ m_raw_data_receiver_connection_name

◆ m_raw_processor_impl

◆ m_raw_receiver_sleep_us

◆ m_raw_receiver_timeout_ms

◆ m_rawq_timeout_count

◆ m_request_handler_impl

◆ m_request_handler_supports_cutoff_timestamp

◆ m_run_marker

◆ m_run_number

◆ m_sourceid

◆ m_stats_packet_count