dunedaq::trigger::TPProcessor Class Reference

#include <TPProcessor.hpp>

Inheritance diagram for dunedaq::trigger::TPProcessor:

Collaboration diagram for dunedaq::trigger::TPProcessor:

Public Types
using	inherited = datahandlinglibs::TaskRawDataProcessorModel<TriggerPrimitiveTypeAdapter>
using	tpptr = TriggerPrimitiveTypeAdapter*
using	consttpptr = const TriggerPrimitiveTypeAdapter*
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
	TPProcessor (std::unique_ptr< datahandlinglibs::FrameErrorRegistry > &error_registry, bool post_processing_enabled)
	~TPProcessor ()
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< TriggerPrimitiveTypeAdapter >
	TaskRawDataProcessorModel (std::unique_ptr< FrameErrorRegistry > &error_registry, bool post_processing_enabled)
	~TaskRawDataProcessorModel ()
void	reset_last_daq_time ()
std::uint64_t	get_last_daq_time () override
	Get newest timestamp of last seen packet.
void	preprocess_item (TriggerPrimitiveTypeAdapter *item) override
	Preprocess one element.
void	postprocess_item (const TriggerPrimitiveTypeAdapter *item) override
	Postprocess one element.
void	add_preprocess_task (Task &&task)
void	add_postprocess_task (Task &&task)
void	invoke_all_preprocess_functions (TriggerPrimitiveTypeAdapter *item)
void	launch_all_preprocess_functions (TriggerPrimitiveTypeAdapter *item)
Public Member Functions inherited from dunedaq::datahandlinglibs::RawDataProcessorConcept< TriggerPrimitiveTypeAdapter >
	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_ta (const TriggerPrimitiveTypeAdapter *tp, std::shared_ptr< triggeralgs::TriggerActivityMaker > tam)
Protected Member Functions inherited from dunedaq::datahandlinglibs::TaskRawDataProcessorModel< TriggerPrimitiveTypeAdapter >
void	run_post_processing_thread (std::function< void(const TriggerPrimitiveTypeAdapter *)> &function, folly::ProducerConsumerQueue< const TriggerPrimitiveTypeAdapter * > &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< TriggerPrimitiveTypeAdapter >
std::vector< std::function< void(TriggerPrimitiveTypeAdapter *)> >	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 TriggerPrimitiveTypeAdapter *)> >	m_post_process_functions
std::vector< std::unique_ptr< folly::ProducerConsumerQueue< const TriggerPrimitiveTypeAdapter * > > >	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::TriggerActivityMaker > >	m_tams
std::shared_ptr< iomanager::SenderConcept< triggeralgs::TriggerActivity > >	m_ta_sink
daqdataformats::SourceID	m_sourceid
std::atomic< metric_counter_type >	m_tp_received_count { 0 }
std::atomic< metric_counter_type >	m_ta_made_count { 0 }
std::atomic< metric_counter_type >	m_ta_sent_count { 0 }
std::atomic< metric_counter_type >	m_ta_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 TPProcessor.hpp.

Member Typedef Documentation

consttpptr

using dunedaq::trigger::TPProcessor::consttpptr = const TriggerPrimitiveTypeAdapter*

Definition at line 38 of file TPProcessor.hpp.

inherited

using dunedaq::trigger::TPProcessor::inherited = datahandlinglibs::TaskRawDataProcessorModel<TriggerPrimitiveTypeAdapter>

Definition at line 36 of file TPProcessor.hpp.

metric_counter_type

using dunedaq::trigger::TPProcessor::metric_counter_type = uint64_t

private

Definition at line 74 of file TPProcessor.hpp.

tpptr

using dunedaq::trigger::TPProcessor::tpptr = TriggerPrimitiveTypeAdapter*

Definition at line 37 of file TPProcessor.hpp.

Constructor & Destructor Documentation

TPProcessor()

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

explicit

Definition at line 37 of file TPProcessor.cpp.

  : TaskRawDataProcessorModel<TriggerPrimitiveTypeAdapter>(error_registry, post_processing_enabled)
{
}

~TPProcessor()

dunedaq::trigger::TPProcessor::~TPProcessor

(

)

Definition at line 42 of file TPProcessor.cpp.

{}

Member Function Documentation

conf()

void dunedaq::trigger::TPProcessor::conf ( const appmodel::DataHandlerModule * conf )

overridevirtual

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

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

Definition at line 76 of file TPProcessor.cpp.

{
  TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << "TPProcessor: Entering conf() method";
 
  for (auto output : conf->get_outputs()) {
   try {
      if (output->get_data_type() == "TriggerActivity") {
         m_ta_sink = get_iom_sender<triggeralgs::TriggerActivity>(output->UID());
      }
    } catch (const ers::Issue& excpt) {
      ers::error(datahandlinglibs::ResourceQueueError(ERS_HERE, "ta", "DefaultRequestHandlerModel", excpt));
    }
  }
 
  m_sourceid.id = conf->get_source_id();
  m_sourceid.subsystem = TriggerPrimitiveTypeAdapter::subsystem;
  
  std::vector<const appmodel::TAAlgorithm*> ta_algorithms;
  auto dp = conf->get_module_configuration()->get_data_processor();
  auto proc_conf = dp->cast<appmodel::TPDataProcessor>();
  if (proc_conf != nullptr && m_post_processing_enabled) {
    ta_algorithms = proc_conf->get_algorithms();
    }
 
  for (auto algo : ta_algorithms)  {
    TLOG() << "Selected TA algorithm: " << algo->UID() << " from class " << algo->class_name();
    std::shared_ptr<triggeralgs::TriggerActivityMaker> maker = make_ta_maker(algo->class_name());
    nlohmann::json algo_json = algo->to_json(true);
 
    TLOG() << "Algo config:\n" << algo_json.dump();
 
    maker->configure(algo_json[algo->UID()]);
    inherited::add_postprocess_task(std::bind(&TPProcessor::find_ta, this, std::placeholders::_1, maker));
    m_tams.push_back(maker);
  }
  m_latency_monitoring.store( dp->get_latency_monitoring() );
  inherited::conf(conf);
 
  TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << "TPProcessor: Exiting conf() method";
}

find_ta()

void dunedaq::trigger::TPProcessor::find_ta ( const TriggerPrimitiveTypeAdapter * tp, std::shared_ptr< triggeralgs::TriggerActivityMaker > taa )

protected

Pipeline Stage 2.: Do TA finding

Pipeline Stage 2.: Do software TPG

Definition at line 153 of file TPProcessor.cpp.

{
  if (m_latency_monitoring.load()) m_latency_instance.update_latency_in( tp->tp.time_start ); // time_start or time_peak ?
  m_tp_received_count++;    
  std::vector<triggeralgs::TriggerActivity> tas;
  taa->operator()(tp->tp, tas);
 
  while (tas.size()) {
      m_ta_made_count++;
      if (m_latency_monitoring.load()) m_latency_instance.update_latency_out( tas.back().time_start );
      if (!m_ta_sink->try_send(std::move(tas.back()), iomanager::Sender::s_no_block)) {
        ers::warning(TADropped(ERS_HERE, tp->tp.time_start, m_sourceid.id));
        m_ta_failed_sent_count++;
      } else {
        m_ta_sent_count++;
      }
      tas.pop_back();
  }
  m_last_processed_daq_ts = tp->tp.time_start;
  return;
}

generate_opmon_data()

void dunedaq::trigger::TPProcessor::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< TriggerPrimitiveTypeAdapter >.

Definition at line 128 of file TPProcessor.cpp.

{
  opmon::TPProcessorInfo info;
 
  info.set_tp_received_count( m_tp_received_count.load() );
  info.set_ta_made_count( m_ta_made_count.load() );
  info.set_ta_sent_count( m_ta_sent_count.load() );
  info.set_ta_failed_sent_count( m_ta_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::TPProcessor::print_opmon_stats ( )

private

Definition at line 176 of file TPProcessor.cpp.

{
  TLOG() << "TPProcessor opmon counters summary:";
  TLOG() << "------------------------------";
  TLOG() << "TPs received: \t\t" << m_tp_received_count;
  TLOG() << "TAs made: \t\t" << m_ta_made_count;
  TLOG() << "TAs sent: \t\t" << m_ta_sent_count;
  TLOG() << "TAs failed to send: \t" << m_ta_failed_sent_count;
  TLOG();
}

scrap()

void dunedaq::trigger::TPProcessor::scrap ( const nlohmann::json & )

overridevirtual

Unconfigure.

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

Definition at line 118 of file TPProcessor.cpp.

{
  TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << "TPProcessor: Entering scrap() method";
  m_tams.clear();
  m_ta_sink.reset();
  inherited::scrap(args);
  TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << "TPProcessor: Exiting scrap() method";
}

start()

void dunedaq::trigger::TPProcessor::start ( const nlohmann::json & )

overridevirtual

Start operation.

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

Definition at line 46 of file TPProcessor.cpp.

{
  TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << "TPProcessor: Entering start() method";
 
  // Reset stats
  m_tp_received_count.store(0);
  m_ta_made_count.store(0);
  m_ta_sent_count.store(0);
  m_ta_failed_sent_count.store(0);
 
  m_running_flag.store(true);
 
  inherited::start(args);
 
  TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << "TPProcessor: Exiting start() method";
}

stop()

void dunedaq::trigger::TPProcessor::stop ( const nlohmann::json & )

overridevirtual

Stop operation.

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

Definition at line 64 of file TPProcessor.cpp.

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

Member Data Documentation

m_current_ts

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

protected

Definition at line 58 of file TPProcessor.hpp.

m_latency_in

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

private

Definition at line 85 of file TPProcessor.hpp.

{ 0 };

m_latency_instance

dunedaq::trigger::Latency dunedaq::trigger::TPProcessor::m_latency_instance

private

Definition at line 84 of file TPProcessor.hpp.

m_latency_monitoring

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

private

Definition at line 83 of file TPProcessor.hpp.

{ false };

m_latency_out

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

private

Definition at line 86 of file TPProcessor.hpp.

{ 0 };

m_previous_ts

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

protected

Definition at line 57 of file TPProcessor.hpp.

m_running_flag

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

private

Definition at line 82 of file TPProcessor.hpp.

{ false };

m_sourceid

daqdataformats::SourceID dunedaq::trigger::TPProcessor::m_sourceid

private

Definition at line 72 of file TPProcessor.hpp.

m_ta_failed_sent_count

std::atomic<metric_counter_type> dunedaq::trigger::TPProcessor::m_ta_failed_sent_count { 0 }

private

Definition at line 78 of file TPProcessor.hpp.

{ 0 };

m_ta_made_count

std::atomic<metric_counter_type> dunedaq::trigger::TPProcessor::m_ta_made_count { 0 }

private

Definition at line 76 of file TPProcessor.hpp.

{ 0 };

m_ta_sent_count

std::atomic<metric_counter_type> dunedaq::trigger::TPProcessor::m_ta_sent_count { 0 }

private

Definition at line 77 of file TPProcessor.hpp.

{ 0 };

m_ta_sink

std::shared_ptr<iomanager::SenderConcept<triggeralgs::TriggerActivity> > dunedaq::trigger::TPProcessor::m_ta_sink

private

Definition at line 70 of file TPProcessor.hpp.

m_tams

std::vector<std::shared_ptr<triggeralgs::TriggerActivityMaker> > dunedaq::trigger::TPProcessor::m_tams

private

Definition at line 68 of file TPProcessor.hpp.

m_tp_received_count

std::atomic<metric_counter_type> dunedaq::trigger::TPProcessor::m_tp_received_count { 0 }

private

Definition at line 75 of file TPProcessor.hpp.

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

---

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

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