#include <TPRequestHandler.hpp>

Inheritance diagram for dunedaq::trigger::TPRequestHandler:

Collaboration diagram for dunedaq::trigger::TPRequestHandler:

Public Types
using	inherited2 = datahandlinglibs::DefaultSkipListRequestHandler<TriggerPrimitiveTypeAdapter>

Public Types inherited from dunedaq::datahandlinglibs::DefaultSkipListRequestHandler< TriggerPrimitiveTypeAdapter >
using	inherited

using	SkipListAcc

using	SkipListSkip

Public Types inherited from dunedaq::datahandlinglibs::DefaultRequestHandlerModel< TriggerPrimitiveTypeAdapter, datahandlinglibs::SkipListLatencyBufferModel< TriggerPrimitiveTypeAdapter > >
using	RDT

using	LBT

using	RequestResult

using	ResultCode

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
	TPRequestHandler (std::shared_ptr< datahandlinglibs::SkipListLatencyBufferModel< TriggerPrimitiveTypeAdapter > > &latency_buffer, std::unique_ptr< datahandlinglibs::FrameErrorRegistry > &error_registry)

void	conf (const appmodel::DataHandlerModule *conf) override

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

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

void	periodic_data_transmission () override
	Periodic data transmission - relevant for trigger in particular.

Public Member Functions inherited from dunedaq::datahandlinglibs::DefaultSkipListRequestHandler< TriggerPrimitiveTypeAdapter >
	DefaultSkipListRequestHandler (std::shared_ptr< datahandlinglibs::SkipListLatencyBufferModel< TriggerPrimitiveTypeAdapter > > &latency_buffer, std::unique_ptr< datahandlinglibs::FrameErrorRegistry > &error_registry)

Public Member Functions inherited from dunedaq::datahandlinglibs::DefaultRequestHandlerModel< TriggerPrimitiveTypeAdapter, datahandlinglibs::SkipListLatencyBufferModel< TriggerPrimitiveTypeAdapter > >
	DefaultRequestHandlerModel (std::shared_ptr< datahandlinglibs::SkipListLatencyBufferModel< TriggerPrimitiveTypeAdapter > > &latency_buffer, std::unique_ptr< FrameErrorRegistry > &error_registry)

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

void	record (const appfwk::DAQModule::CommandData_t &args) override

void	cleanup_check () override
	Check if cleanup is necessary and execute it if necessary.

void	issue_request (dfmessages::DataRequest datarequest, bool is_retry=false) override
	Issue a data request to the request handler.

virtual dunedaq::daqdataformats::timestamp_t	get_cutoff_timestamp ()

virtual bool	supports_cutoff_timestamp ()

void	reset_oldest_time ()

std::uint64_t	get_oldest_time () override
	Get oldest timestamp in the buffer.

Public Member Functions inherited from dunedaq::datahandlinglibs::RequestHandlerConcept< TriggerPrimitiveTypeAdapter, datahandlinglibs::SkipListLatencyBufferModel< TriggerPrimitiveTypeAdapter > >
	RequestHandlerConcept ()

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

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

virtual	~RequestHandlerConcept ()

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

RequestHandlerConcept &	operator= (RequestHandlerConcept &&)=delete
	RequestHandlerConcept 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

Private Types
using	timestamp_t = std::uint64_t

Private Attributes
std::shared_ptr< iomanager::SenderConcept< dunedaq::trigger::TPSet > >	m_tpset_sink

uint64_t	m_run_number

uint64_t	m_next_tpset_seqno

timestamp_t	m_oldest_ts =0

timestamp_t	m_newest_ts =0

timestamp_t	m_start_win_ts =0

timestamp_t	m_end_win_ts =0

bool	m_first_cycle = true

uint64_t	m_ts_set_sender_offset_ticks = 6250000

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

Protected Types inherited from dunedaq::datahandlinglibs::RequestHandlerConcept< TriggerPrimitiveTypeAdapter, datahandlinglibs::SkipListLatencyBufferModel< TriggerPrimitiveTypeAdapter > >
enum	ResultCode

Protected Member Functions inherited from dunedaq::datahandlinglibs::DefaultSkipListRequestHandler< TriggerPrimitiveTypeAdapter >
void	cleanup () override

void	skip_list_cleanup_request ()

Protected Member Functions inherited from dunedaq::datahandlinglibs::DefaultRequestHandlerModel< TriggerPrimitiveTypeAdapter, datahandlinglibs::SkipListLatencyBufferModel< TriggerPrimitiveTypeAdapter > >
daqdataformats::FragmentHeader	create_fragment_header (const dfmessages::DataRequest &dr)

std::unique_ptr< daqdataformats::Fragment >	create_empty_fragment (const dfmessages::DataRequest &dr)

void	dump_to_buffer (const void data, std::size_t size, void buffer, uint32_t buffer_pos, const std::size_t &buffer_size)

void	periodic_cleanups ()

void	periodic_data_transmissions ()

void	check_waiting_requests ()

std::vector< std::pair< void *, size_t > >	get_fragment_pieces (uint64_t start_win_ts, uint64_t end_win_ts, RequestResult &rres)

RequestResult	data_request (dfmessages::DataRequest dr) override

virtual void	generate_opmon_data () override

Protected Member Functions inherited from dunedaq::datahandlinglibs::RequestHandlerConcept< TriggerPrimitiveTypeAdapter, datahandlinglibs::SkipListLatencyBufferModel< TriggerPrimitiveTypeAdapter > >
const std::string &	resultCodeAsString (ResultCode rc)

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 inherited from dunedaq::datahandlinglibs::DefaultRequestHandlerModel< TriggerPrimitiveTypeAdapter, datahandlinglibs::SkipListLatencyBufferModel< TriggerPrimitiveTypeAdapter > >
std::shared_ptr< datahandlinglibs::SkipListLatencyBufferModel< TriggerPrimitiveTypeAdapter > > &	m_latency_buffer

BufferedFileWriter	m_buffered_writer

utilities::ReusableThread	m_recording_thread

utilities::ReusableThread	m_cleanup_thread

utilities::ReusableThread	m_periodic_transmission_thread

std::map< dfmessages::DataRequest, int >	m_request_counter

std::size_t	m_max_requested_elements

std::mutex	m_cv_mutex

std::condition_variable	m_cv

std::atomic< bool >	m_cleanup_requested

std::atomic< int >	m_requests_running

std::vector< RequestElement >	m_waiting_requests

std::mutex	m_waiting_requests_lock

std::unique_ptr< boost::asio::thread_pool >	m_request_handler_thread_pool

size_t	m_num_request_handling_threads

std::unique_ptr< FrameErrorRegistry > &	m_error_registry

std::chrono::time_point< std::chrono::high_resolution_clock >	m_t0

std::atomic< bool >	m_run_marker

std::thread	m_waiting_queue_thread

std::atomic< bool >	m_recording

std::atomic< uint64_t >	m_next_timestamp_to_record

bool	m_configured

float	m_pop_limit_pct

float	m_pop_size_pct

unsigned	m_pop_limit_size

size_t	m_buffer_capacity

daqdataformats::SourceID	m_sourceid

uint16_t	m_detid

std::string	m_output_file

size_t	m_stream_buffer_size

bool	m_recording_configured

bool	m_warn_on_timeout

bool	m_warn_about_empty_buffer

uint32_t	m_periodic_data_transmission_ms

std::vector< std::string >	m_frag_out_conn_ids

std::atomic< int >	m_pop_counter

std::atomic< int >	m_num_buffer_cleanups

std::atomic< int >	m_pop_reqs

std::atomic< int >	m_pops_count

std::atomic< int >	m_occupancy

std::atomic< int >	m_num_requests_found

std::atomic< int >	m_num_requests_bad

std::atomic< int >	m_num_requests_old_window

std::atomic< int >	m_num_requests_delayed

std::atomic< int >	m_num_requests_uncategorized

std::atomic< int >	m_num_requests_timed_out

std::atomic< int >	m_handled_requests

std::atomic< int >	m_response_time_acc

std::atomic< int >	m_response_time_min

std::atomic< int >	m_response_time_max

std::atomic< int >	m_payloads_written

std::atomic< int >	m_bytes_written

std::atomic< uint64_t >	m_num_periodic_sent

std::atomic< uint64_t >	m_num_periodic_send_failed

std::atomic< uint64_t >	m_oldest_timestamp

int	m_fragment_send_timeout_ms

Protected Attributes inherited from dunedaq::datahandlinglibs::RequestHandlerConcept< TriggerPrimitiveTypeAdapter, datahandlinglibs::SkipListLatencyBufferModel< TriggerPrimitiveTypeAdapter > >
std::map< ResultCode, std::string >	ResultCodeStrings

Detailed Description

Definition at line 48 of file TPRequestHandler.hpp.

Member Typedef Documentation

◆ inherited2

using dunedaq::trigger::TPRequestHandler::inherited2 = datahandlinglibs::DefaultSkipListRequestHandler<TriggerPrimitiveTypeAdapter>

Definition at line 51 of file TPRequestHandler.hpp.

◆ timestamp_t

using dunedaq::trigger::TPRequestHandler::timestamp_t = std::uint64_t

private

Definition at line 70 of file TPRequestHandler.hpp.

Constructor & Destructor Documentation

◆ TPRequestHandler()

dunedaq::trigger::TPRequestHandler::TPRequestHandler	(	std::shared_ptr< datahandlinglibs::SkipListLatencyBufferModel< TriggerPrimitiveTypeAdapter > > &	latency_buffer,
		std::unique_ptr< datahandlinglibs::FrameErrorRegistry > &	error_registry )

inline

Definition at line 55 of file TPRequestHandler.hpp.

    : datahandlinglibs::DefaultSkipListRequestHandler<TriggerPrimitiveTypeAdapter>(
        latency_buffer,
        error_registry)
  {
    TLOG_DEBUG(TLVL_WORK_STEPS) << "TPRequestHandler created...";
  }

Member Function Documentation

◆ conf()

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

overridevirtual

Reimplemented from dunedaq::datahandlinglibs::DefaultRequestHandlerModel< TriggerPrimitiveTypeAdapter, datahandlinglibs::SkipListLatencyBufferModel< TriggerPrimitiveTypeAdapter > >.

Definition at line 11 of file TPRequestHandler.cpp.

                                                            {
 
   for (auto output : conf->get_outputs()) {
      if (output->get_data_type() == "TPSet") {
         try {
            m_tpset_sink = iomanager::IOManager::get()->get_sender<dunedaq::trigger::TPSet>(output->UID());
         } catch (const ers::Issue& excpt) {
            throw datahandlinglibs::ResourceQueueError(ERS_HERE, "tp queue", "DefaultRequestHandlerModel", excpt);
         }
      }
   }
   inherited2::conf(conf);
}

◆ periodic_data_transmission()

void dunedaq::trigger::TPRequestHandler::periodic_data_transmission ( )

overridevirtual

Periodic data transmission - relevant for trigger in particular.

Reimplemented from dunedaq::datahandlinglibs::DefaultRequestHandlerModel< TriggerPrimitiveTypeAdapter, datahandlinglibs::SkipListLatencyBufferModel< TriggerPrimitiveTypeAdapter > >.

Definition at line 49 of file TPRequestHandler.cpp.

                                             {
   
   if (m_tpset_sink == nullptr) return;
 
   dunedaq::dfmessages::DataRequest dr;
 
   {
      std::unique_lock<std::mutex> lock(m_cv_mutex);
      m_cv.wait(lock, [&] { return !m_cleanup_requested; });
      m_requests_running++;
   }
   m_cv.notify_all();
   if(m_latency_buffer->occupancy() != 0) {
       // Prepare response
       RequestResult rres(ResultCode::kUnknown, dr);
       std::vector<std::pair<void*, size_t>> frag_pieces;
 
       // Get the newest TP
       SkipListAcc acc(inherited2::m_latency_buffer->get_skip_list());
       auto tail = acc.last();
       auto head = acc.first();
       m_newest_ts = (*tail).get_timestamp();
       m_oldest_ts = (*head).get_timestamp();
       
       if (m_first_cycle) {
          m_start_win_ts = m_oldest_ts;
      m_first_cycle = false;
       }
       if (m_newest_ts - m_start_win_ts > m_ts_set_sender_offset_ticks) {
         m_end_win_ts = m_newest_ts - m_ts_set_sender_offset_ticks;
         frag_pieces = get_fragment_pieces(m_start_win_ts, m_end_win_ts, rres);
         auto num_tps = frag_pieces.size();
         trigger::TPSet tpset;
         tpset.run_number = m_run_number;
         tpset.type = num_tps>0 ? trigger::TPSet::Type::kPayload : trigger::TPSet::Type::kHeartbeat;
         tpset.origin = m_sourceid;
         tpset.start_time = m_start_win_ts; // provisory timestamp, will be filled with first TP
         tpset.end_time = m_end_win_ts; // provisory timestamp, will be filled with last TP
         tpset.seqno = m_next_tpset_seqno++; // NOLINT(runtime/increment_decrement)
       // reserve the space for efficiency
         if (num_tps > 0) {    
            tpset.objects.reserve(frag_pieces.size());
            bool first_tp = true;
            for( auto f : frag_pieces) {
               trgdataformats::TriggerPrimitive tp = *(static_cast<trgdataformats::TriggerPrimitive*>(f.first));
           
               if(first_tp) {
                  tpset.start_time = tp.time_start;
                  first_tp = false;
               }
           tpset.end_time = tp.time_start;
               tpset.objects.emplace_back(std::move(tp)); 
            }
          } 
         if(!m_tpset_sink->try_send(std::move(tpset), iomanager::Sender::s_no_block)) {
            ers::warning(DroppedTPSet(ERS_HERE, m_start_win_ts, m_end_win_ts));
            m_num_periodic_send_failed++;
         }
         m_num_periodic_sent++;
 
         //remember what we sent for the next loop
         m_start_win_ts = m_end_win_ts;
       }
    }
    {
      std::lock_guard<std::mutex> lock(m_cv_mutex);
      m_requests_running--;
    }
    m_cv.notify_all();  
   return;
}

◆ scrap()

void dunedaq::trigger::TPRequestHandler::scrap ( const appfwk::DAQModule::CommandData_t & args )

overridevirtual

Reimplemented from dunedaq::datahandlinglibs::DefaultRequestHandlerModel< TriggerPrimitiveTypeAdapter, datahandlinglibs::SkipListLatencyBufferModel< TriggerPrimitiveTypeAdapter > >.

Definition at line 26 of file TPRequestHandler.cpp.

{
  m_tpset_sink.reset();
  inherited2::scrap(args);
}

◆ start()

void dunedaq::trigger::TPRequestHandler::start ( const appfwk::DAQModule::CommandData_t & args )

overridevirtual

Reimplemented from dunedaq::datahandlinglibs::DefaultRequestHandlerModel< TriggerPrimitiveTypeAdapter, datahandlinglibs::SkipListLatencyBufferModel< TriggerPrimitiveTypeAdapter > >.

Definition at line 33 of file TPRequestHandler.cpp.

{
 
   m_oldest_ts=0;
   m_newest_ts=0;
   m_start_win_ts=0;
   m_end_win_ts=0;
   m_first_cycle = true;
    
   inherited2::start(args);
   rcif::cmd::StartParams start_params = args.get<rcif::cmd::StartParams>();
   m_run_number = start_params.run;
 
}

Member Data Documentation

◆ m_end_win_ts

timestamp_t dunedaq::trigger::TPRequestHandler::m_end_win_ts =0

private

Definition at line 78 of file TPRequestHandler.hpp.

◆ m_first_cycle

bool dunedaq::trigger::TPRequestHandler::m_first_cycle = true

private

Definition at line 79 of file TPRequestHandler.hpp.

◆ m_newest_ts

timestamp_t dunedaq::trigger::TPRequestHandler::m_newest_ts =0

private

Definition at line 76 of file TPRequestHandler.hpp.

◆ m_next_tpset_seqno

uint64_t dunedaq::trigger::TPRequestHandler::m_next_tpset_seqno

private

Definition at line 73 of file TPRequestHandler.hpp.

◆ m_oldest_ts

timestamp_t dunedaq::trigger::TPRequestHandler::m_oldest_ts =0

private

Definition at line 75 of file TPRequestHandler.hpp.

◆ m_run_number

uint64_t dunedaq::trigger::TPRequestHandler::m_run_number

private

Definition at line 72 of file TPRequestHandler.hpp.

◆ m_start_win_ts

timestamp_t dunedaq::trigger::TPRequestHandler::m_start_win_ts =0

private

Definition at line 77 of file TPRequestHandler.hpp.

◆ m_tpset_sink

std::shared_ptr<iomanager::SenderConcept<dunedaq::trigger::TPSet> > dunedaq::trigger::TPRequestHandler::m_tpset_sink

private

Definition at line 71 of file TPRequestHandler.hpp.

◆ m_ts_set_sender_offset_ticks

uint64_t dunedaq::trigger::TPRequestHandler::m_ts_set_sender_offset_ticks = 6250000

private

Definition at line 80 of file TPRequestHandler.hpp.

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

/github/workspace/dunedaq/sourcecode/trigger/src/trigger/TPRequestHandler.hpp
/github/workspace/dunedaq/sourcecode/trigger/src/TPRequestHandler.cpp

Public Types

Public Member Functions

Private Types

Private Attributes

Additional Inherited Members

Detailed Description

Member Typedef Documentation

◆ inherited2

◆ timestamp_t

Constructor & Destructor Documentation

◆ TPRequestHandler()

Member Function Documentation

◆ conf()

◆ periodic_data_transmission()

◆ scrap()

◆ start()

Member Data Documentation

◆ m_end_win_ts

◆ m_first_cycle

◆ m_newest_ts

◆ m_next_tpset_seqno

◆ m_oldest_ts

◆ m_run_number

◆ m_start_win_ts

◆ m_tpset_sink

◆ m_ts_set_sender_offset_ticks