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;
 };
 
 } // namespace types

include/datahandlinglibs/models/DataHandlingModel.hpp

@@ -132,6 +132,102 @@ class DataHandlingModel : public DataHandlingConcept
   std::function<void(IDT&&)> m_consume_callback;
 
 protected:
+  class PostprocessManager {
+  public:
+    PostprocessManager(
+      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) : 
+      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_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 perform_postprocessing(bool timeout) {
+      if (m_latency_buffer_impl.occupancy() == 0) {
+        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;
+
+      if (timeout) {      
+        ++m_consecutive_timeouts;
+        timestamp_t timeout_accumulated = m_consecutive_timeouts * m_max_wait_in_ticks;  
+
+        // Cap to prevent end_win_ts from becoming unnecessarily large
+        timestamp_t timeout_cap = newest_ts + 1;
+        timeout_accumulated = std::min(timeout_accumulated, timeout_cap);  
+
+        end_win_ts = newest_ts - m_processing_delay_ticks + timeout_accumulated;
+
+      } else {
+        m_consecutive_timeouts = 0;
+        now = std::chrono::system_clock::now();
+        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;
+          }
+        }
+      }
+
+      if (end_win_ts == 0) {
+        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";
+        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;
+    const uint64_t m_post_processing_delay_min_wait;
+    const uint64_t m_post_processing_delay_max_wait;  
+    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 +259,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;
 

include/datahandlinglibs/models/detail/DataHandlingModel.hxx

@@ -315,63 +315,31 @@ folly::coro::Task<void>
 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  
+  PostprocessManager manager{
+    *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()) {
-    try {
+    bool timeout = false;
+
+     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&) {
+      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 = manager.perform_postprocessing(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/testutils/UnitTestUtilities.hpp

@@ -0,0 +1,32 @@
+#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::PostprocessManager;
+};
+
+}
+}
+}
+
+#endif // DATAHANDLINGLIBS_INCLUDE_DATAHANDLINGLIBS_TESTUTILS_UNITTESTUTILITIES_HPP

unittest/datahandlinglibs_DataHandlingModel_test.cxx

@@ -0,0 +1,70 @@
+/**
+ * @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/testutils/UnitTestUtilities.hpp"
+#include "datahandlinglibs/ReadoutTypes.hpp"
+#include "datahandlinglibs/models/SkipListLatencyBufferModel.hpp"
+
+BOOST_AUTO_TEST_SUITE(datahandlinglibs_DataHandlingModel_test)
+
+using namespace dunedaq::datahandlinglibs;
+
+using ReadoutType = types::DUMMY_FRAME_STRUCT;
+
+BOOST_AUTO_TEST_CASE(DataHandlingModel_postprocess_schedule_SkipListLatencyBufferModel)
+{
+  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;
+  const uint64_t delay_min_wait = 1;
+  const uint64_t delay_max_wait = 2;
+
+  typename decltype(model)::PostprocessManager manager{
+    *buffer, *raw_processor, delay_ticks, delay_min_wait, delay_max_wait};     
+
+  // First pass
+  bool timeout = false;
+  int processed_count = manager.perform_postprocessing(timeout);
+  BOOST_REQUIRE_EQUAL(processed_count, 0);
+
+  timeout = true;
+  processed_count += manager.perform_postprocessing(timeout);
+  BOOST_REQUIRE_EQUAL(processed_count, 2);
+
+  processed_count += manager.perform_postprocessing(timeout);
+  BOOST_REQUIRE_EQUAL(processed_count, 4);
+
+  processed_count += manager.perform_postprocessing(timeout);
+  BOOST_REQUIRE_EQUAL(processed_count, 5);
+}
+
+BOOST_AUTO_TEST_SUITE_END()