#include <TaskRawDataProcessorModel.hpp>

Inheritance diagram for dunedaq::datahandlinglibs::TaskRawDataProcessorModel< ReadoutType >:

Collaboration diagram for dunedaq::datahandlinglibs::TaskRawDataProcessorModel< ReadoutType >:

Public Member Functions
	TaskRawDataProcessorModel (std::unique_ptr< FrameErrorRegistry > &error_registry, bool post_processing_enabled)

	~TaskRawDataProcessorModel ()

void	conf (const appmodel::DataHandlerModule *conf) override
	Set the emulator mode, if active, timestamps of processed packets are overwritten with new ones.

void	scrap (const appfwk::DAQModule::CommandData_t &) override
	Unconfigure.

void	start (const appfwk::DAQModule::CommandData_t &) override
	Start operation.

void	stop (const appfwk::DAQModule::CommandData_t &) override
	Stop operation.

void	reset_last_daq_time ()

std::uint64_t	get_last_daq_time () override
	Get newest timestamp of last seen packet.

void	preprocess_item (ReadoutType *item) override
	Preprocess one element.

void	postprocess_item (const ReadoutType *item) override
	Postprocess one element.

void	invoke_postprocess_schedule_timeout_policy (std::uint64_t accumilated_timeout_ticks) override
	Handle postprocess timeout event.

template<typename Task >
void	add_preprocess_task (Task &&task)

template<typename Task >
void	add_postprocess_task (Task &&task)

void	invoke_all_preprocess_functions (ReadoutType *item)

void	launch_all_preprocess_functions (ReadoutType *item)

Public Member Functions inherited from dunedaq::datahandlinglibs::RawDataProcessorConcept< ReadoutType >
	RawDataProcessorConcept ()

virtual	~RawDataProcessorConcept ()

	RawDataProcessorConcept (const RawDataProcessorConcept &)=delete
	RawDataProcessorConcept is not copy-constructible.

RawDataProcessorConcept &	operator= (const RawDataProcessorConcept &)=delete
	RawDataProcessorConcept is not copy-assginable.

	RawDataProcessorConcept (RawDataProcessorConcept &&)=delete
	RawDataProcessorConcept is not move-constructible.

RawDataProcessorConcept &	operator= (RawDataProcessorConcept &&)=delete
	RawDataProcessorConcept is not move-assignable.

Public Member Functions inherited from dunedaq::opmonlib::MonitorableObject
	MonitorableObject (const MonitorableObject &)=delete

MonitorableObject &	operator= (const MonitorableObject &)=delete

	MonitorableObject (MonitorableObject &&)=delete

MonitorableObject &	operator= (MonitorableObject &&)=delete

virtual	~MonitorableObject ()=default

auto	get_opmon_id () const noexcept

auto	get_opmon_level () const noexcept

Protected Member Functions
virtual void	generate_opmon_data () override

void	run_post_processing_thread (std::function< void(const ReadoutType )> &function, folly::ProducerConsumerQueue< const ReadoutType > &queue)

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::vector< std::function< void(ReadoutType *)> >	m_preprocess_functions

std::unique_ptr< FrameErrorRegistry > &	m_error_registry

bool	m_post_processing_enabled

std::atomic< bool >	m_run_marker { false }

std::vector< std::function< void(const ReadoutType *)> >	m_post_process_functions

std::vector< std::unique_ptr< folly::ProducerConsumerQueue< const ReadoutType * > > >	m_items_to_postprocess_queues

std::vector< std::unique_ptr< utilities::ReusableThread > >	m_post_process_threads

size_t	m_postprocess_queue_sizes

daqdataformats::SourceID	m_sourceid

std::atomic< uint64_t >	m_last_processed_daq_ts { 0 }

Additional Inherited Members
Public Types inherited from dunedaq::opmonlib::MonitorableObject
using	NodePtr = std::weak_ptr<MonitorableObject>

using	NewNodePtr = std::shared_ptr<MonitorableObject>

using	ElementId = std::string

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

Detailed Description

template<class ReadoutType>
class dunedaq::datahandlinglibs::TaskRawDataProcessorModel< ReadoutType >

Definition at line 46 of file TaskRawDataProcessorModel.hpp.

Constructor & Destructor Documentation

◆ TaskRawDataProcessorModel()

template<class ReadoutType >

dunedaq::datahandlinglibs::TaskRawDataProcessorModel< ReadoutType >::TaskRawDataProcessorModel	(	std::unique_ptr< FrameErrorRegistry > &	error_registry,
		bool	post_processing_enabled )

inlineexplicit

Definition at line 50 of file TaskRawDataProcessorModel.hpp.

    : RawDataProcessorConcept<ReadoutType>()
    , m_error_registry(error_registry)
    , m_post_processing_enabled(post_processing_enabled)
  {}

◆ ~TaskRawDataProcessorModel()

template<class ReadoutType >

dunedaq::datahandlinglibs::TaskRawDataProcessorModel< ReadoutType >::~TaskRawDataProcessorModel ( )

inline

Definition at line 57 of file TaskRawDataProcessorModel.hpp.

57{}

Member Function Documentation

◆ add_postprocess_task()

template<class ReadoutType >

template<typename Task >

void dunedaq::datahandlinglibs::TaskRawDataProcessorModel< ReadoutType >::add_postprocess_task ( Task && task )

Definition at line 81 of file TaskRawDataProcessorModel.hxx.

{
  m_post_process_threads.emplace_back(std::make_unique<utilities::ReusableThread>(0));
  m_post_process_functions.push_back(std::forward<Task>(task));
}

◆ add_preprocess_task()

template<class ReadoutType >

template<typename Task >

void dunedaq::datahandlinglibs::TaskRawDataProcessorModel< ReadoutType >::add_preprocess_task ( Task && task )

Definition at line 73 of file TaskRawDataProcessorModel.hxx.

{
  m_preprocess_functions.push_back(std::forward<Task>(task));
}

◆ conf()

template<class ReadoutType >

void dunedaq::datahandlinglibs::TaskRawDataProcessorModel< ReadoutType >::conf ( const appmodel::DataHandlerModule * conf )

overridevirtual

Set the emulator mode, if active, timestamps of processed packets are overwritten with new ones.

Implements dunedaq::datahandlinglibs::RawDataProcessorConcept< ReadoutType >.

Reimplemented in dunedaq::fdreadoutlibs::TDEFrameProcessor, dunedaq::fdreadoutlibs::TPCEthFrameProcessor< ReadoutTypeAdapter >, dunedaq::fdreadoutlibs::TPCEthFrameProcessor< types::DUNEWIBEthTypeAdapter >, dunedaq::fdreadoutlibs::TPCEthFrameProcessor< types::TDEEthTypeAdapter >, dunedaq::trigger::TCProcessor, and dunedaq::trigger::TPProcessor.

Definition at line 8 of file TaskRawDataProcessorModel.hxx.

{
  auto cfg = conf->get_module_configuration()->get_data_processor();
  m_postprocess_queue_sizes = cfg->get_queue_sizes();
  m_sourceid.id = conf->get_source_id();
  m_sourceid.subsystem = ReadoutType::subsystem;
 
  for (size_t i = 0; i < m_post_process_functions.size(); ++i) {
    m_items_to_postprocess_queues.push_back(
      std::make_unique<folly::ProducerConsumerQueue<const ReadoutType*>>(m_postprocess_queue_sizes));
    m_post_process_threads[i]->set_name(cfg->get_thread_names_prefix() + std::to_string(i), m_sourceid.id);
  }
}

◆ generate_opmon_data()

template<class ReadoutType >

void dunedaq::datahandlinglibs::TaskRawDataProcessorModel< ReadoutType >::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.

Reimplemented in dunedaq::fdreadoutlibs::DAPHNEEthFrameProcessor, dunedaq::fdreadoutlibs::DAPHNEFrameProcessor, dunedaq::fdreadoutlibs::TPCEthFrameProcessor< ReadoutTypeAdapter >, dunedaq::fdreadoutlibs::TPCEthFrameProcessor< types::DUNEWIBEthTypeAdapter >, dunedaq::fdreadoutlibs::TPCEthFrameProcessor< types::TDEEthTypeAdapter >, dunedaq::trigger::TAProcessor, dunedaq::trigger::TCProcessor, and dunedaq::trigger::TPProcessor.

Definition at line 123 of file TaskRawDataProcessorModel.hxx.

{
   for (size_t i = 0; i < m_items_to_postprocess_queues.size(); ++i) {
    opmon::DataProcessorInfo info;
    info.set_elements_queued(m_items_to_postprocess_queues[i]->sizeGuess());
    this->publish(std::move(info), {{"post_processor_id", std::to_string(i)}});
  }
}

◆ get_last_daq_time()

template<class ReadoutType >

std::uint64_t dunedaq::datahandlinglibs::TaskRawDataProcessorModel< ReadoutType >::get_last_daq_time ( )

inlineoverridevirtual

Get newest timestamp of last seen packet.

Implements dunedaq::datahandlinglibs::RawDataProcessorConcept< ReadoutType >.

Definition at line 75 of file TaskRawDataProcessorModel.hpp.

75{ return m_last_processed_daq_ts.load(); } // NOLINT(build/unsigned)

dunedaq::datahandlinglibs::TaskRawDataProcessorModel::m_last_processed_daq_ts

std::atomic< uint64_t > m_last_processed_daq_ts

Definition TaskRawDataProcessorModel.hpp:126

◆ invoke_all_preprocess_functions()

template<class ReadoutType >

void dunedaq::datahandlinglibs::TaskRawDataProcessorModel< ReadoutType >::invoke_all_preprocess_functions ( ReadoutType * item )

Definition at line 89 of file TaskRawDataProcessorModel.hxx.

{
  for (auto&& task : m_preprocess_functions) {
    task(item);
  }
}

◆ invoke_postprocess_schedule_timeout_policy()

template<class ReadoutType >

void dunedaq::datahandlinglibs::TaskRawDataProcessorModel< ReadoutType >::invoke_postprocess_schedule_timeout_policy ( std::uint64_t accumulated_timeout_ticks )

inlineoverridevirtual

Handle postprocess timeout event.

Implements dunedaq::datahandlinglibs::RawDataProcessorConcept< ReadoutType >.

Definition at line 84 of file TaskRawDataProcessorModel.hpp.

84{}

◆ launch_all_preprocess_functions()

template<class ReadoutType >

void dunedaq::datahandlinglibs::TaskRawDataProcessorModel< ReadoutType >::launch_all_preprocess_functions ( ReadoutType * item )

Definition at line 98 of file TaskRawDataProcessorModel.hxx.

{
  for (auto&& task : m_preprocess_functions) {
    auto fut = std::async(std::launch::async, task, item);
  }
}

◆ postprocess_item()

template<class ReadoutType >

void dunedaq::datahandlinglibs::TaskRawDataProcessorModel< ReadoutType >::postprocess_item ( const ReadoutType * item )

overridevirtual

Postprocess one element.

Implements dunedaq::datahandlinglibs::RawDataProcessorConcept< ReadoutType >.

Definition at line 61 of file TaskRawDataProcessorModel.hxx.

{
  for (size_t i = 0; i < m_items_to_postprocess_queues.size(); ++i) {
    if (!m_items_to_postprocess_queues[i]->write(item)) {
      ers::warning(PostprocessingNotKeepingUp(ERS_HERE, m_sourceid, i));
    }
  }
}

◆ preprocess_item()

template<class ReadoutType >

void dunedaq::datahandlinglibs::TaskRawDataProcessorModel< ReadoutType >::preprocess_item ( ReadoutType * item )

inlineoverridevirtual

Preprocess one element.

Implements dunedaq::datahandlinglibs::RawDataProcessorConcept< ReadoutType >.

Definition at line 78 of file TaskRawDataProcessorModel.hpp.

78{ invoke_all_preprocess_functions(item); }

dunedaq::datahandlinglibs::TaskRawDataProcessorModel::invoke_all_preprocess_functions

void invoke_all_preprocess_functions(ReadoutType *item)

Definition TaskRawDataProcessorModel.hxx:89

◆ reset_last_daq_time()

template<class ReadoutType >

void dunedaq::datahandlinglibs::TaskRawDataProcessorModel< ReadoutType >::reset_last_daq_time ( )

inline

Definition at line 72 of file TaskRawDataProcessorModel.hpp.

72{ m_last_processed_daq_ts.store(0); }

◆ run_post_processing_thread()

template<class ReadoutType >

void dunedaq::datahandlinglibs::TaskRawDataProcessorModel< ReadoutType >::run_post_processing_thread	(	std::function< void(const ReadoutType *)> &	function,
		folly::ProducerConsumerQueue< const ReadoutType * > &	queue )

protected

Definition at line 107 of file TaskRawDataProcessorModel.hxx.

{
  while (m_run_marker.load() || queue.sizeGuess() > 0) {
    const ReadoutType* element;
    if (queue.read(element)) {
      function(element);
    } else {
      std::this_thread::sleep_for(std::chrono::microseconds(50));
    }
  }
}

◆ scrap()

template<class ReadoutType >

void dunedaq::datahandlinglibs::TaskRawDataProcessorModel< ReadoutType >::scrap ( const appfwk::DAQModule::CommandData_t & args )

overridevirtual

Unconfigure.

Implements dunedaq::datahandlinglibs::RawDataProcessorConcept< ReadoutType >.

Reimplemented in dunedaq::fdreadoutlibs::TPCEthFrameProcessor< ReadoutTypeAdapter >, dunedaq::fdreadoutlibs::TPCEthFrameProcessor< types::DUNEWIBEthTypeAdapter >, dunedaq::fdreadoutlibs::TPCEthFrameProcessor< types::TDEEthTypeAdapter >, dunedaq::trigger::TCProcessor, and dunedaq::trigger::TPProcessor.

Definition at line 24 of file TaskRawDataProcessorModel.hxx.

{
  m_items_to_postprocess_queues.clear();
  m_post_process_threads.clear();
  m_post_process_functions.clear();
  m_preprocess_functions.clear();
}

◆ start()

template<class ReadoutType >

void dunedaq::datahandlinglibs::TaskRawDataProcessorModel< ReadoutType >::start ( const appfwk::DAQModule::CommandData_t & args )

overridevirtual

Start operation.

Implements dunedaq::datahandlinglibs::RawDataProcessorConcept< ReadoutType >.

Reimplemented in dunedaq::fdreadoutlibs::TPCEthFrameProcessor< ReadoutTypeAdapter >, dunedaq::fdreadoutlibs::TPCEthFrameProcessor< types::DUNEWIBEthTypeAdapter >, dunedaq::fdreadoutlibs::TPCEthFrameProcessor< types::TDEEthTypeAdapter >, dunedaq::trigger::TCProcessor, and dunedaq::trigger::TPProcessor.

Definition at line 34 of file TaskRawDataProcessorModel.hxx.

{
  // m_last_processed_daq_ts =
  // std::chrono::duration_cast<std::chrono::nanoseconds>(std::chrono::system_clock::now().time_since_epoch()).count();
  m_run_marker.store(true);
  for (size_t i = 0; i < m_post_process_threads.size(); ++i) {
    m_post_process_threads[i]->set_work(&TaskRawDataProcessorModel<ReadoutType>::run_post_processing_thread,
                                        this,
                                        std::ref(m_post_process_functions[i]),
                                        std::ref(*m_items_to_postprocess_queues[i]));
  }
}

◆ stop()

template<class ReadoutType >

void dunedaq::datahandlinglibs::TaskRawDataProcessorModel< ReadoutType >::stop ( const appfwk::DAQModule::CommandData_t & args )

overridevirtual

Stop operation.

Implements dunedaq::datahandlinglibs::RawDataProcessorConcept< ReadoutType >.

Reimplemented in dunedaq::fdreadoutlibs::TPCEthFrameProcessor< ReadoutTypeAdapter >, dunedaq::fdreadoutlibs::TPCEthFrameProcessor< types::DUNEWIBEthTypeAdapter >, dunedaq::fdreadoutlibs::TPCEthFrameProcessor< types::TDEEthTypeAdapter >, dunedaq::trigger::TCProcessor, and dunedaq::trigger::TPProcessor.

Definition at line 49 of file TaskRawDataProcessorModel.hxx.

{
  m_run_marker.store(false);
  for (auto& thread : m_post_process_threads) {
    while (!thread->get_readiness()) {
      std::this_thread::sleep_for(std::chrono::milliseconds(50));
    }
  }
}

Member Data Documentation

◆ m_error_registry

template<class ReadoutType >

std::unique_ptr<FrameErrorRegistry>& dunedaq::datahandlinglibs::TaskRawDataProcessorModel< ReadoutType >::m_error_registry

protected

Definition at line 106 of file TaskRawDataProcessorModel.hpp.

◆ m_items_to_postprocess_queues

template<class ReadoutType >

std::vector<std::unique_ptr<folly::ProducerConsumerQueue<const ReadoutType*> > > dunedaq::datahandlinglibs::TaskRawDataProcessorModel< ReadoutType >::m_items_to_postprocess_queues

protected

Definition at line 118 of file TaskRawDataProcessorModel.hpp.

◆ m_last_processed_daq_ts

template<class ReadoutType >

std::atomic<uint64_t> dunedaq::datahandlinglibs::TaskRawDataProcessorModel< ReadoutType >::m_last_processed_daq_ts { 0 }

protected

Definition at line 126 of file TaskRawDataProcessorModel.hpp.

126{ 0 }; // NOLINT(build/unsigned)

◆ m_post_process_functions

template<class ReadoutType >

std::vector<std::function<void(const ReadoutType*)> > dunedaq::datahandlinglibs::TaskRawDataProcessorModel< ReadoutType >::m_post_process_functions

protected

Definition at line 117 of file TaskRawDataProcessorModel.hpp.

◆ m_post_process_threads

template<class ReadoutType >

std::vector<std::unique_ptr<utilities::ReusableThread> > dunedaq::datahandlinglibs::TaskRawDataProcessorModel< ReadoutType >::m_post_process_threads

protected

Definition at line 119 of file TaskRawDataProcessorModel.hpp.

◆ m_post_processing_enabled

template<class ReadoutType >

bool dunedaq::datahandlinglibs::TaskRawDataProcessorModel< ReadoutType >::m_post_processing_enabled

protected

Definition at line 111 of file TaskRawDataProcessorModel.hpp.

◆ m_postprocess_queue_sizes

template<class ReadoutType >

size_t dunedaq::datahandlinglibs::TaskRawDataProcessorModel< ReadoutType >::m_postprocess_queue_sizes

protected

Definition at line 122 of file TaskRawDataProcessorModel.hpp.

◆ m_preprocess_functions

template<class ReadoutType >

std::vector<std::function<void(ReadoutType*)> > dunedaq::datahandlinglibs::TaskRawDataProcessorModel< ReadoutType >::m_preprocess_functions

protected

Definition at line 105 of file TaskRawDataProcessorModel.hpp.

◆ m_run_marker

template<class ReadoutType >

std::atomic<bool> dunedaq::datahandlinglibs::TaskRawDataProcessorModel< ReadoutType >::m_run_marker { false }

protected

Definition at line 114 of file TaskRawDataProcessorModel.hpp.

114{ false };

◆ m_sourceid

template<class ReadoutType >

daqdataformats::SourceID dunedaq::datahandlinglibs::TaskRawDataProcessorModel< ReadoutType >::m_sourceid

protected

Definition at line 124 of file TaskRawDataProcessorModel.hpp.

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

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

Public Member Functions

Protected Member Functions

Protected Attributes

Additional Inherited Members

Detailed Description

Constructor & Destructor Documentation

◆ TaskRawDataProcessorModel()

◆ ~TaskRawDataProcessorModel()

Member Function Documentation

◆ add_postprocess_task()

◆ add_preprocess_task()

◆ conf()

◆ generate_opmon_data()

◆ get_last_daq_time()

◆ invoke_all_preprocess_functions()

◆ invoke_postprocess_schedule_timeout_policy()

◆ launch_all_preprocess_functions()

◆ postprocess_item()

◆ preprocess_item()

◆ reset_last_daq_time()

◆ run_post_processing_thread()

◆ scrap()

◆ start()

◆ stop()

Member Data Documentation

◆ m_error_registry

◆ m_items_to_postprocess_queues

◆ m_last_processed_daq_ts

◆ m_post_process_functions

◆ m_post_process_threads

◆ m_post_processing_enabled

◆ m_postprocess_queue_sizes

◆ m_preprocess_functions

◆ m_run_marker

◆ m_sourceid