Delayed postprocessing timeout logic

CMakeLists.txt

@@ -83,6 +83,7 @@ daq_add_application(datahandlinglibs_test_composite_key test_composite_key_app.c
 #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/models/DataHandlingModel.hpp

@@ -49,6 +49,7 @@
 #include <folly/coro/Task.h>
 #include <folly/futures/ThreadWheelTimekeeper.h>
 
+#include <algorithm>
 #include <functional>
 #include <memory>
 #include <string>
@@ -132,6 +133,105 @@ 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_unprocessed_element{}
+      , 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 }
+    {
+    }
+
+    // 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 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_unprocessed_element.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) {      
+        ++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_unprocessed_element.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_unprocessed_element, false);
+      m_unprocessed_element.set_timestamp(end_win_ts);
+      auto end_iter = m_latency_buffer_impl.lower_bound(m_unprocessed_element, 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_unprocessed_element;
+    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);
@@ -163,6 +263,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;
 
@@ -177,9 +283,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;

include/datahandlinglibs/models/detail/DataHandlingModel.hxx

@@ -312,66 +312,38 @@ 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  
+  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 };
+
   while (m_run_marker.load()) {
+    bool timeout = false;
+
     try {
       co_await folly::coro::timeout(
         m_baton.operator co_await(),
-        std::chrono::milliseconds{m_post_processing_delay_max_wait},
+        std::chrono::milliseconds{ m_post_processing_delay_max_wait },
         m_timekeeper.get());
       m_baton.reset();
     } catch (const folly::FutureTimeout&) {
+      timeout = true;
       ++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);
-
-    if (milliseconds.count() <= m_post_processing_delay_min_wait) {
-      continue;
-    }
-
-    last_post_proc_time = now;
-
-    // Get the LB boundaries
-    auto tail = m_latency_buffer_impl->back();
-    newest_ts = tail->get_timestamp();
+    if (auto processed = sched_algo.run(timeout); processed > 0) {
+      m_num_payloads += processed;
+      m_sum_payloads += processed;
+      m_stats_packet_count += processed;
 
-    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 (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;
-      }
+      if (timeout) {
+        invoke_postprocess_schedule_timeout_policy();
+      }      
     }
   }
 }

include/datahandlinglibs/models/detail/SkipListLatencyBufferModel.hxx

@@ -1,5 +1,7 @@
 // Declarations for SkipListLatencyBufferModel
 
+#include "datahandlinglibs/opmon/datahandling_info.pb.h"
+
 namespace dunedaq {
 namespace datahandlinglibs {
 

include/datahandlinglibs/testutils/UnitTestUtilities.hpp

@@ -0,0 +1,39 @@
+/**
+ * @file UnitTestUtilities.hpp Unit test helper classes
+ *
+ * This is part of the DUNE DAQ Application Framework, copyright 2020.
+ * Licensing/copyright details are in the COPYING file that you should have
+ * received with this code.
+ */
+
+#ifndef DATAHANDLINGLIBS_INCLUDE_DATAHANDLINGLIBS_TESTUTILS_UNITTESTUTILITIES_HPP
+#define DATAHANDLINGLIBS_INCLUDE_DATAHANDLINGLIBS_TESTUTILS_UNITTESTUTILITIES_HPP
+
+#include "datahandlinglibs/models/DataHandlingModel.hpp"
+#include "datahandlinglibs/models/DefaultRequestHandlerModel.hpp"
+#include "datahandlinglibs/models/TaskRawDataProcessorModel.hpp"
+
+namespace dunedaq {
+namespace datahandlinglibs {
+namespace unittest {
+
+template<typename ReadoutType,
+         typename RequestHandlerType,
+         typename LatencyBufferType,
+         typename RawDataProcessorType,
+         typename InputDataType = ReadoutType>
+class MockDataHandlingModel
+  : public DataHandlingModel<ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType>
+{
+public:
+  using Base =
+    DataHandlingModel<ReadoutType, RequestHandlerType, LatencyBufferType, RawDataProcessorType, InputDataType>;
+  using Base::Base;
+  using Base::PostprocessScheduleAlgorithm;
+};
+
+} // namespace unittest
+} // namespace datahandlinglibs
+} // namespace dunedaq
+
+#endif // DATAHANDLINGLIBS_INCLUDE_DATAHANDLINGLIBS_TESTUTILS_UNITTESTUTILITIES_HPP

unittest/datahandlinglibs_DataHandlingModel_test.cxx

@@ -0,0 +1,82 @@
+/**
+ * @file datahandlinglibs_DataHandlingModel_test.cxx Unit Tests for DataHandlingModel
+ *
+ * This is part of the DUNE DAQ Application Framework, copyright 2020.
+ * Licensing/copyright details are in the COPYING file that you should have
+ * received with this code.
+ */
+
+#define BOOST_TEST_MODULE datahandlinglibs_DataHandlingModel_test // NOLINT
+
+#include "boost/test/unit_test.hpp"
+
+#include "datahandlinglibs/ReadoutTypes.hpp"
+#include "datahandlinglibs/models/SkipListLatencyBufferModel.hpp"
+#include "datahandlinglibs/testutils/UnitTestUtilities.hpp"
+
+#include <memory>
+#include <utility>
+
+BOOST_AUTO_TEST_SUITE(datahandlinglibs_DataHandlingModel_test)
+
+using namespace dunedaq::datahandlinglibs;
+
+using ReadoutType = types::DUMMY_FRAME_STRUCT;
+
+BOOST_AUTO_TEST_CASE(datahandlinglibs_DataHandlingModel_PostprocessScheduleAlgorithm_timeout)
+{
+  std::atomic<bool> run_marker = true;
+
+  auto model =
+    unittest::MockDataHandlingModel<ReadoutType,
+                                    DefaultRequestHandlerModel<ReadoutType, SkipListLatencyBufferModel<ReadoutType>>,
+                                    SkipListLatencyBufferModel<ReadoutType>,
+                                    TaskRawDataProcessorModel<ReadoutType>>(run_marker);
+
+  auto buffer = std::make_shared<SkipListLatencyBufferModel<ReadoutType>>();
+
+  for (int i = 1; i < 6; i++) {
+    ReadoutType frame{};
+    frame.timestamp = i * 62500;
+    buffer->write(std::move(frame));
+  }
+
+  const bool post_processing_enabled = true;
+  auto error_registry = std::make_unique<FrameErrorRegistry>();
+
+  auto raw_processor =
+    std::make_shared<TaskRawDataProcessorModel<ReadoutType>>(error_registry, post_processing_enabled);
+
+  const uint64_t delay_ticks = 4 * 62500; // NOLINT(build/unsigned)
+  const uint64_t delay_min_wait = 1; // NOLINT(build/unsigned)
+  const uint64_t delay_max_wait = 2; // NOLINT(build/unsigned)
+
+  typename decltype(model)::PostprocessScheduleAlgorithm sched_algo{
+    *buffer, *raw_processor, delay_ticks, delay_min_wait, delay_max_wait
+  };
+
+  // First pass
+  bool timeout = false;
+  int processed_count = sched_algo.run(timeout);
+  // Buffer = {1, 2, 3, 4, 5} delay_ticks = 4
+  // 5 - 1 > 4 is false => no postprocessing
+  BOOST_REQUIRE_EQUAL(processed_count, 0);
+
+  timeout = true;
+  // 1st timeout => timeout_accumulated = 1 * 2 (delay_max_wait = 2)
+  // end_win_ts = 5 - 4 + 2 => postprocess until 3 {1, 2}
+  processed_count += sched_algo.run(timeout);
+  BOOST_REQUIRE_EQUAL(processed_count, 2);
+
+  // 2nd timeout => timeout_accumulated = 2 * 2
+  // end_win_ts = 5 - 4 + 4 => postprocess until 5 {3, 4}
+  processed_count += sched_algo.run(timeout);
+  BOOST_REQUIRE_EQUAL(processed_count, 4);
+
+  // 3rd timeout => timeout_accumulated = 3 * 2
+  // end_win_ts = 5 - 4 + 6 => postprocess until 6 (capped to newest_ts + 1) {5}
+  processed_count += sched_algo.run(timeout);
+  BOOST_REQUIRE_EQUAL(processed_count, 5);
+}
+
+BOOST_AUTO_TEST_SUITE_END()