Delayed postprocessing timeout logic for 5.4.x line

CMakeLists.txt

@@ -76,13 +76,15 @@ daq_add_application(datahandlinglibs_test_bufferedfilewriter test_bufferedfilewr
 daq_add_application(datahandlinglibs_test_bufferedfilereader test_bufferedfilereader_app.cxx TEST LINK_LIBRARIES datahandlinglibs ${BOOST_LIBS})
 daq_add_application(datahandlinglibs_test_skiplist test_skiplist_app.cxx TEST LINK_LIBRARIES datahandlinglibs ${BOOST_LIBS})
 daq_add_application(datahandlinglibs_test_composite_key test_composite_key_app.cxx TEST LINK_LIBRARIES datahandlinglibs ${BOOST_LIBS})
+daq_add_application(datahandlinglibs_test_co_timeout test_co_timeout_app.cxx TEST LINK_LIBRARIES datahandlinglibs ${BOOST_LIBS})
 
 ##############################################################################
 # Unit Tests
 
 #daq_add_unit_test(datahandlinglibs_BufferedReadWrite_test LINK_LIBRARIES datahandlinglibs ${BOOST_LIBS})
 #daq_add_unit_test(datahandlinglibs_VariableSizeElementQueue_test LINK_LIBRARIES datahandlinglibs ${BOOST_LIBS})
 daq_add_unit_test(datahandlinglibs_DataMoveCallbackRegistry_test LINK_LIBRARIES datahandlinglibs ${BOOST_LIBS})
+daq_add_unit_test(datahandlinglibs_DataHandlingModel_test LINK_LIBRARIES datahandlinglibs ${BOOST_LIBS})
 
 ##############################################################################
 # Installation

include/datahandlinglibs/ReadoutTypes.hpp

@@ -8,6 +8,8 @@
 #ifndef DATAHANDLINGLIBS_INCLUDE_DATAHANDLINGLIBS_READOUTTYPES_HPP_
 #define DATAHANDLINGLIBS_INCLUDE_DATAHANDLINGLIBS_READOUTTYPES_HPP_
 
+#include "daqdataformats/FragmentHeader.hpp"
+
 #include <cstdint> // uint_t types
 #include <memory>  // unique_ptr
 #include <tuple>   // std::tie
@@ -43,6 +45,12 @@ struct DUMMY_FRAME_STRUCT
     return timestamp;
   }
 
+  size_t get_num_frames() const { return frames_per_element; }
+
+  size_t get_frame_size() const { return frame_size; }
+
+  size_t get_payload_size() const { return get_num_frames() * get_frame_size(); }
+
   void set_another_key(uint64_t compkey)
   {
     another_key = compkey; 
@@ -65,6 +73,11 @@ struct DUMMY_FRAME_STRUCT
   static const constexpr size_t frame_size = DUMMY_FRAME_SIZE;
   static const constexpr uint8_t frames_per_element = 1; // NOLINT(build/unsigned)
   static const constexpr size_t element_size = DUMMY_FRAME_SIZE;
+  static const constexpr dunedaq::daqdataformats::SourceID::Subsystem subsystem =
+    dunedaq::daqdataformats::SourceID::Subsystem::kUnknown;
+  static const constexpr dunedaq::daqdataformats::FragmentType fragment_type =
+    dunedaq::daqdataformats::FragmentType::kUnknown;
+  static const constexpr uint64_t expected_tick_difference = 1; // NOLINT(build/unsigned)
 };
 
 } // namespace types

include/datahandlinglibs/concepts/LatencyBufferConcept.hpp

@@ -13,10 +13,14 @@
 #include "opmonlib/MonitorableObject.hpp"
 
 #include <cstddef>
+#include <concepts>
 
 namespace dunedaq {
 namespace datahandlinglibs {
 
+template<typename T>
+concept SupportsDelayedPostprocessing = T::supports_delayed_postprocessing;
+
 /**
  * Concept of a LatencyBuffer.
  *

include/datahandlinglibs/concepts/RawDataProcessorConcept.hpp

@@ -47,6 +47,9 @@ class RawDataProcessorConcept : public opmonlib::MonitorableObject
   virtual void preprocess_item(ReadoutType* item) = 0;
   //! Postprocess one element
   virtual void postprocess_item(const ReadoutType* item) = 0;
+  //! Handle postprocess timeout event
+  virtual void invoke_postprocess_schedule_timeout_policy(std::uint64_t accumulated_timeout_ticks) = 0;
+
 };
 
 } // namespace datahandlinglibs

include/datahandlinglibs/models/DataHandlingModel.hpp

@@ -47,8 +47,9 @@
 
 #include <folly/coro/Baton.h>
 #include <folly/coro/Task.h>
-#include <folly/futures/ThreadWheelTimekeeper.h>
+#include <folly/futures/Future.h>
 
+#include <algorithm>
 #include <functional>
 #include <memory>
 #include <string>
@@ -133,6 +134,126 @@ class DataHandlingModel : public DataHandlingConcept
   std::function<void(IDT&&)> m_consume_callback;
 
 protected:
+  class PostprocessScheduleAlgorithm
+  {
+  public:
+    PostprocessScheduleAlgorithm(LatencyBufferType& latency_buffer_impl,
+                                 RawDataProcessorType& raw_processor_impl,
+                                 uint64_t processing_delay_ticks, // NOLINT(build/unsigned)
+                                 uint64_t post_processing_delay_min_wait, // NOLINT(build/unsigned)
+                                 uint64_t post_processing_delay_max_wait) // NOLINT(build/unsigned)
+      : 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
+    {
+    }
+
+    // High-level interface
+    // Schedule deferred post-processing and notify timeout expiration to the processor
+    int run(bool timeout) {
+      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;
+    }
+
+
+    // Deferral of the post processing, to allow elements being reordered in the LB
+    // Basically, find data older than a certain timestamp and process all data since the last post-processed element up to that value
+    int do_run(bool timeout)
+    {
+      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;
+        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);
+
+      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) {
+        m_raw_processor_impl.postprocess_item(&(*it));
+        ++processed;
+      }
+
+      m_last_post_proc_time = now;
+
+      return processed;
+    }  
+
+  private:
+    LatencyBufferType& m_latency_buffer_impl;
+    RawDataProcessorType& m_raw_processor_impl;
+    const uint64_t m_processing_delay_ticks; // NOLINT(build/unsigned)
+    const uint64_t m_post_processing_delay_min_wait; // NOLINT(build/unsigned)
+    const uint64_t m_post_processing_delay_max_wait; // NOLINT(build/unsigned)
+    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;
+  };
 
   // Perform processing operations on payload
   void process_item(RDT&& payload);
@@ -164,6 +285,12 @@ class DataHandlingModel : public DataHandlingConcept
     return { reinterpret_cast<RDT&>(original) };
   }
 
+  // Actions postprocess scheduler takes if no data arrives in a configured time
+  virtual void invoke_postprocess_schedule_timeout_policy() const
+  {
+    return; // No-op for this class
+  }    
+
   // Operational monitoring
   virtual void generate_opmon_data() override;
 
@@ -178,9 +305,9 @@ class DataHandlingModel : public DataHandlingConcept
   int m_current_fake_trigger_id;
   daqdataformats::SourceID m_sourceid;
   daqdataformats::run_number_t m_run_number;
-  uint64_t m_processing_delay_ticks;
-  uint64_t m_post_processing_delay_min_wait;
-  uint64_t m_post_processing_delay_max_wait;
+  uint64_t m_processing_delay_ticks; // NOLINT(build/unsigned)
+  uint64_t m_post_processing_delay_min_wait; // NOLINT(build/unsigned)
+  uint64_t m_post_processing_delay_max_wait; // NOLINT(build/unsigned)
 
   // STATS
   using metric_t = dunedaq::datahandlinglibs::opmon::DataHandlerInfo;
@@ -230,7 +357,7 @@ class DataHandlingModel : public DataHandlingConcept
   // POSTPROCESS SCHEDULER
   utilities::ReusableThread m_postprocess_scheduler_thread;
   folly::coro::Baton m_baton;
-  std::unique_ptr<folly::ThreadWheelTimekeeper> m_timekeeper;
+  std::unique_ptr<folly::Timekeeper> m_timekeeper;
 
   // LATENCY BUFFER
   std::shared_ptr<LatencyBufferType> m_latency_buffer_impl;

include/datahandlinglibs/models/SkipListLatencyBufferModel.hpp

@@ -35,6 +35,8 @@ class SkipListLatencyBufferModel : public LatencyBufferConcept<T>
   using SkipListTAcc = typename folly::ConcurrentSkipList<T>::Accessor; // SKL Accessor
   using SkipListTSkip = typename folly::ConcurrentSkipList<T>::Skipper; // Skipper accessor
 
+  static constexpr bool supports_delayed_postprocessing = true;
+
   // Constructor
   SkipListLatencyBufferModel()
     : m_skip_list(folly::ConcurrentSkipList<T>::createInstance(unconfigured_head_height))

include/datahandlinglibs/models/TaskRawDataProcessorModel.hpp

@@ -80,6 +80,9 @@ class TaskRawDataProcessorModel : public RawDataProcessorConcept<ReadoutType>
   // Registers ReadoutType item pointer to to the post-processing queue
   void postprocess_item(const ReadoutType* item) override;
 
+  // Handle a timeout event 
+  void invoke_postprocess_schedule_timeout_policy(std::uint64_t accumilated_timeout_ticks) override {}
+
   // Registers a pre-processing task to the pre-processor pipeline
   template<typename Task>
   void add_preprocess_task(Task&& task);

include/datahandlinglibs/models/detail/DataHandlingModel.hxx

@@ -2,6 +2,9 @@
 
 #include <folly/coro/BlockingWait.h>
 #include <folly/coro/Timeout.h>
+#include <folly/futures/ThreadWheelTimekeeper.h>
+#include <folly/coro/CurrentExecutor.h>
+#include <folly/CancellationToken.h>
 
 #include <typeinfo>
 
@@ -85,6 +88,13 @@ DataHandlingModel<RDT, RHT, LBT, RPT, IDT>::init(const appmodel::DataHandlerModu
   m_post_processing_delay_min_wait = mcfg->get_module_configuration()->get_post_processing_delay_min_wait();
   m_post_processing_delay_max_wait = mcfg->get_module_configuration()->get_post_processing_delay_max_wait();
 
+  if (m_processing_delay_ticks) {
+    if constexpr (!SupportsDelayedPostprocessing<LBT>) {
+      ers::error(ConfigurationError(ERS_HERE, m_sourceid,
+        "Delayed postprocessing (post_processing_delay_ticks > 0) requires a sorted buffer (SkipList). "
+        "Queue buffers (FixedRateQueue, BinarySearchQueue) expect in-order data and must use post_processing_delay_ticks = 0."));
+    }
+  }
 
   // Configure implementations:
   m_raw_processor_impl->conf(mcfg);
@@ -310,66 +320,52 @@ DataHandlingModel<RDT, RHT, LBT, RPT, IDT>::run_consume()
 
 template<class RDT, class RHT, class LBT, class RPT, class IDT>
 folly::coro::Task<void>
-DataHandlingModel<RDT, RHT, LBT, RPT, IDT>::postprocess_schedule() {  
+DataHandlingModel<RDT, RHT, LBT, RPT, IDT>::postprocess_schedule()
+{
 
   TLOG_DEBUG(TLVL_WORK_STEPS) << "Postprocess schedule coroutine started...";
-  timestamp_t newest_ts = 0;
-  timestamp_t end_win_ts = 0;
-  bool first_cycle = true;
-  auto last_post_proc_time = std::chrono::system_clock::now();
-  auto now = last_post_proc_time;
-  std::chrono::milliseconds milliseconds;
-  RDT processed_element;
-
-  // Deferral of the post processing, to allow elements being reordered in the LB
-  // Basically, find data older than a certain timestamp and process all data since the last post-processed element up to that value  
-  while (m_run_marker.load()) {
-    try {
-      co_await folly::coro::timeout(
-        m_baton.operator co_await(),
-        std::chrono::milliseconds{m_post_processing_delay_max_wait},
-        m_timekeeper.get());
-      m_baton.reset();
-    } catch (const folly::FutureTimeout&) {
-      ++m_num_post_processing_delay_max_waits;
-    }
-
-    if (m_latency_buffer_impl->occupancy() == 0) {
-      continue;
-    }
-
-    now = std::chrono::system_clock::now();
-    milliseconds = std::chrono::duration_cast<std::chrono::milliseconds>(now - last_post_proc_time);
+  TLOG() << "***** Starting post-process coroutine with timout " << m_post_processing_delay_max_wait << " *****";
+
+
+  PostprocessScheduleAlgorithm sched_algo{ *m_latency_buffer_impl,
+                                           *m_raw_processor_impl,
+                                           m_processing_delay_ticks,
+                                           m_post_processing_delay_min_wait,
+                                           m_post_processing_delay_max_wait };
+
+  const auto wait_data = [this]() -> folly::coro::Task<void> {
+    // folly::coro::timeout cancels the task on timeout.
+    // Baton is not cancellable, so we attach a callback to resume the coroutine.
+    auto token = co_await folly::coro::co_current_cancellation_token;
+    folly::CancellationCallback cb(token, [this] { m_baton.post(); });
+    co_await m_baton; // Wait data
+  };
 
-    if (milliseconds.count() <= m_post_processing_delay_min_wait) {
-      continue;
+  while (m_run_marker.load()) {
+    bool timeout = false;
+
+    if ( m_post_processing_delay_max_wait > 0 ) {
+      try {
+        co_await folly::coro::timeout(
+          wait_data(),
+          std::chrono::milliseconds{ m_post_processing_delay_max_wait },
+          m_timekeeper.get());
+
+      } catch (const folly::FutureTimeout&) {
+        timeout = true;
+        ++m_num_post_processing_delay_max_waits;
+      }
+    } else {
+      co_await m_baton;
     }
 
-    last_post_proc_time = now;
-
-    // Get the LB boundaries
-    auto tail = m_latency_buffer_impl->back();
-    newest_ts = tail->get_timestamp();
+    m_baton.reset();
 
-    if (first_cycle) {
-      auto head = m_latency_buffer_impl->front();
-      processed_element.set_timestamp(head->get_timestamp());
-      first_cycle = false;
-      TLOG() << "***** First pass post processing *****";
-    }
+    if (auto processed = sched_algo.run(timeout); processed > 0) {
+      m_num_payloads += processed;
+      m_sum_payloads += processed;
+      m_stats_packet_count += processed;
 
-    if (newest_ts - processed_element.get_timestamp() > m_processing_delay_ticks) {
-      end_win_ts = newest_ts - m_processing_delay_ticks;
-      auto start_iter = m_latency_buffer_impl->lower_bound(processed_element, false);
-      processed_element.set_timestamp(end_win_ts);
-      auto end_iter = m_latency_buffer_impl->lower_bound(processed_element, false);
-
-      for (auto it = start_iter; it != end_iter; ++it) {
-        m_raw_processor_impl->postprocess_item(&(*it));
-        ++m_num_payloads;
-        ++m_sum_payloads;
-        ++m_stats_packet_count;
-      }
     }
   }
 }