dunedaq::trigger::TAProcessor Class Reference

#include <TAProcessor.hpp>

Inheritance diagram for dunedaq::trigger::TAProcessor:

Collaboration diagram for dunedaq::trigger::TAProcessor:

Public Types
using	inherited = datahandlinglibs::TaskRawDataProcessorModel<TAWrapper>
using	taptr = TAWrapper*
using	consttaptr = const TAWrapper*
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
	TAProcessor (std::unique_ptr< datahandlinglibs::FrameErrorRegistry > &error_registry, bool post_processing_enabled)
	~TAProcessor ()
void	start (const nlohmann::json &args) override
	Start operation.
void	stop (const nlohmann::json &args) override
	Stop operation.
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 nlohmann::json &args) override
	Unconfigure.
void	generate_opmon_data () override
Public Member Functions inherited from dunedaq::datahandlinglibs::TaskRawDataProcessorModel< TAWrapper >
	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 nlohmann::json &) override
	Unconfigure.
void	start (const nlohmann::json &) override
	Start operation.
void	stop (const nlohmann::json &) 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 (TAWrapper *item) override
	Preprocess one element.
void	postprocess_item (const TAWrapper *item) override
	Postprocess one element.
void	add_preprocess_task (Task &&task)
void	add_postprocess_task (Task &&task)
void	invoke_all_preprocess_functions (TAWrapper *item)
void	launch_all_preprocess_functions (TAWrapper *item)
Public Member Functions inherited from dunedaq::datahandlinglibs::RawDataProcessorConcept< TAWrapper >
	RawDataProcessorConcept ()
	RawDataProcessorConcept (const RawDataProcessorConcept &)=delete
	RawDataProcessorConcept is not copy-constructible.
	RawDataProcessorConcept (RawDataProcessorConcept &&)=delete
	RawDataProcessorConcept is not move-constructible.
virtual	~RawDataProcessorConcept ()
RawDataProcessorConcept &	operator= (const RawDataProcessorConcept &)=delete
	RawDataProcessorConcept is not copy-assginable.
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
void	find_tc (const TAWrapper *ta, std::shared_ptr< triggeralgs::TriggerCandidateMaker > tcm)
Protected Member Functions inherited from dunedaq::datahandlinglibs::TaskRawDataProcessorModel< TAWrapper >
void	run_post_processing_thread (std::function< void(const TAWrapper *)> &function, folly::ProducerConsumerQueue< const TAWrapper * > &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
dunedaq::daqdataformats::timestamp_t	m_previous_ts = 0
dunedaq::daqdataformats::timestamp_t	m_current_ts = 0
Protected Attributes inherited from dunedaq::datahandlinglibs::TaskRawDataProcessorModel< TAWrapper >
std::vector< std::function< void(TAWrapper *)> >	m_preprocess_functions
std::unique_ptr< FrameErrorRegistry > &	m_error_registry
bool	m_post_processing_enabled
std::atomic< bool >	m_run_marker
std::vector< std::function< void(const TAWrapper *)> >	m_post_process_functions
std::vector< std::unique_ptr< folly::ProducerConsumerQueue< const TAWrapper * > > >	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

Private Types
using	metric_counter_type = uint64_t

Private Member Functions
void	print_opmon_stats ()

Private Attributes
std::vector< std::shared_ptr< triggeralgs::TriggerCandidateMaker > >	m_tcms
std::shared_ptr< iomanager::SenderConcept< triggeralgs::TriggerCandidate > >	m_tc_sink
daqdataformats::SourceID	m_sourceid
std::atomic< metric_counter_type >	m_ta_received_count { 0 }
std::atomic< metric_counter_type >	m_tc_made_count { 0 }
std::atomic< metric_counter_type >	m_tc_sent_count { 0 }
std::atomic< metric_counter_type >	m_tc_failed_sent_count { 0 }
std::atomic< bool >	m_running_flag { false }
std::atomic< bool >	m_latency_monitoring { false }
dunedaq::trigger::Latency	m_latency_instance
std::atomic< metric_counter_type >	m_latency_in { 0 }
std::atomic< metric_counter_type >	m_latency_out { 0 }

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

Detailed Description

Definition at line 32 of file TAProcessor.hpp.

Member Typedef Documentation

consttaptr

using dunedaq::trigger::TAProcessor::consttaptr = const TAWrapper*

Definition at line 38 of file TAProcessor.hpp.

inherited

using dunedaq::trigger::TAProcessor::inherited = datahandlinglibs::TaskRawDataProcessorModel<TAWrapper>

Definition at line 36 of file TAProcessor.hpp.

metric_counter_type

using dunedaq::trigger::TAProcessor::metric_counter_type = uint64_t

private

Definition at line 74 of file TAProcessor.hpp.

taptr

using dunedaq::trigger::TAProcessor::taptr = TAWrapper*

Definition at line 37 of file TAProcessor.hpp.

Constructor & Destructor Documentation

TAProcessor()

dunedaq::trigger::TAProcessor::TAProcessor ( std::unique_ptr< datahandlinglibs::FrameErrorRegistry > & error_registry, bool post_processing_enabled )

explicit

Definition at line 39 of file TAProcessor.cpp.

  : datahandlinglibs::TaskRawDataProcessorModel<TAWrapper>(error_registry, post_processing_enabled)
{
}

~TAProcessor()

dunedaq::trigger::TAProcessor::~TAProcessor

(

)

Definition at line 44 of file TAProcessor.cpp.

{}

Member Function Documentation

conf()

void dunedaq::trigger::TAProcessor::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< TAWrapper >.

Definition at line 78 of file TAProcessor.cpp.

{
  TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << "TAProcessor: Entering conf() method";
 
  for (auto output : conf->get_outputs()) {
   try {
      if (output->get_data_type() == "TriggerCandidate") {
         m_tc_sink = get_iom_sender<triggeralgs::TriggerCandidate>(output->UID());
      }
    } catch (const ers::Issue& excpt) {
      ers::error(datahandlinglibs::ResourceQueueError(ERS_HERE, "tc", "DefaultRequestHandlerModel", excpt));
    }
  }
 
  m_sourceid.id = conf->get_source_id();
  m_sourceid.subsystem = trigger::TAWrapper::subsystem;
  std::vector<const appmodel::TCAlgorithm*> tc_algorithms;
  auto dp = conf->get_module_configuration()->get_data_processor();
  auto proc_conf = dp->cast<appmodel::TADataProcessor>();
  if (proc_conf != nullptr && m_post_processing_enabled ) {
    tc_algorithms = proc_conf->get_algorithms();
    }
 
  for (auto algo : tc_algorithms)  {
    TLOG() << "Selected TC algorithm: " << algo->UID();
    std::shared_ptr<triggeralgs::TriggerCandidateMaker> maker = make_tc_maker(algo->class_name());
    nlohmann::json algo_json = algo->to_json(true);
    maker->configure(algo_json[algo->UID()]);
    inherited::add_postprocess_task(std::bind(&TAProcessor::find_tc, this, std::placeholders::_1, maker));
    m_tcms.push_back(maker);
  }
  m_latency_monitoring.store( dp->get_latency_monitoring() );
  inherited::conf(conf);
 
  TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << "TAProcessor: Exiting conf() method";
}

find_tc()

void dunedaq::trigger::TAProcessor::find_tc ( const TAWrapper * ta, std::shared_ptr< triggeralgs::TriggerCandidateMaker > tca )

protected

Pipeline Stage 2.: Do TA finding

Pipeline Stage 2.: Do software TPG

Definition at line 151 of file TAProcessor.cpp.

{
  //time_activity gave 0 :/
  if (m_latency_monitoring.load()) m_latency_instance.update_latency_in( ta->activity.time_start );
  m_ta_received_count++;
  std::vector<triggeralgs::TriggerCandidate> tcs;
  tca->operator()(ta->activity, tcs);
  for (auto tc : tcs) {
    m_tc_made_count++;
    if (m_latency_monitoring.load()) m_latency_instance.update_latency_out( tc.time_candidate );
    if(!m_tc_sink->try_send(std::move(tc), iomanager::Sender::s_no_block)) {
        ers::warning(TCDropped(ERS_HERE, tc.time_start, m_sourceid.id));
        m_tc_failed_sent_count++;
    } else {
      m_tc_sent_count++;
    }
  }
  m_last_processed_daq_ts = ta->activity.time_start;
  return;
}

generate_opmon_data()

void dunedaq::trigger::TAProcessor::generate_opmon_data ( )

overridevirtual

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

Reimplemented from dunedaq::datahandlinglibs::TaskRawDataProcessorModel< TAWrapper >.

Definition at line 126 of file TAProcessor.cpp.

{
  opmon::TAProcessorInfo info;
 
  info.set_ta_received_count( m_ta_received_count.load() );
  info.set_tc_made_count( m_tc_made_count.load() );
  info.set_tc_sent_count( m_tc_sent_count.load() );
  info.set_tc_failed_sent_count( m_tc_failed_sent_count.load() );
 
  this->publish(std::move(info));
 
  if ( m_latency_monitoring.load() && m_running_flag.load() ) {
    opmon::TriggerLatency lat_info;
 
    lat_info.set_latency_in( m_latency_instance.get_latency_in() );
    lat_info.set_latency_out( m_latency_instance.get_latency_out() );
 
    this->publish(std::move(lat_info));
  }
}

print_opmon_stats()

void dunedaq::trigger::TAProcessor::print_opmon_stats ( )

private

Definition at line 173 of file TAProcessor.cpp.

{
  TLOG() << "TAProcessor opmon counters summary:";
  TLOG() << "------------------------------";
  TLOG() << "TAs received: \t\t" << m_ta_received_count;
  TLOG() << "TCs made: \t\t" << m_tc_made_count;
  TLOG() << "TCs sent: \t\t" << m_tc_sent_count;
  TLOG() << "TCs failed to send: \t" << m_tc_failed_sent_count;
  TLOG();
}

scrap()

void dunedaq::trigger::TAProcessor::scrap ( const nlohmann::json & args )

overridevirtual

Unconfigure.

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

Definition at line 116 of file TAProcessor.cpp.

{
  TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << "TAProcessor: Entering scrap() method";
  m_tcms.clear();
  m_tc_sink.reset();
  inherited::scrap(args);
  TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << "TAProcessor: Exiting scrap() method";
}

start()

void dunedaq::trigger::TAProcessor::start ( const nlohmann::json & args )

overridevirtual

Start operation.

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

Definition at line 48 of file TAProcessor.cpp.

{
  TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << "TAProcessor: Entering start() method";
 
  // Reset stats
  m_ta_received_count.store(0);
  m_tc_made_count.store(0);
  m_tc_sent_count.store(0);
  m_tc_failed_sent_count.store(0);
 
  m_running_flag.store(true);
 
  inherited::start(args);
 
  TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << "TAProcessor: Exiting start() method";
}

stop()

void dunedaq::trigger::TAProcessor::stop ( const nlohmann::json & args )

overridevirtual

Stop operation.

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

Definition at line 66 of file TAProcessor.cpp.

{
  TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << "TAProcessor: Entering stop() method";
 
  inherited::stop(args);
  m_running_flag.store(false);
  print_opmon_stats();
 
  TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << "TAProcessor: Exiting stop() method";
}

Member Data Documentation

m_current_ts

dunedaq::daqdataformats::timestamp_t dunedaq::trigger::TAProcessor::m_current_ts = 0

protected

Definition at line 58 of file TAProcessor.hpp.

m_latency_in

std::atomic<metric_counter_type> dunedaq::trigger::TAProcessor::m_latency_in { 0 }

private

Definition at line 85 of file TAProcessor.hpp.

{ 0 };

m_latency_instance

dunedaq::trigger::Latency dunedaq::trigger::TAProcessor::m_latency_instance

private

Definition at line 84 of file TAProcessor.hpp.

m_latency_monitoring

std::atomic<bool> dunedaq::trigger::TAProcessor::m_latency_monitoring { false }

private

Definition at line 83 of file TAProcessor.hpp.

{ false };

m_latency_out

std::atomic<metric_counter_type> dunedaq::trigger::TAProcessor::m_latency_out { 0 }

private

Definition at line 86 of file TAProcessor.hpp.

{ 0 };

m_previous_ts

dunedaq::daqdataformats::timestamp_t dunedaq::trigger::TAProcessor::m_previous_ts = 0

protected

Definition at line 57 of file TAProcessor.hpp.

m_running_flag

std::atomic<bool> dunedaq::trigger::TAProcessor::m_running_flag { false }

private

Definition at line 82 of file TAProcessor.hpp.

{ false };

m_sourceid

daqdataformats::SourceID dunedaq::trigger::TAProcessor::m_sourceid

private

Definition at line 72 of file TAProcessor.hpp.

m_ta_received_count

std::atomic<metric_counter_type> dunedaq::trigger::TAProcessor::m_ta_received_count { 0 }

private

Definition at line 75 of file TAProcessor.hpp.

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

m_tc_failed_sent_count

std::atomic<metric_counter_type> dunedaq::trigger::TAProcessor::m_tc_failed_sent_count { 0 }

private

Definition at line 78 of file TAProcessor.hpp.

{ 0 };

m_tc_made_count

std::atomic<metric_counter_type> dunedaq::trigger::TAProcessor::m_tc_made_count { 0 }

private

Definition at line 76 of file TAProcessor.hpp.

{ 0 };

m_tc_sent_count

std::atomic<metric_counter_type> dunedaq::trigger::TAProcessor::m_tc_sent_count { 0 }

private

Definition at line 77 of file TAProcessor.hpp.

{ 0 };

m_tc_sink

std::shared_ptr<iomanager::SenderConcept<triggeralgs::TriggerCandidate> > dunedaq::trigger::TAProcessor::m_tc_sink

private

Definition at line 70 of file TAProcessor.hpp.

m_tcms

std::vector<std::shared_ptr<triggeralgs::TriggerCandidateMaker> > dunedaq::trigger::TAProcessor::m_tcms

private

Definition at line 68 of file TAProcessor.hpp.

---

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

• /github/workspace/dunedaq/sourcecode/trigger/src/trigger/TAProcessor.hpp
• /github/workspace/dunedaq/sourcecode/trigger/src/TAProcessor.cpp