dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType > Class Template Reference

#include <DefaultRequestHandlerModel.hpp>

Inheritance diagram for dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >:

Collaboration diagram for dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >:

Classes
struct	RequestElement

Public Types
using	RDT = ReadoutType
using	LBT = LatencyBufferType
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
	DefaultRequestHandlerModel (std::shared_ptr< LatencyBufferType > &latency_buffer, std::unique_ptr< FrameErrorRegistry > &error_registry)
void	conf (const dunedaq::appmodel::DataHandlerModule *)
void	scrap (const appfwk::DAQModule::CommandData_t &) override
void	start (const appfwk::DAQModule::CommandData_t &)
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.
virtual void	periodic_data_transmission () override
	Periodic data transmission - relevant for trigger in particular.
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< ReadoutType, LatencyBufferType >
	RequestHandlerConcept ()
virtual	~RequestHandlerConcept ()
	RequestHandlerConcept (const RequestHandlerConcept &)=delete
	RequestHandlerConcept is not copy-constructible.
RequestHandlerConcept &	operator= (const RequestHandlerConcept &)=delete
	RequestHandlerConcept is not copy-assginable.
	RequestHandlerConcept (RequestHandlerConcept &&)=delete
	RequestHandlerConcept is not move-constructible.
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

Protected Member Functions
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	cleanup ()
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< ReadoutType, LatencyBufferType >
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
std::shared_ptr< LatencyBufferType > &	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 = false
std::atomic< int >	m_requests_running = 0
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 = 0
std::unique_ptr< FrameErrorRegistry > &	m_error_registry
std::chrono::time_point< std::chrono::high_resolution_clock >	m_t0
std::atomic< bool >	m_run_marker = false
std::thread	m_waiting_queue_thread
std::atomic< bool >	m_recording = false
std::atomic< uint64_t >	m_next_timestamp_to_record = std::numeric_limits<uint64_t>::max()
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 = 0
bool	m_recording_configured = false
bool	m_warn_on_timeout = true
bool	m_warn_about_empty_buffer = true
uint32_t	m_periodic_data_transmission_ms = 0
std::vector< std::string >	m_frag_out_conn_ids
std::atomic< int >	m_pop_counter
std::atomic< int >	m_num_buffer_cleanups { 0 }
std::atomic< int >	m_pop_reqs
std::atomic< int >	m_pops_count
std::atomic< int >	m_occupancy
std::atomic< int >	m_num_requests_found { 0 }
std::atomic< int >	m_num_requests_bad { 0 }
std::atomic< int >	m_num_requests_old_window { 0 }
std::atomic< int >	m_num_requests_delayed { 0 }
std::atomic< int >	m_num_requests_uncategorized { 0 }
std::atomic< int >	m_num_requests_timed_out { 0 }
std::atomic< int >	m_handled_requests { 0 }
std::atomic< int >	m_response_time_acc { 0 }
std::atomic< int >	m_response_time_min { std::numeric_limits<int>::max() }
std::atomic< int >	m_response_time_max { 0 }
std::atomic< int >	m_payloads_written { 0 }
std::atomic< int >	m_bytes_written { 0 }
std::atomic< uint64_t >	m_num_periodic_sent { 0 }
std::atomic< uint64_t >	m_num_periodic_send_failed { 0 }
std::atomic< uint64_t >	m_oldest_timestamp { 0 }
int	m_fragment_send_timeout_ms
Protected Attributes inherited from dunedaq::datahandlinglibs::RequestHandlerConcept< ReadoutType, LatencyBufferType >
std::map< ResultCode, std::string >	ResultCodeStrings

Private Attributes
int	m_request_timeout_ms

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< ReadoutType, LatencyBufferType >
enum	ResultCode {   kFound = 0 , kNotFound , kTooOld , kNotYet ,   kPartial , kPartiallyOld , kCleanup , kUnknown }

Detailed Description

template<class ReadoutType, class LatencyBufferType>
class dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >

Definition at line 82 of file DefaultRequestHandlerModel.hpp.

Member Typedef Documentation

LBT

template<class ReadoutType , class LatencyBufferType >

using dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::LBT = LatencyBufferType

Definition at line 87 of file DefaultRequestHandlerModel.hpp.

RDT

template<class ReadoutType , class LatencyBufferType >

using dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::RDT = ReadoutType

Definition at line 86 of file DefaultRequestHandlerModel.hpp.

RequestResult

template<class ReadoutType , class LatencyBufferType >

using dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::RequestResult

Initial value:

 
    typename dunedaq::datahandlinglibs::RequestHandlerConcept<ReadoutType, LatencyBufferType>::RequestResult

Definition at line 89 of file DefaultRequestHandlerModel.hpp.

ResultCode

template<class ReadoutType , class LatencyBufferType >

using dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::ResultCode

Initial value:

 
    typename dunedaq::datahandlinglibs::RequestHandlerConcept<ReadoutType, LatencyBufferType>::ResultCode

Definition at line 91 of file DefaultRequestHandlerModel.hpp.

Constructor & Destructor Documentation

DefaultRequestHandlerModel()

template<class ReadoutType , class LatencyBufferType >

dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::DefaultRequestHandlerModel ( std::shared_ptr< LatencyBufferType > & latency_buffer, std::unique_ptr< FrameErrorRegistry > & error_registry )

inlineexplicit

Definition at line 95 of file DefaultRequestHandlerModel.hpp.

    : m_latency_buffer(latency_buffer)
    , m_recording_thread(0)
    , m_cleanup_thread(0)
    , m_waiting_requests()
    , m_waiting_requests_lock()
    , m_error_registry(error_registry)
    , m_pop_limit_pct(0.0f)
    , m_pop_size_pct(0.0f)
    , m_pop_limit_size(0)
    , m_buffer_capacity(0)
    , m_pop_counter{ 0 }
    , m_pop_reqs(0)
    , m_pops_count(0)
    , m_occupancy(0)
  //, m_response_time_log()
  //, m_response_time_log_lock()
  {
    TLOG_DEBUG(TLVL_WORK_STEPS) << "DefaultRequestHandlerModel created...";
  }

Member Function Documentation

check_waiting_requests()

template<class RDT , class LBT >

void dunedaq::datahandlinglibs::DefaultRequestHandlerModel< RDT, LBT >::check_waiting_requests ( )

protected

Definition at line 399 of file DefaultRequestHandlerModel.hxx.

{
  // At run stop, we wait until all waiting requests have either:
  //
  // 1. been serviced because an item past the end of the window arrived in the buffer
  // 2. timed out by going past m_request_timeout_ms, and returned a partial fragment
  while (m_run_marker.load()) {
    if (m_waiting_requests.size() > 0) {
        
      std::lock_guard<std::mutex> lock_guard(m_waiting_requests_lock);
 
      auto last_frame = m_latency_buffer->back();                                       // NOLINT
      uint64_t newest_ts = last_frame == nullptr ? std::numeric_limits<uint64_t>::min() // NOLINT(build/unsigned)
                                                 : last_frame->get_timestamp();
 
      for (auto iter = m_waiting_requests.begin(); iter!= m_waiting_requests.end();) {
    if((*iter).request.request_information.window_end < newest_ts) {
          issue_request((*iter).request, true);
      iter = m_waiting_requests.erase(iter);
    }
    else if (std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::high_resolution_clock::now() - (*iter).start_time).count() >= m_request_timeout_ms) {
          issue_request((*iter).request, true);
          if (m_warn_on_timeout) {
            ers::warning(dunedaq::datahandlinglibs::VerboseRequestTimedOut(ERS_HERE, m_sourceid,
                                                                      (*iter).request.trigger_number,
                                                                      (*iter).request.sequence_number,
                                                                      (*iter).request.run_number,
                                                                      (*iter).request.request_information.window_begin,
                                                                      (*iter).request.request_information.window_end,
                                                                      (*iter).request.data_destination));
          }
      m_num_requests_bad++;
          m_num_requests_timed_out++;
      iter = m_waiting_requests.erase(iter);
    }
    else {
    ++iter;
    }
      }
    }
    std::this_thread::sleep_for(std::chrono::milliseconds(1));
  }
}

cleanup()

template<class RDT , class LBT >

void dunedaq::datahandlinglibs::DefaultRequestHandlerModel< RDT, LBT >::cleanup ( )

protectedvirtual

Implements dunedaq::datahandlinglibs::RequestHandlerConcept< ReadoutType, LatencyBufferType >.

Reimplemented in dunedaq::datahandlinglibs::DefaultSkipListRequestHandler< T >, and dunedaq::datahandlinglibs::DefaultSkipListRequestHandler< TriggerPrimitiveTypeAdapter >.

Definition at line 372 of file DefaultRequestHandlerModel.hxx.

{
  auto size_guess = m_latency_buffer->occupancy();
  if (size_guess > m_pop_limit_size) {
    ++m_pop_reqs;
    unsigned to_pop = m_pop_size_pct * m_latency_buffer->occupancy();
 
    unsigned popped = 0;
    for (size_t i = 0; i < to_pop; ++i) {
      if (m_latency_buffer->front()->get_timestamp() < m_next_timestamp_to_record) {
        m_latency_buffer->pop(1);
        popped++;
      } else {
        break;
      }
    }
    m_occupancy = m_latency_buffer->occupancy();
    m_pops_count += popped;
    m_error_registry->remove_errors_until(m_latency_buffer->front()->get_timestamp());
  }
  // Update hte oldest timestamp monitorable
  m_oldest_timestamp = m_latency_buffer->front()->get_timestamp();
  m_num_buffer_cleanups++;
}

cleanup_check()

template<class RDT , class LBT >

void dunedaq::datahandlinglibs::DefaultRequestHandlerModel< RDT, LBT >::cleanup_check ( )

overridevirtual

Check if cleanup is necessary and execute it if necessary.

Implements dunedaq::datahandlinglibs::RequestHandlerConcept< ReadoutType, LatencyBufferType >.

Definition at line 215 of file DefaultRequestHandlerModel.hxx.

{
  std::unique_lock<std::mutex> lock(m_cv_mutex);
  if (m_latency_buffer->occupancy() > m_pop_limit_size && !m_cleanup_requested.exchange(true)) {
    m_cv.wait(lock, [&] { return m_requests_running == 0; });
    cleanup();
    m_cleanup_requested = false;
    m_cv.notify_all();
  }
}

conf()

template<class RDT , class LBT >

void dunedaq::datahandlinglibs::DefaultRequestHandlerModel< RDT, LBT >::conf ( const dunedaq::appmodel::DataHandlerModule * conf )

virtual

Implements dunedaq::datahandlinglibs::RequestHandlerConcept< ReadoutType, LatencyBufferType >.

Reimplemented in dunedaq::datahandlinglibs::ZeroCopyRecordingRequestHandlerModel< ReadoutType, LatencyBufferType >, and dunedaq::trigger::TPRequestHandler.

Definition at line 8 of file DefaultRequestHandlerModel.hxx.

{
 
  auto reqh_conf = conf->get_module_configuration()->get_request_handler();
  m_sourceid.id = conf->get_source_id();
  m_sourceid.subsystem = RDT::subsystem;
  m_detid = conf->get_detector_id();
  m_pop_limit_pct = reqh_conf->get_pop_limit_pct();
  m_pop_size_pct = reqh_conf->get_pop_size_pct();
 
  m_buffer_capacity = conf->get_module_configuration()->get_latency_buffer()->get_size();
  m_num_request_handling_threads = reqh_conf->get_handler_threads();
  m_request_timeout_ms = reqh_conf->get_request_timeout();
 
  for (auto output : conf->get_outputs()) {
    if (output->get_data_type() == "Fragment") {
      m_fragment_send_timeout_ms = output->get_send_timeout_ms();
      // 19-Dec-2024, KAB: store the names/IDs of the Fragment output connections so that
      // we can confirm that they are ready for sending at 'start' time.
      m_frag_out_conn_ids.push_back(output->UID());
    }
  }
 
  if (m_recording_configured == false) {
    auto dr = reqh_conf->get_data_recorder();
    if(dr != nullptr) {
      m_output_file = dr->get_output_file();
      if (remove(m_output_file.c_str()) == 0) {
        TLOG_DEBUG(TLVL_WORK_STEPS) << "Removed existing output file from previous run: " << m_output_file << std::endl;
      }
      m_stream_buffer_size = dr->get_streaming_buffer_size();
      m_buffered_writer.open(m_output_file, m_stream_buffer_size, dr->get_compression_algorithm(), dr->get_use_o_direct());
      m_recording_configured = true;
    }
  }
 
  m_warn_on_timeout = reqh_conf->get_warn_on_timeout();
  m_warn_about_empty_buffer = reqh_conf->get_warn_on_empty_buffer();
  m_periodic_data_transmission_ms = reqh_conf->get_periodic_data_transmission_ms();
  
  if (m_pop_limit_pct < 0.0f || m_pop_limit_pct > 1.0f || m_pop_size_pct < 0.0f || m_pop_size_pct > 1.0f) {
    ers::error(ConfigurationError(ERS_HERE, m_sourceid, "Auto-pop percentage out of range."));
  } else {
    m_pop_limit_size = m_pop_limit_pct * m_buffer_capacity;
    m_max_requested_elements = m_pop_limit_size - m_pop_limit_size * m_pop_size_pct;
  }
 
  m_recording_thread.set_name("recording", m_sourceid.id);
  m_cleanup_thread.set_name("cleanup", m_sourceid.id);
  m_periodic_transmission_thread.set_name("periodic", m_sourceid.id);
 
  std::ostringstream oss;
  oss << "RequestHandler configured. " << std::fixed << std::setprecision(2)
      << "auto-pop limit: " << m_pop_limit_pct * 100.0f << "% "
      << "auto-pop size: " << m_pop_size_pct * 100.0f << "% "
      << "max requested elements: " << m_max_requested_elements;
  TLOG_DEBUG(TLVL_WORK_STEPS) << oss.str();
}

create_empty_fragment()

template<class RDT , class LBT >

std::unique_ptr< daqdataformats::Fragment > dunedaq::datahandlinglibs::DefaultRequestHandlerModel< RDT, LBT >::create_empty_fragment ( const dfmessages::DataRequest & dr )

protected

Definition at line 336 of file DefaultRequestHandlerModel.hxx.

{
  auto frag_header = create_fragment_header(dr);
  frag_header.error_bits |= (0x1 << static_cast<size_t>(daqdataformats::FragmentErrorBits::kDataNotFound));
  auto fragment = std::make_unique<daqdataformats::Fragment>(std::vector<std::pair<void*, size_t>>());
  fragment->set_header_fields(frag_header);
  return fragment;
}

create_fragment_header()

template<class ReadoutType , class LatencyBufferType >

daqdataformats::FragmentHeader dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::create_fragment_header ( const dfmessages::DataRequest & dr )

inlineprotected

Definition at line 169 of file DefaultRequestHandlerModel.hpp.

  {
    daqdataformats::FragmentHeader fh;
    fh.size = sizeof(fh);
    fh.trigger_number = dr.trigger_number;
    fh.trigger_timestamp = dr.trigger_timestamp;
    fh.window_begin = dr.request_information.window_begin;
    fh.window_end = dr.request_information.window_end;
    fh.run_number = dr.run_number;
    fh.fragment_type = static_cast<daqdataformats::fragment_type_t>(ReadoutType::fragment_type);
    fh.sequence_number = dr.sequence_number;
    fh.detector_id = m_detid;
    fh.element_id = { m_sourceid.subsystem, m_sourceid.id };
    return fh;
  }

data_request()

template<class RDT , class LBT >

DefaultRequestHandlerModel< RDT, LBT >::RequestResult dunedaq::datahandlinglibs::DefaultRequestHandlerModel< RDT, LBT >::data_request ( dfmessages::DataRequest dr )

overrideprotectedvirtual

Implements dunedaq::datahandlinglibs::RequestHandlerConcept< ReadoutType, LatencyBufferType >.

Definition at line 534 of file DefaultRequestHandlerModel.hxx.

{
  // Prepare response
  RequestResult rres(ResultCode::kUnknown, dr);
 
  // Prepare FragmentHeader and empty Fragment pieces list
  auto frag_header = create_fragment_header(dr);
  std::vector<std::pair<void*, size_t>> frag_pieces;
  std::ostringstream oss;
 
  //bool local_data_not_found_flag = false;
  if (m_latency_buffer->occupancy() == 0) {
    if (m_warn_about_empty_buffer) {
      ers::warning(RequestOnEmptyBuffer(ERS_HERE, m_sourceid, "Data not found"));
    } 
    frag_header.error_bits |= (0x1 << static_cast<size_t>(daqdataformats::FragmentErrorBits::kDataNotFound));
    rres.result_code = ResultCode::kNotFound;
    ++m_num_requests_bad;    
  }
  else {
    frag_pieces = get_fragment_pieces(dr.request_information.window_begin, dr.request_information.window_end, rres);
 
    auto front_element = m_latency_buffer->front();           // NOLINT
    auto last_element = m_latency_buffer->back();             // NOLINT
    uint64_t last_ts = front_element->get_timestamp();  // NOLINT(build/unsigned)
    uint64_t newest_ts = last_element->get_timestamp(); // NOLINT(build/unsigned)
    TLOG_DEBUG(TLVL_WORK_STEPS) << "Data request for trig/seq_num=" << dr.trigger_number
      << "." << dr.sequence_number << " and SourceID[" << m_sourceid << "] with"
      << " Trigger TS=" << dr.trigger_timestamp
      << " Oldest stored TS=" << last_ts
      << " Newest stored TS=" << newest_ts
      << " Start of window TS=" << dr.request_information.window_begin
      << " End of window TS=" << dr.request_information.window_end
      << " Latency buffer occupancy=" << m_latency_buffer->occupancy()
      << " frag_pieces result_code=" << rres.result_code
      << " number of frag_pieces=" << frag_pieces.size();
 
    switch (rres.result_code) {
    case ResultCode::kTooOld:
        // return empty frag
            ++m_num_requests_old_window;
                ++m_num_requests_bad;
        frag_header.error_bits |= (0x1 << static_cast<size_t>(daqdataformats::FragmentErrorBits::kDataNotFound));
        break;
    case ResultCode::kPartiallyOld:
                ++m_num_requests_old_window;
                ++m_num_requests_found;
        frag_header.error_bits |= (0x1 << static_cast<size_t>(daqdataformats::FragmentErrorBits::kIncomplete));
        frag_header.error_bits |= (0x1 << static_cast<size_t>(daqdataformats::FragmentErrorBits::kDataNotFound));
                break;
    case ResultCode::kFound:
        ++m_num_requests_found;
        break;
    case ResultCode::kPartial:
                frag_header.error_bits |= (0x1 << static_cast<size_t>(daqdataformats::FragmentErrorBits::kIncomplete));
        ++m_num_requests_delayed;
                break;
    case ResultCode::kNotYet:
        frag_header.error_bits |= (0x1 << static_cast<size_t>(daqdataformats::FragmentErrorBits::kDataNotFound));
        ++m_num_requests_delayed;
        break;
    default:
        // Unknown result of data search
        ++m_num_requests_bad;
        frag_header.error_bits |= (0x1 << static_cast<size_t>(daqdataformats::FragmentErrorBits::kDataNotFound));
    }
  }
  // Create fragment from pieces
  rres.fragment = std::make_unique<daqdataformats::Fragment>(frag_pieces);
 
  // Set header
  rres.fragment->set_header_fields(frag_header);
 
  return rres;
}

dump_to_buffer()

template<class ReadoutType , class LatencyBufferType >

void dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::dump_to_buffer ( const void * data, std::size_t size, void * buffer, uint32_t buffer_pos, const std::size_t & buffer_size )

inlineprotected

Definition at line 190 of file DefaultRequestHandlerModel.hpp.

  {
    auto bytes_to_copy = size;
    while (bytes_to_copy > 0) {
      auto n = std::min(bytes_to_copy, buffer_size - buffer_pos);
      std::memcpy(static_cast<char*>(buffer) + buffer_pos, static_cast<const char*>(data), n);
      buffer_pos += n;
      bytes_to_copy -= n;
      if (buffer_pos == buffer_size) {
        buffer_pos = 0;
      }
    }
  }

generate_opmon_data()

template<class RDT , class LBT >

void dunedaq::datahandlinglibs::DefaultRequestHandlerModel< RDT, LBT >::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 283 of file DefaultRequestHandlerModel.hxx.

 {
   opmon::RequestHandlerInfo info;
 
   info.set_num_requests_handled(m_handled_requests.exchange(0));
   info.set_num_requests_found(m_num_requests_found.exchange(0));
   info.set_num_requests_bad(m_num_requests_bad.exchange(0));
   info.set_num_requests_old_window(m_num_requests_old_window.exchange(0));
   info.set_num_requests_delayed(m_num_requests_delayed.exchange(0));
   info.set_num_requests_uncategorized(m_num_requests_uncategorized.exchange(0));
   info.set_num_requests_timed_out(m_num_requests_timed_out.exchange(0));
   info.set_num_requests_waiting(m_waiting_requests.size());
 
   int new_pop_reqs = 0;
   int new_pop_count = 0;
   int new_occupancy = 0;
   info.set_tot_request_response_time(m_response_time_acc.exchange(0));
   info.set_max_request_response_time(m_response_time_max.exchange(0));
   info.set_min_request_response_time(m_response_time_min.exchange(std::numeric_limits<int>::max()));
   auto now = std::chrono::high_resolution_clock::now();
   new_pop_reqs = m_pop_reqs.exchange(0);
   new_pop_count = m_pops_count.exchange(0);
   new_occupancy = m_occupancy;
   double seconds = std::chrono::duration_cast<std::chrono::microseconds>(now - m_t0).count() / 1000000.;
   TLOG_DEBUG(TLVL_HOUSEKEEPING) << "Cleanup request rate: " << new_pop_reqs / seconds / 1. << " [Hz]"
                                 << " Dropped: " << new_pop_count << " Occupancy: " << new_occupancy;
 
   if (info.num_requests_handled() > 0) {
 
     info.set_avg_request_response_time(info.tot_request_response_time() / info.num_requests_handled());
     TLOG_DEBUG(TLVL_HOUSEKEEPING) << "Completed requests: " << info.num_requests_handled()
                                   << " | Avarage response time: " << info.avg_request_response_time() << "[us]"
                     << " | Periodic sends: " << info.num_periodic_sent();
   }
 
   m_t0 = now;
 
   info.set_num_buffer_cleanups(m_num_buffer_cleanups.exchange(0));
   info.set_num_periodic_sent(m_num_periodic_sent.exchange(0));
   info.set_num_periodic_send_failed(m_num_periodic_send_failed.exchange(0));
 
   this->publish(std::move(info));
 
   opmon::RecordingInfo rinfo;
   rinfo.set_recording_status(m_recording? "Y" : "N");
   rinfo.set_packets_recorded(m_payloads_written.exchange(0));   
   rinfo.set_bytes_recorded(m_bytes_written.exchange(0));   
   this->publish(std::move(rinfo));
 }

get_cutoff_timestamp()

template<class ReadoutType , class LatencyBufferType >

virtual dunedaq::daqdataformats::timestamp_t dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::get_cutoff_timestamp ( )

inlinevirtual

Definition at line 157 of file DefaultRequestHandlerModel.hpp.

{return 0;}

get_fragment_pieces()

template<class RDT , class LBT >

std::vector< std::pair< void *, size_t > > dunedaq::datahandlinglibs::DefaultRequestHandlerModel< RDT, LBT >::get_fragment_pieces ( uint64_t start_win_ts, uint64_t end_win_ts, RequestResult & rres )

protected

Definition at line 445 of file DefaultRequestHandlerModel.hxx.

{
 
  TLOG_DEBUG(TLVL_WORK_STEPS) << "Looking for frags between " << start_win_ts << " and " << end_win_ts;
 
  std::vector<std::pair<void*, size_t>> frag_pieces;
  // Data availability is calculated here
  auto front_element = m_latency_buffer->front();           // NOLINT
  auto last_element = m_latency_buffer->back();             // NOLINT
  uint64_t last_ts = front_element->get_timestamp();  // NOLINT(build/unsigned)
  uint64_t newest_ts = last_element->get_timestamp(); // NOLINT(build/unsigned)
 
  if (start_win_ts > newest_ts) {
  // No element is as small as the start window-> request is far in the future
     rres.result_code = ResultCode::kNotYet; // give it another chance
  }
  else if (end_win_ts < last_ts ) {
     rres.result_code = ResultCode::kTooOld;
  }
  else {
    RDT request_element = RDT();
    request_element.set_timestamp(start_win_ts-(request_element.get_num_frames() * RDT::expected_tick_difference));
    //request_element.set_timestamp(start_win_ts);
 
    auto start_iter = m_error_registry->has_error("MISSING_FRAMES")
                      ? m_latency_buffer->lower_bound(request_element, true)
                      : m_latency_buffer->lower_bound(request_element, false);
    if (!start_iter.good()) {
      // Accessor problem 
      rres.result_code = ResultCode::kNotFound;
    } 
    else {
      TLOG_DEBUG(TLVL_WORK_STEPS) << "Lower bound found " << start_iter->get_timestamp() << ", --> distance from window: " 
          << int64_t(start_win_ts) - int64_t(start_iter->get_timestamp()) ;  
      if (end_win_ts > newest_ts) {
         rres.result_code = ResultCode::kPartial;
      }
      else if (start_win_ts < last_ts) {
        rres.result_code = ResultCode::kPartiallyOld;
      }
      else {
        rres.result_code = ResultCode::kFound;
      }
 
      auto elements_handled = 0;
 
      RDT* element = &(*start_iter);
   
      while (start_iter.good() && element->get_timestamp() < end_win_ts) {
        if ( element->get_timestamp() + element->get_num_frames() * RDT::expected_tick_difference <= start_win_ts) {
        //TLOG() << "skip processing for current element " << element->get_timestamp() << ", out of readout window.";
        } 
      
        else if ( element->get_num_frames()>1 &&
         ((element->get_timestamp() < start_win_ts &&
          element->get_timestamp() + element->get_num_frames() * RDT::expected_tick_difference > start_win_ts) 
         ||
          element->get_timestamp() + element->get_num_frames() * RDT::expected_tick_difference >
            end_win_ts)) {
          //TLOG() << "We don't need the whole aggregated object (e.g.: superchunk)" ;
          for (auto frame_iter = element->begin(); frame_iter != element->end(); frame_iter++) {
            if (get_frame_iterator_timestamp(frame_iter) > (start_win_ts - RDT::expected_tick_difference)&&
                get_frame_iterator_timestamp(frame_iter) < end_win_ts ) {
              frag_pieces.emplace_back(
                std::make_pair<void*, size_t>(static_cast<void*>(&(*frame_iter)), element->get_frame_size()));
            }
          }
        }
        else {
      //TLOG() << "Add element " << element->get_timestamp();      
          // We are somewhere in the middle -> the whole aggregated object (e.g.: superchunk) can be copied
          frag_pieces.emplace_back(
            std::make_pair<void*, size_t>(static_cast<void*>((*start_iter).begin()), element->get_payload_size()));
        }
 
        elements_handled++;
        ++start_iter;
        element = &(*start_iter);
      }
    }
  }
  TLOG_DEBUG(TLVL_WORK_STEPS) << "*** Number of frames retrieved: " << frag_pieces.size();
  return frag_pieces;
}

get_oldest_time()

template<class ReadoutType , class LatencyBufferType >

std::uint64_t dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::get_oldest_time ( )

inlineoverridevirtual

Get oldest timestamp in the buffer.

Implements dunedaq::datahandlinglibs::RequestHandlerConcept< ReadoutType, LatencyBufferType >.

Definition at line 164 of file DefaultRequestHandlerModel.hpp.

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

issue_request()

template<class RDT , class LBT >

void dunedaq::datahandlinglibs::DefaultRequestHandlerModel< RDT, LBT >::issue_request ( dfmessages::DataRequest , bool = false )

overridevirtual

Issue a data request to the request handler.

Implements dunedaq::datahandlinglibs::RequestHandlerConcept< ReadoutType, LatencyBufferType >.

Reimplemented in dunedaq::datahandlinglibs::EmptyFragmentRequestHandlerModel< ReadoutType, LatencyBufferType >.

Definition at line 228 of file DefaultRequestHandlerModel.hxx.

{
  boost::asio::post(*m_request_handler_thread_pool, [&, datarequest, is_retry]() { // start a thread from pool
    auto t_req_begin = std::chrono::high_resolution_clock::now();
    {
      std::unique_lock<std::mutex> lock(m_cv_mutex);
      m_cv.wait(lock, [&] { return !m_cleanup_requested; });
      m_requests_running++;
    }
    m_cv.notify_all();
    auto result = data_request(datarequest);
    {
      std::lock_guard<std::mutex> lock(m_cv_mutex);
      m_requests_running--;
    }
    m_cv.notify_all();
    if ((result.result_code == ResultCode::kNotYet || result.result_code == ResultCode::kPartial) && m_request_timeout_ms >0 && is_retry == false) {
      TLOG_DEBUG(TLVL_WORK_STEPS) << "Re-queue request. "
                                  << " with timestamp=" << result.data_request.trigger_timestamp;
      std::lock_guard<std::mutex> wait_lock_guard(m_waiting_requests_lock);
      m_waiting_requests.push_back(RequestElement(datarequest, std::chrono::high_resolution_clock::now()));
    }
    else {
      try { // Send to fragment connection
        TLOG_DEBUG(TLVL_WORK_STEPS) << "Sending fragment with trigger/sequence_number "
          << result.fragment->get_trigger_number() << "."
          << result.fragment->get_sequence_number() << ", run number "
          << result.fragment->get_run_number() << ", and DetectorID "
          << result.fragment->get_detector_id() << ", and SourceID "
          << result.fragment->get_element_id() << ", and size "
          << result.fragment->get_size() << ", and result code "
      << result.result_code;
        // Send fragment
        get_iom_sender<std::unique_ptr<daqdataformats::Fragment>>(datarequest.data_destination)
          ->send(std::move(result.fragment), std::chrono::milliseconds(m_fragment_send_timeout_ms));
 
      } catch (const ers::Issue& excpt) {
        ers::warning(CannotWriteToQueue(ERS_HERE, m_sourceid, datarequest.data_destination, excpt));
      }
    }
 
    auto t_req_end = std::chrono::high_resolution_clock::now();
    auto us_req_took = std::chrono::duration_cast<std::chrono::microseconds>(t_req_end - t_req_begin);
    TLOG_DEBUG(TLVL_WORK_STEPS) << "Responding to data request took: " << us_req_took.count() << "[us]";
    m_response_time_acc.fetch_add(us_req_took.count());
    if ( us_req_took.count() > m_response_time_max.load() )
      m_response_time_max.store(us_req_took.count());
    if ( us_req_took.count() < m_response_time_min.load() )
      m_response_time_min.store(us_req_took.count());
    m_handled_requests++;
  });
}

periodic_cleanups()

template<class RDT , class LBT >

void dunedaq::datahandlinglibs::DefaultRequestHandlerModel< RDT, LBT >::periodic_cleanups ( )

protected

Definition at line 347 of file DefaultRequestHandlerModel.hxx.

{
  while (m_run_marker.load()) {
    cleanup_check();
    std::this_thread::sleep_for(std::chrono::milliseconds(50));
  }
}

periodic_data_transmission()

template<class RDT , class LBT >

void dunedaq::datahandlinglibs::DefaultRequestHandlerModel< RDT, LBT >::periodic_data_transmission ( )

overridevirtual

Periodic data transmission - relevant for trigger in particular.

Implements dunedaq::datahandlinglibs::RequestHandlerConcept< ReadoutType, LatencyBufferType >.

Reimplemented in dunedaq::trigger::TPRequestHandler.

Definition at line 367 of file DefaultRequestHandlerModel.hxx.

{}

periodic_data_transmissions()

template<class RDT , class LBT >

void dunedaq::datahandlinglibs::DefaultRequestHandlerModel< RDT, LBT >::periodic_data_transmissions ( )

protected

Definition at line 357 of file DefaultRequestHandlerModel.hxx.

{
 while (m_run_marker.load()) {
    periodic_data_transmission();
    std::this_thread::sleep_for(std::chrono::milliseconds(m_periodic_data_transmission_ms));
  }
}

record()

template<class RDT , class LBT >

void dunedaq::datahandlinglibs::DefaultRequestHandlerModel< RDT, LBT >::record ( const appfwk::DAQModule::CommandData_t & args )

overridevirtual

Implements dunedaq::datahandlinglibs::RequestHandlerConcept< ReadoutType, LatencyBufferType >.

Reimplemented in dunedaq::datahandlinglibs::ZeroCopyRecordingRequestHandlerModel< ReadoutType, LatencyBufferType >.

Definition at line 144 of file DefaultRequestHandlerModel.hxx.

{
  //auto conf = args.get<readoutconfig::RecordingParams>();
  //FIXME: how do we pass the duration or recording?
  int recording_time_sec = 1;
  if (m_recording.load()) {
    ers::error(CommandError(ERS_HERE, m_sourceid, "A recording is still running, no new recording was started!"));
    return;
  } else if (!m_buffered_writer.is_open()) {
    ers::error(CommandError(ERS_HERE, m_sourceid, "DLH is not configured for recording"));
    return;
  }
  m_recording_thread.set_work(
    [&](int duration) {
      TLOG() << "Start recording for " << duration << " second(s)" << std::endl;
      m_recording.exchange(true);
      auto start_of_recording = std::chrono::high_resolution_clock::now();
      auto current_time = start_of_recording;
      m_next_timestamp_to_record = 0;
      RDT element_to_search;
      while (std::chrono::duration_cast<std::chrono::seconds>(current_time - start_of_recording).count() < duration) {
        if (!m_cleanup_requested || (m_next_timestamp_to_record == 0)) {
          if (m_next_timestamp_to_record == 0) {
            auto front = m_latency_buffer->front();
            m_next_timestamp_to_record = front == nullptr ? 0 : front->get_timestamp();
          }
          element_to_search.set_timestamp(m_next_timestamp_to_record);
          size_t processed_chunks_in_loop = 0;
 
          {
            std::unique_lock<std::mutex> lock(m_cv_mutex);
            m_cv.wait(lock, [&] { return !m_cleanup_requested; });
            m_requests_running++;
          }
          m_cv.notify_all();
          auto chunk_iter = m_latency_buffer->lower_bound(element_to_search, true);
          auto end = m_latency_buffer->end();
          {
            std::lock_guard<std::mutex> lock(m_cv_mutex);
            m_requests_running--;
          }
          m_cv.notify_all();
 
          for (; chunk_iter != end && chunk_iter.good() && processed_chunks_in_loop < 1000;) {
            if ((*chunk_iter).get_timestamp() >= m_next_timestamp_to_record) {
              if (!m_buffered_writer.write(reinterpret_cast<char*>(chunk_iter->begin()), // NOLINT
                                           chunk_iter->get_payload_size())) {
                ers::warning(CannotWriteToFile(ERS_HERE, m_output_file));
              }
              m_payloads_written++;
          m_bytes_written += chunk_iter->get_payload_size();
              processed_chunks_in_loop++;
              m_next_timestamp_to_record = (*chunk_iter).get_timestamp() +
                                           RDT::expected_tick_difference * (*chunk_iter).get_num_frames();
            }
            ++chunk_iter;
          }
        }
        current_time = std::chrono::high_resolution_clock::now();
      }
      m_next_timestamp_to_record = std::numeric_limits<uint64_t>::max(); // NOLINT (build/unsigned)
 
      TLOG() << "Stop recording" << std::endl;
      m_recording.exchange(false);
      m_buffered_writer.flush();
    },
    recording_time_sec);
}

reset_oldest_time()

template<class ReadoutType , class LatencyBufferType >

void dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::reset_oldest_time ( )

inline

Definition at line 161 of file DefaultRequestHandlerModel.hpp.

{ m_oldest_timestamp.store(0); }

scrap()

template<class RDT , class LBT >

void dunedaq::datahandlinglibs::DefaultRequestHandlerModel< RDT, LBT >::scrap ( const appfwk::DAQModule::CommandData_t & )

overridevirtual

Implements dunedaq::datahandlinglibs::RequestHandlerConcept< ReadoutType, LatencyBufferType >.

Reimplemented in dunedaq::trigger::TPRequestHandler.

Definition at line 69 of file DefaultRequestHandlerModel.hxx.

{
  if (m_buffered_writer.is_open()) {
    m_buffered_writer.close();
  }
}

start()

template<class RDT , class LBT >

void dunedaq::datahandlinglibs::DefaultRequestHandlerModel< RDT, LBT >::start ( const appfwk::DAQModule::CommandData_t & )

virtual

Implements dunedaq::datahandlinglibs::RequestHandlerConcept< ReadoutType, LatencyBufferType >.

Reimplemented in dunedaq::trigger::TPRequestHandler.

Definition at line 78 of file DefaultRequestHandlerModel.hxx.

{
  // Reset opmon variables
  m_num_requests_found = 0;
  m_num_requests_bad = 0;
  m_num_requests_old_window = 0;
  m_num_requests_delayed = 0;
  m_num_requests_uncategorized = 0;
  m_num_buffer_cleanups = 0;
  m_num_requests_timed_out = 0;
  m_handled_requests = 0;
  m_response_time_acc = 0;
  m_pop_reqs = 0;
  m_pops_count = 0;
  m_payloads_written = 0;
  m_bytes_written = 0;
 
  m_t0 = std::chrono::high_resolution_clock::now();
 
  // 19-Dec-2024, KAB: check that Fragment senders are ready to send. This is done so
  // that the IOManager infrastructure fetches the necessary connection details from
  // the ConnectivityService at 'start' time, instead of the first time that the sender
  // is used to send data.  This avoids delays in the sending of the first fragment in
  // the first data-taking run in a DAQ session. Such delays can lead to undesirable
  // system behavior like trigger inhibits.
  for (auto frag_out_conn : m_frag_out_conn_ids) {
    auto sender = get_iom_sender<std::unique_ptr<daqdataformats::Fragment>>(frag_out_conn);
    if (sender != nullptr) {
      bool is_ready = sender->is_ready_for_sending(std::chrono::milliseconds(100));
      TLOG_DEBUG(0) << "The Fragment sender for " << frag_out_conn << " " << (is_ready ? "is" : "is not")
                    << " ready, my source_id is [" << m_sourceid << "]";
    }
  }
 
  m_request_handler_thread_pool = std::make_unique<boost::asio::thread_pool>(m_num_request_handling_threads);
 
  m_run_marker.store(true);
  m_cleanup_thread.set_work(&DefaultRequestHandlerModel<RDT, LBT>::periodic_cleanups, this);
  if(m_periodic_data_transmission_ms > 0) {
    m_periodic_transmission_thread.set_work(&DefaultRequestHandlerModel<RDT, LBT>::periodic_data_transmissions, this);
  }
 
  m_waiting_queue_thread = 
    std::thread(&DefaultRequestHandlerModel<RDT, LBT>::check_waiting_requests, this);
}

stop()

template<class RDT , class LBT >

void dunedaq::datahandlinglibs::DefaultRequestHandlerModel< RDT, LBT >::stop ( const appfwk::DAQModule::CommandData_t & )

virtual

Implements dunedaq::datahandlinglibs::RequestHandlerConcept< ReadoutType, LatencyBufferType >.

Definition at line 126 of file DefaultRequestHandlerModel.hxx.

{
  m_run_marker.store(false);
  while (!m_recording_thread.get_readiness()) {
    std::this_thread::sleep_for(std::chrono::milliseconds(10));
  }
  while (!m_cleanup_thread.get_readiness()) {
    std::this_thread::sleep_for(std::chrono::milliseconds(10));
  }
  while (!m_periodic_transmission_thread.get_readiness()) {
    std::this_thread::sleep_for(std::chrono::milliseconds(10));
  }
  m_waiting_queue_thread.join();
  m_request_handler_thread_pool->join();
}

supports_cutoff_timestamp()

template<class ReadoutType , class LatencyBufferType >

virtual bool dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::supports_cutoff_timestamp ( )

inlinevirtual

Definition at line 158 of file DefaultRequestHandlerModel.hpp.

{return false;}

Member Data Documentation

m_buffer_capacity

template<class ReadoutType , class LatencyBufferType >

size_t dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::m_buffer_capacity

protected

Definition at line 272 of file DefaultRequestHandlerModel.hpp.

m_buffered_writer

template<class ReadoutType , class LatencyBufferType >

BufferedFileWriter dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::m_buffered_writer

protected

Definition at line 234 of file DefaultRequestHandlerModel.hpp.

m_bytes_written

template<class ReadoutType , class LatencyBufferType >

std::atomic<int> dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::m_bytes_written { 0 }

protected

Definition at line 300 of file DefaultRequestHandlerModel.hpp.

{ 0 };

m_cleanup_requested

template<class ReadoutType , class LatencyBufferType >

std::atomic<bool> dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::m_cleanup_requested = false

protected

Definition at line 247 of file DefaultRequestHandlerModel.hpp.

m_cleanup_thread

template<class ReadoutType , class LatencyBufferType >

utilities::ReusableThread dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::m_cleanup_thread

protected

Definition at line 237 of file DefaultRequestHandlerModel.hpp.

m_configured

template<class ReadoutType , class LatencyBufferType >

bool dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::m_configured

protected

Definition at line 268 of file DefaultRequestHandlerModel.hpp.

m_cv

template<class ReadoutType , class LatencyBufferType >

std::condition_variable dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::m_cv

protected

Definition at line 246 of file DefaultRequestHandlerModel.hpp.

m_cv_mutex

template<class ReadoutType , class LatencyBufferType >

std::mutex dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::m_cv_mutex

protected

Definition at line 245 of file DefaultRequestHandlerModel.hpp.

m_detid

template<class ReadoutType , class LatencyBufferType >

uint16_t dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::m_detid

protected

Definition at line 274 of file DefaultRequestHandlerModel.hpp.

m_error_registry

template<class ReadoutType , class LatencyBufferType >

std::unique_ptr<FrameErrorRegistry>& dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::m_error_registry

protected

Definition at line 257 of file DefaultRequestHandlerModel.hpp.

m_frag_out_conn_ids

template<class ReadoutType , class LatencyBufferType >

std::vector<std::string> dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::m_frag_out_conn_ids

protected

Definition at line 281 of file DefaultRequestHandlerModel.hpp.

m_fragment_send_timeout_ms

template<class ReadoutType , class LatencyBufferType >

int dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::m_fragment_send_timeout_ms

protected

Definition at line 311 of file DefaultRequestHandlerModel.hpp.

m_handled_requests

template<class ReadoutType , class LatencyBufferType >

std::atomic<int> dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::m_handled_requests { 0 }

protected

Definition at line 295 of file DefaultRequestHandlerModel.hpp.

{ 0 };

m_latency_buffer

template<class ReadoutType , class LatencyBufferType >

std::shared_ptr<LatencyBufferType>& dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::m_latency_buffer

protected

Definition at line 231 of file DefaultRequestHandlerModel.hpp.

m_max_requested_elements

template<class ReadoutType , class LatencyBufferType >

std::size_t dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::m_max_requested_elements

protected

Definition at line 244 of file DefaultRequestHandlerModel.hpp.

m_next_timestamp_to_record

template<class ReadoutType , class LatencyBufferType >

std::atomic<uint64_t> dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::m_next_timestamp_to_record = std::numeric_limits<uint64_t>::max()

protected

Definition at line 265 of file DefaultRequestHandlerModel.hpp.

m_num_buffer_cleanups

template<class ReadoutType , class LatencyBufferType >

std::atomic<int> dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::m_num_buffer_cleanups { 0 }

protected

Definition at line 285 of file DefaultRequestHandlerModel.hpp.

{ 0 };

m_num_periodic_send_failed

template<class ReadoutType , class LatencyBufferType >

std::atomic<uint64_t> dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::m_num_periodic_send_failed { 0 }

protected

Definition at line 302 of file DefaultRequestHandlerModel.hpp.

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

m_num_periodic_sent

template<class ReadoutType , class LatencyBufferType >

std::atomic<uint64_t> dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::m_num_periodic_sent { 0 }

protected

Definition at line 301 of file DefaultRequestHandlerModel.hpp.

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

m_num_request_handling_threads

template<class ReadoutType , class LatencyBufferType >

size_t dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::m_num_request_handling_threads = 0

protected

Definition at line 254 of file DefaultRequestHandlerModel.hpp.

m_num_requests_bad

template<class ReadoutType , class LatencyBufferType >

std::atomic<int> dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::m_num_requests_bad { 0 }

protected

Definition at line 290 of file DefaultRequestHandlerModel.hpp.

{ 0 };

m_num_requests_delayed

template<class ReadoutType , class LatencyBufferType >

std::atomic<int> dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::m_num_requests_delayed { 0 }

protected

Definition at line 292 of file DefaultRequestHandlerModel.hpp.

{ 0 };

m_num_requests_found

template<class ReadoutType , class LatencyBufferType >

std::atomic<int> dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::m_num_requests_found { 0 }

protected

Definition at line 289 of file DefaultRequestHandlerModel.hpp.

{ 0 };

m_num_requests_old_window

template<class ReadoutType , class LatencyBufferType >

std::atomic<int> dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::m_num_requests_old_window { 0 }

protected

Definition at line 291 of file DefaultRequestHandlerModel.hpp.

{ 0 };

m_num_requests_timed_out

template<class ReadoutType , class LatencyBufferType >

std::atomic<int> dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::m_num_requests_timed_out { 0 }

protected

Definition at line 294 of file DefaultRequestHandlerModel.hpp.

{ 0 };

m_num_requests_uncategorized

template<class ReadoutType , class LatencyBufferType >

std::atomic<int> dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::m_num_requests_uncategorized { 0 }

protected

Definition at line 293 of file DefaultRequestHandlerModel.hpp.

{ 0 };

m_occupancy

template<class ReadoutType , class LatencyBufferType >

std::atomic<int> dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::m_occupancy

protected

Definition at line 288 of file DefaultRequestHandlerModel.hpp.

m_oldest_timestamp

template<class ReadoutType , class LatencyBufferType >

std::atomic<uint64_t> dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::m_oldest_timestamp { 0 }

protected

Definition at line 303 of file DefaultRequestHandlerModel.hpp.

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

m_output_file

template<class ReadoutType , class LatencyBufferType >

std::string dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::m_output_file

protected

Definition at line 275 of file DefaultRequestHandlerModel.hpp.

m_payloads_written

template<class ReadoutType , class LatencyBufferType >

std::atomic<int> dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::m_payloads_written { 0 }

protected

Definition at line 299 of file DefaultRequestHandlerModel.hpp.

{ 0 };

m_periodic_data_transmission_ms

template<class ReadoutType , class LatencyBufferType >

uint32_t dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::m_periodic_data_transmission_ms = 0

protected

Definition at line 280 of file DefaultRequestHandlerModel.hpp.

m_periodic_transmission_thread

template<class ReadoutType , class LatencyBufferType >

utilities::ReusableThread dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::m_periodic_transmission_thread

protected

Definition at line 238 of file DefaultRequestHandlerModel.hpp.

m_pop_counter

template<class ReadoutType , class LatencyBufferType >

std::atomic<int> dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::m_pop_counter

protected

Definition at line 284 of file DefaultRequestHandlerModel.hpp.

m_pop_limit_pct

template<class ReadoutType , class LatencyBufferType >

float dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::m_pop_limit_pct

protected

Definition at line 269 of file DefaultRequestHandlerModel.hpp.

m_pop_limit_size

template<class ReadoutType , class LatencyBufferType >

unsigned dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::m_pop_limit_size

protected

Definition at line 271 of file DefaultRequestHandlerModel.hpp.

m_pop_reqs

template<class ReadoutType , class LatencyBufferType >

std::atomic<int> dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::m_pop_reqs

protected

Definition at line 286 of file DefaultRequestHandlerModel.hpp.

m_pop_size_pct

template<class ReadoutType , class LatencyBufferType >

float dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::m_pop_size_pct

protected

Definition at line 270 of file DefaultRequestHandlerModel.hpp.

m_pops_count

template<class ReadoutType , class LatencyBufferType >

std::atomic<int> dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::m_pops_count

protected

Definition at line 287 of file DefaultRequestHandlerModel.hpp.

m_recording

template<class ReadoutType , class LatencyBufferType >

std::atomic<bool> dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::m_recording = false

protected

Definition at line 264 of file DefaultRequestHandlerModel.hpp.

m_recording_configured

template<class ReadoutType , class LatencyBufferType >

bool dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::m_recording_configured = false

protected

Definition at line 277 of file DefaultRequestHandlerModel.hpp.

m_recording_thread

template<class ReadoutType , class LatencyBufferType >

utilities::ReusableThread dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::m_recording_thread

protected

Definition at line 235 of file DefaultRequestHandlerModel.hpp.

m_request_counter

template<class ReadoutType , class LatencyBufferType >

std::map<dfmessages::DataRequest, int> dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::m_request_counter

protected

Definition at line 241 of file DefaultRequestHandlerModel.hpp.

m_request_handler_thread_pool

template<class ReadoutType , class LatencyBufferType >

std::unique_ptr<boost::asio::thread_pool> dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::m_request_handler_thread_pool

protected

Definition at line 253 of file DefaultRequestHandlerModel.hpp.

m_request_timeout_ms

template<class ReadoutType , class LatencyBufferType >

int dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::m_request_timeout_ms

private

Definition at line 313 of file DefaultRequestHandlerModel.hpp.

m_requests_running

template<class ReadoutType , class LatencyBufferType >

std::atomic<int> dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::m_requests_running = 0

protected

Definition at line 248 of file DefaultRequestHandlerModel.hpp.

m_response_time_acc

template<class ReadoutType , class LatencyBufferType >

std::atomic<int> dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::m_response_time_acc { 0 }

protected

Definition at line 296 of file DefaultRequestHandlerModel.hpp.

{ 0 };

m_response_time_max

template<class ReadoutType , class LatencyBufferType >

std::atomic<int> dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::m_response_time_max { 0 }

protected

Definition at line 298 of file DefaultRequestHandlerModel.hpp.

{ 0 };

m_response_time_min

template<class ReadoutType , class LatencyBufferType >

std::atomic<int> dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::m_response_time_min { std::numeric_limits<int>::max() }

protected

Definition at line 297 of file DefaultRequestHandlerModel.hpp.

{ std::numeric_limits<int>::max() };

m_run_marker

template<class ReadoutType , class LatencyBufferType >

std::atomic<bool> dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::m_run_marker = false

protected

Definition at line 261 of file DefaultRequestHandlerModel.hpp.

m_sourceid

template<class ReadoutType , class LatencyBufferType >

daqdataformats::SourceID dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::m_sourceid

protected

Definition at line 273 of file DefaultRequestHandlerModel.hpp.

m_stream_buffer_size

template<class ReadoutType , class LatencyBufferType >

size_t dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::m_stream_buffer_size = 0

protected

Definition at line 276 of file DefaultRequestHandlerModel.hpp.

m_t0

template<class ReadoutType , class LatencyBufferType >

std::chrono::time_point<std::chrono::high_resolution_clock> dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::m_t0

protected

Definition at line 258 of file DefaultRequestHandlerModel.hpp.

m_waiting_queue_thread

template<class ReadoutType , class LatencyBufferType >

std::thread dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::m_waiting_queue_thread

protected

Definition at line 263 of file DefaultRequestHandlerModel.hpp.

m_waiting_requests

template<class ReadoutType , class LatencyBufferType >

std::vector<RequestElement> dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::m_waiting_requests

protected

Definition at line 249 of file DefaultRequestHandlerModel.hpp.

m_waiting_requests_lock

template<class ReadoutType , class LatencyBufferType >

std::mutex dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::m_waiting_requests_lock

protected

Definition at line 250 of file DefaultRequestHandlerModel.hpp.

m_warn_about_empty_buffer

template<class ReadoutType , class LatencyBufferType >

bool dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::m_warn_about_empty_buffer = true

protected

Definition at line 279 of file DefaultRequestHandlerModel.hpp.

m_warn_on_timeout

template<class ReadoutType , class LatencyBufferType >

bool dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >::m_warn_on_timeout = true

protected

Definition at line 278 of file DefaultRequestHandlerModel.hpp.

---

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

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