dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::PostprocessScheduleAlgorithm Class Reference

#include <DataHandlingModel.hpp>

Public Member Functions
	PostprocessScheduleAlgorithm (LatencyBufferType &latency_buffer_impl, RawDataProcessorType &raw_processor_impl, uint64_t processing_delay_ticks, uint64_t post_processing_delay_min_wait, uint64_t post_processing_delay_max_wait)
int	run (bool timeout)
int	do_run (bool timeout)

Private Attributes
LatencyBufferType &	m_latency_buffer_impl
RawDataProcessorType &	m_raw_processor_impl
const uint64_t	m_processing_delay_ticks
const uint64_t	m_post_processing_delay_min_wait
const uint64_t	m_post_processing_delay_max_wait
bool	m_first_cycle
RDT	m_processed_up_to
int	m_consecutive_timeouts
const timestamp_t	m_max_wait_in_ticks
std::chrono::time_point< std::chrono::system_clock >	m_last_post_proc_time

Detailed Description

template<class ReadoutType, class RequestHandlerType, class LatencyBufferType, class RawDataProcessorType, class InputDataType = ReadoutType>
class dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::PostprocessScheduleAlgorithm

Definition at line 137 of file DataHandlingModel.hpp.

Constructor & Destructor Documentation

PostprocessScheduleAlgorithm()

template<class ReadoutType, class RequestHandlerType, class LatencyBufferType, class RawDataProcessorType, class InputDataType = ReadoutType>

dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::PostprocessScheduleAlgorithm::PostprocessScheduleAlgorithm ( LatencyBufferType & latency_buffer_impl, RawDataProcessorType & raw_processor_impl, uint64_t processing_delay_ticks, uint64_t post_processing_delay_min_wait, uint64_t post_processing_delay_max_wait )

inline

Definition at line 140 of file DataHandlingModel.hpp.

      : m_latency_buffer_impl{ latency_buffer_impl }
      , m_raw_processor_impl{ raw_processor_impl }
      , m_processing_delay_ticks{ processing_delay_ticks }
      , m_post_processing_delay_min_wait{ post_processing_delay_min_wait }
      , m_post_processing_delay_max_wait{ post_processing_delay_max_wait }
      , m_first_cycle{ true }
      , m_processed_up_to{}
      , m_last_post_proc_time{ std::chrono::system_clock::now() }
      , m_consecutive_timeouts{ 0 }
      , m_max_wait_in_ticks{ post_processing_delay_max_wait * 62500 } // FIXME: hardcoded clock frequency
    {
    }

Member Function Documentation

do_run()

template<class ReadoutType, class RequestHandlerType, class LatencyBufferType, class RawDataProcessorType, class InputDataType = ReadoutType>

int dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::PostprocessScheduleAlgorithm::do_run ( bool timeout )

inline

Definition at line 175 of file DataHandlingModel.hpp.

    {
      if (m_latency_buffer_impl.occupancy() == 0) {
        TLOG_DEBUG(TLVL_WORK_STEPS) << "Nothing to postprocess (empty buffer)";
        return 0;
      }
 
      if (m_first_cycle) {
        auto head = m_latency_buffer_impl.front();
        m_processed_up_to.set_timestamp(head->get_timestamp());
        m_first_cycle = false;
        TLOG() << "***** First pass post processing *****";
      }
 
      // Get the LB boundaries
      auto tail = m_latency_buffer_impl.back();
      auto newest_ts = tail->get_timestamp();
 
      timestamp_t end_win_ts = 0;
      std::chrono::time_point<std::chrono::system_clock> now{ std::chrono::system_clock::now() };
 
      if (timeout) {
        // Return if the last processed timestamp is greater than the newest timestamp
        // This condition occurs after a timeout
        if (m_processed_up_to.get_timestamp() >= newest_ts + 1) {
          TLOG_DEBUG(TLVL_WORK_STEPS) << "Nothing to postprocess (at or past cap)";
          return 0;
        }
 
        ++m_consecutive_timeouts;
        timestamp_t timeout_accumulated = m_consecutive_timeouts * m_max_wait_in_ticks;
 
        end_win_ts = newest_ts - m_processing_delay_ticks + timeout_accumulated;
        end_win_ts = std::min(end_win_ts, newest_ts + 1); // Cap to prevent end_win_ts from becoming unnecessarily large
      } else {
        m_consecutive_timeouts = 0;
 
        if (m_processed_up_to.get_timestamp() >= newest_ts + 1) {
          TLOG_DEBUG(TLVL_WORK_STEPS) << "Nothing to postprocess (data arrived too late, will be ignored)";
          return 0;
        }
 
        auto milliseconds = std::chrono::duration_cast<std::chrono::milliseconds>(now - m_last_post_proc_time);
 
        if (milliseconds.count() > m_post_processing_delay_min_wait) {
          if (newest_ts - m_processed_up_to.get_timestamp() > m_processing_delay_ticks) {
            end_win_ts = newest_ts - m_processing_delay_ticks;
          } else {
            TLOG_DEBUG(TLVL_WORK_STEPS) << "Not ready to postprocess (m_processing_delay_ticks is greater)";
            return 0;
          }
        } else {
          TLOG_DEBUG(TLVL_WORK_STEPS) << "Not ready to postprocess (too fast)";
          return 0;
        }
      }
 
      auto start_iter = m_latency_buffer_impl.lower_bound(m_processed_up_to, false);
      m_processed_up_to.set_timestamp(end_win_ts);
      auto end_iter = m_latency_buffer_impl.lower_bound(m_processed_up_to, false);
 
      // This likely happens when RDT uses a composite key
      // The current algorithm does not support composite keys
      // Our search item `m_processed_up_to` will have its other keys set to their defaults
      // E.g., for TriggerPrimitive, channel = INVALID_TP_CHANNEL
      // Even if an entry with the same ts exists in the buffer, its channel will be a valid (smaller) value,
      // so `lower_bound` will not be able to find it
      // We should verify that this is the only scenario in which we end up here
      if (!start_iter.good()) {
        TLOG_DEBUG(TLVL_WORK_STEPS) << "Nothing to postprocess (!start_iter.good())";
        return 0;
      }
 
      if (start_iter == end_iter) {
        TLOG_DEBUG(TLVL_WORK_STEPS) << "Nothing to postprocess (start_iter == end_iter)";
        return 0;
      }
 
      int processed = 0;
      for (auto it = start_iter; it != end_iter; ++it) {
        // Just to be completely safe
        // We should understand why we end up here
        if (!it.good()) {
          TLOG_DEBUG(TLVL_WORK_STEPS) << "Invalid iterator in postprocessing loop";
          break;
        }
        m_raw_processor_impl.postprocess_item(&(*it));
        ++processed;
      }
 
      m_last_post_proc_time = now;
 
      return processed;
    }

run()

template<class ReadoutType, class RequestHandlerType, class LatencyBufferType, class RawDataProcessorType, class InputDataType = ReadoutType>

int dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::PostprocessScheduleAlgorithm::run ( bool timeout )

inline

Definition at line 160 of file DataHandlingModel.hpp.

    {
      int processed = this->do_run(timeout);
 
      if (timeout) {
        timestamp_t timeout_accumulated = m_consecutive_timeouts * m_max_wait_in_ticks;
        m_raw_processor_impl.invoke_postprocess_schedule_timeout_policy(timeout_accumulated);
      }
 
      return processed;
    }

Member Data Documentation

m_consecutive_timeouts

template<class ReadoutType, class RequestHandlerType, class LatencyBufferType, class RawDataProcessorType, class InputDataType = ReadoutType>

int dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::PostprocessScheduleAlgorithm::m_consecutive_timeouts

private

Definition at line 278 of file DataHandlingModel.hpp.

m_first_cycle

template<class ReadoutType, class RequestHandlerType, class LatencyBufferType, class RawDataProcessorType, class InputDataType = ReadoutType>

bool dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::PostprocessScheduleAlgorithm::m_first_cycle

private

Definition at line 276 of file DataHandlingModel.hpp.

m_last_post_proc_time

template<class ReadoutType, class RequestHandlerType, class LatencyBufferType, class RawDataProcessorType, class InputDataType = ReadoutType>

std::chrono::time_point<std::chrono::system_clock> dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::PostprocessScheduleAlgorithm::m_last_post_proc_time

private

Definition at line 280 of file DataHandlingModel.hpp.

m_latency_buffer_impl

template<class ReadoutType, class RequestHandlerType, class LatencyBufferType, class RawDataProcessorType, class InputDataType = ReadoutType>

LatencyBufferType& dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::PostprocessScheduleAlgorithm::m_latency_buffer_impl

private

Definition at line 271 of file DataHandlingModel.hpp.

m_max_wait_in_ticks

template<class ReadoutType, class RequestHandlerType, class LatencyBufferType, class RawDataProcessorType, class InputDataType = ReadoutType>

const timestamp_t dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::PostprocessScheduleAlgorithm::m_max_wait_in_ticks

private

Definition at line 279 of file DataHandlingModel.hpp.

m_post_processing_delay_max_wait

template<class ReadoutType, class RequestHandlerType, class LatencyBufferType, class RawDataProcessorType, class InputDataType = ReadoutType>

const uint64_t dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::PostprocessScheduleAlgorithm::m_post_processing_delay_max_wait

private

Definition at line 275 of file DataHandlingModel.hpp.

m_post_processing_delay_min_wait

template<class ReadoutType, class RequestHandlerType, class LatencyBufferType, class RawDataProcessorType, class InputDataType = ReadoutType>

const uint64_t dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::PostprocessScheduleAlgorithm::m_post_processing_delay_min_wait

private

Definition at line 274 of file DataHandlingModel.hpp.

m_processed_up_to

template<class ReadoutType, class RequestHandlerType, class LatencyBufferType, class RawDataProcessorType, class InputDataType = ReadoutType>

RDT dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::PostprocessScheduleAlgorithm::m_processed_up_to

private

Definition at line 277 of file DataHandlingModel.hpp.

m_processing_delay_ticks

template<class ReadoutType, class RequestHandlerType, class LatencyBufferType, class RawDataProcessorType, class InputDataType = ReadoutType>

const uint64_t dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::PostprocessScheduleAlgorithm::m_processing_delay_ticks

private

Definition at line 273 of file DataHandlingModel.hpp.

m_raw_processor_impl

template<class ReadoutType, class RequestHandlerType, class LatencyBufferType, class RawDataProcessorType, class InputDataType = ReadoutType>

RawDataProcessorType& dunedaq::datahandlinglibs::DataHandlingModel< ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType >::PostprocessScheduleAlgorithm::m_raw_processor_impl

private

Definition at line 272 of file DataHandlingModel.hpp.

---

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

• /github/workspace/dunedaq/sourcecode/datahandlinglibs/include/datahandlinglibs/models/DataHandlingModel.hpp