Fix state tracking for sequenced triggers. Improve logging

integtest/insufficient_disk_space_test.py

@@ -60,7 +60,8 @@
         r"There are \d+ bytes free, and the required minimum is \d+ bytes based on a safety factor of \d+ times the trigger record size",
     ],
     "mlt": [r"Trigger is inhibited in run \d+"],
-    "dfo": [r"TriggerDecision \d+ didn't complete within timeout in run \d+"],
+    # TODO: Eric Flumerfelt <eflumerf@github.com>, Nov-06-2026: Removing this message, see Issue https://github.com/DUNE-DAQ/dfmodules/issues/525
+    #"dfo": [r"TriggerDecision \d+ didn't complete within timeout in run \d+"],
 }
 ignored_logfile_problems = {
     "connectivity-service": [

plugins/DFOModule.cpp

@@ -42,8 +42,9 @@ enum
   TLVL_WORK_STEPS = 10,
   TLVL_TRIGDEC_RECEIVED = 21,
   TLVL_NOTIFY_TRIGGER = 22,
-  TLVL_DISPATCH_TO_TRB = 23,
-  TLVL_TDTOKEN_RECEIVED = 24
+  TLVL_NOTIFY_TRIGGER_VERBOSE = 23,
+  TLVL_DISPATCH_TO_TRB = 24,
+  TLVL_TDTOKEN_RECEIVED = 25
 };
 
 namespace dunedaq::dfmodules {
@@ -55,7 +56,8 @@ DFOModule::DFOModule(const std::string& name)
 {
   register_command("conf", &DFOModule::do_conf);
   register_command("start", &DFOModule::do_start);
-  register_command("drain_dataflow", &DFOModule::do_stop);
+  register_command("drain_dataflow", &DFOModule::do_drain);
+  register_command("stop", &DFOModule::do_stop);
   register_command("scrap", &DFOModule::do_scrap);
 }
 
@@ -170,15 +172,24 @@ DFOModule::do_start(const CommandData_t& payload)
 }
 
 void
-DFOModule::do_stop(const CommandData_t& /*args*/)
+DFOModule::do_drain(const CommandData_t& /*args*/)
 {
-  TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Entering do_stop() method";
+  TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Entering do_drain() method";
 
   m_running_status.store(false);
 
   auto iom = iomanager::IOManager::get();
   iom->remove_callback<dfmessages::TriggerDecision>(m_td_connection);
 
+  TLOG() << get_name() << " successfully signaled drain";
+  TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Exiting do_drain() method";
+}
+
+void
+DFOModule::do_stop(const CommandData_t& /*args*/)
+{
+  TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Entering do_stop() method";
+
   const int wait_steps = 20;
   auto step_timeout = m_stop_timeout / wait_steps;
   int step_counter = 0;
@@ -191,6 +202,7 @@ DFOModule::do_stop(const CommandData_t& /*args*/)
 
   m_status_watchdog_thread->request_stop();
   m_status_watchdog_thread->join();
+  auto iom = iomanager::IOManager::get();
   iom->remove_callback<dfmessages::DataflowStatus>(m_status_connection);
 
   {
@@ -351,7 +363,7 @@ bool
 DFOModule::is_busy() const
 {
   if (m_processing_td.load()) {
-    TLOG(TLVL_NOTIFY_TRIGGER) << get_name() << " is busy processing a TriggerDecision";
+    TLOG_DEBUG(TLVL_NOTIFY_TRIGGER) << get_name() << " is busy processing a TriggerDecision";
     return true; // DFO is busy processing a TriggerDecision
   }
 
@@ -362,14 +374,14 @@ DFOModule::is_busy() const
     // Check if this DF app is not busy (has available slots)
     size_t occupied = dfapp.second->status.triggers_building.size() + dfapp.second->status.triggers_writing.size();
     if (!dfapp.second->status.is_busy && occupied < dfapp.second->status.busy_threshold) {
-      TLOG(TLVL_NOTIFY_TRIGGER) << get_name() << " is not busy: DF app " << dfapp.first
-                                << " has available slots (occupied: " << occupied
-                                << ", busy_threshold: " << dfapp.second->status.busy_threshold << ")";
+      TLOG_DEBUG(TLVL_NOTIFY_TRIGGER_VERBOSE)
+        << get_name() << " is not busy: DF app " << dfapp.first << " has available slots (occupied: " << occupied
+        << ", busy_threshold: " << dfapp.second->status.busy_threshold << ")";
       return false; // At least one DF app is available
     }
   }
 
-  TLOG(TLVL_NOTIFY_TRIGGER) << get_name() << " is busy: all DF apps are either busy or have stale statuses";
+  TLOG_DEBUG(TLVL_NOTIFY_TRIGGER) << get_name() << " is busy: all DF apps are either busy or have stale statuses";
   return true; // All DF apps are busy
 }
 
@@ -461,16 +473,41 @@ DFOModule::receive_dataflow_status(const dfmessages::DataflowStatus& status)
 {
   TLOG_DEBUG(TLVL_WORK_STEPS) << get_name() << " Received DataflowStatus from " << status.decision_destination
                               << " for trigger_number " << status.trigger_id.trigger_number << " in run "
-                              << status.trigger_id.run_number;
+                              << status.trigger_id.run_number << " (current run is " << m_run_number << ")";
 
   {
     std::lock_guard<std::mutex> guard(m_status_mutex);
     m_received_statuses.insert(status.decision_destination);
     // Update or create entry for this dataflow app
     auto it = m_dataflow_statuses.find(status.decision_destination);
     if (it != m_dataflow_statuses.end()) {
+      if (status.trigger_id.run_number != m_run_number) {
+        if (status.trigger_id.run_number !=
+            m_dataflow_statuses[status.decision_destination]->status.trigger_id.run_number) {
+          ers::error(DFOModuleRunNumberMismatch(ERS_HERE,
+                                                get_name(),
+                                                status.trigger_id.run_number,
+                                                m_run_number,
+                                                status.decision_destination,
+                                                status.trigger_id.trigger_number));
+        } else {
+          if (status.triggers_building.size() == 0 && status.triggers_writing.size() == 0) {
+            // Empty DataflowStatus message from previous run, safe to ignore
+            return;
+          }
+        }
+      }
+
       it->second->update(status);
     } else {
+      if (status.trigger_id.run_number != m_run_number) {
+        ers::error(DFOModuleRunNumberMismatch(ERS_HERE,
+                                              get_name(),
+                                              status.trigger_id.run_number,
+                                              m_run_number,
+                                              status.decision_destination,
+                                              status.trigger_id.trigger_number));
+      }
       m_dataflow_statuses[status.decision_destination] =
         std::make_shared<ReceivedDataflowStatus>(status, m_dataflow_status_timeout);
     }

plugins/DFOModule.hpp

@@ -138,6 +138,7 @@ class DFOModule : public dunedaq::appfwk::DAQModule
   // Commands
   void do_conf(const CommandData_t&);
   void do_start(const CommandData_t&);
+  void do_drain(const CommandData_t&);
   void do_stop(const CommandData_t&);
   void do_scrap(const CommandData_t&);
 

plugins/DataWriterModule.cpp

@@ -54,6 +54,7 @@ DataWriterModule::DataWriterModule(const std::string& name)
 {
   register_command("conf", &DataWriterModule::do_conf);
   register_command("start", &DataWriterModule::do_start);
+  register_command("drain_dataflow", &DataWriterModule::do_drain);
   register_command("stop", &DataWriterModule::do_stop);
   register_command("scrap", &DataWriterModule::do_scrap);
 }
@@ -188,6 +189,7 @@ DataWriterModule::do_start(const CommandData_t& payload)
   m_bytes_output = 0;
   m_bytes_output_tot = 0;
 
+  m_draining.store(false);
   m_running.store(true);
 
   m_thread.start_working_thread(get_name());
@@ -198,6 +200,15 @@ DataWriterModule::do_start(const CommandData_t& payload)
   TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Exiting do_start() method";
 }
 
+void
+DataWriterModule::do_drain(const CommandData_t& /*args*/)
+{
+  TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Entering do_drain() method";
+  m_draining.store(true);
+  TLOG() << get_name() << " successfully set drain flag for run number " << m_run_number;
+  TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Exiting do_drain() method";
+}
+
 void
 DataWriterModule::do_stop(const CommandData_t& /*args*/)
 {
@@ -278,7 +289,7 @@ DataWriterModule::receive_trigger_record(std::unique_ptr<daqdataformats::Trigger
           m_bytes_output += trigger_record_ptr->get_total_size_bytes();
           m_bytes_output_tot += trigger_record_ptr->get_total_size_bytes();
         } catch (const RetryableDataStoreProblem& excpt) {
-          should_retry = true;
+          should_retry = !m_draining.load();
           ers::error(DataWritingProblem(ERS_HERE,
                                         get_name(),
                                         trigger_record_ptr->get_header_ref().get_trigger_number(),

plugins/DataWriterModule.hpp

@@ -53,12 +53,14 @@ class DataWriterModule : public dunedaq::appfwk::DAQModule
   // Commands
   void do_conf(const CommandData_t&);
   void do_start(const CommandData_t&);
+  void do_drain(const CommandData_t&);
   void do_stop(const CommandData_t&);
   void do_scrap(const CommandData_t&);
 
   // Callback
   void receive_trigger_record(std::unique_ptr<daqdataformats::TriggerRecord>&);
   std::atomic<bool> m_running = false;
+  std::atomic<bool> m_draining = false;
 
   // Configuration
   std::shared_ptr<appfwk::ConfigurationManager> m_module_configuration;

plugins/DataflowStatusModule.cpp

@@ -198,7 +198,6 @@ DataflowStatusModule::do_conf(const CommandData_t&)
   iom->add_callback<dfmessages::DataflowStatusRequest>(
     m_status_request_connection, std::bind(&DataflowStatusModule::receive_status_request, this, std::placeholders::_1));
 
-  m_heartbeat_thread.start_working_thread();
   m_status_update_thread.start_working_thread();
   TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Exiting do_conf() method";
 }
@@ -252,6 +251,7 @@ DataflowStatusModule::do_start(const CommandData_t& payload)
   iom->add_callback<dfmessages::TRBCompletion>(
     m_trb_completion_connection, std::bind(&DataflowStatusModule::receive_trb_completion, this, std::placeholders::_1));
 
+  m_heartbeat_thread.start_working_thread();
   TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << get_name() << ": Exiting do_start() method";
 }
 
@@ -285,6 +285,9 @@ DataflowStatusModule::do_stop(const CommandData_t& /*args*/)
   }
   iom->remove_callback<dfmessages::TriggerDecisionToken>(m_token_connection);
 
+  if (m_heartbeat_thread.thread_running()) {
+    m_heartbeat_thread.stop_working_thread();
+  }
   m_current_status.trigger_id.run_number = 0;
 
   TLOG() << get_name() << " successfully stopped";
@@ -424,8 +427,12 @@ DataflowStatusModule::process_trb_completion(const dfmessages::TRBCompletion& co
     auto now = std::chrono::steady_clock::now();
     if (!m_current_status.triggers_building.count(completion.trigger_id)) {
       if (now - received > m_pending_message_timeout) {
-        ers::error(UnexpectedTRBCompletion(
-          ERS_HERE, get_name(), completion.trigger_id.trigger_number, completion.trigger_id.run_number));
+        ers::error(UnexpectedTRBCompletion(ERS_HERE,
+                                           get_name(),
+                                           completion.trigger_id.trigger_number,
+                                           completion.sequence_number,
+                                           completion.trigger_id.run_number,
+                                           m_current_status.trigger_id.run_number));
         m_num_pending_trb_completion_timeouts++;
         return true; // Remove from list
       }
@@ -434,6 +441,7 @@ DataflowStatusModule::process_trb_completion(const dfmessages::TRBCompletion& co
 
     ++m_num_trb_completions_received;
 
+    auto remove_from_building_list = true;
     if (completion.trigger_record_max_sequence_number > 0) {
       if (m_trigger_statuses[completion.trigger_id].max_sequence_number == 0) {
         m_trigger_statuses[completion.trigger_id].max_sequence_number = completion.trigger_record_max_sequence_number;
@@ -451,12 +459,14 @@ DataflowStatusModule::process_trb_completion(const dfmessages::TRBCompletion& co
                                     << completion.trigger_id.run_number << ". Total completed sequences: "
                                     << m_trigger_statuses[completion.trigger_id].sequences_built << " of "
                                     << m_trigger_statuses[completion.trigger_id].max_sequence_number + 1;
-        return true;
+        remove_from_building_list = false;
       }
     }
 
     m_pending_trb_completion_time_us += std::chrono::duration_cast<std::chrono::microseconds>(now - received).count();
-    m_current_status.triggers_building.erase(completion.trigger_id);
+    if (remove_from_building_list) {
+      m_current_status.triggers_building.erase(completion.trigger_id);
+    }
     m_current_status.triggers_writing.insert(completion.trigger_id);
     m_status_updated.store(true);
     m_status_update_cv.notify_all();
@@ -480,12 +490,20 @@ DataflowStatusModule::process_trigger_decision_token(const dfmessages::TriggerDe
       if (now - received > m_pending_message_timeout) {
         ++m_num_pending_trigger_decision_token_timeouts;
         if (m_trigger_statuses.count(token.trigger_id) == 0) {
-          ers::error(UnexpectedTriggerDecisionToken(
-            ERS_HERE, get_name(), token.trigger_id.trigger_number, m_current_status.trigger_id.run_number));
+          ers::error(UnexpectedTriggerDecisionToken(ERS_HERE,
+                                                    get_name(),
+                                                    token.trigger_id.trigger_number,
+                                                    token.sequence_number,
+                                                    token.trigger_id.run_number,
+                                                    m_current_status.trigger_id.run_number));
           return true; // Remove from list
         } else {
-          ers::warning(UnexpectedTriggerDecisionToken(
-            ERS_HERE, get_name(), token.trigger_id.trigger_number, m_current_status.trigger_id.run_number));
+          ers::warning(UnexpectedTriggerDecisionToken(ERS_HERE,
+                                                      get_name(),
+                                                      token.trigger_id.trigger_number,
+                                                      token.sequence_number,
+                                                      token.trigger_id.run_number,
+                                                      m_current_status.trigger_id.run_number));
         }
       } else {
         return false;
@@ -564,7 +582,7 @@ DataflowStatusModule::send_dataflow_status_update(std::string const& destination
       if (it != m_status_snapshots.end()) {
         status_to_send = it->second;
       } else {
-        ers::warning(SnapshotNotFound(ERS_HERE, get_name(), trigger_number));
+        ers::warning(SnapshotNotFound(ERS_HERE, get_name(), trigger_number, m_current_status.trigger_id.run_number));
         status_to_send = m_current_status;
       }
     }

plugins/DataflowStatusModule.hpp

@@ -39,28 +39,33 @@ ERS_DECLARE_ISSUE_BASE(dfmodules,
                        ((std::string)name),
                        ((uint32_t)trigger_number)((uint32_t)run_number)((uint32_t)current_run)) // NOLINT
 
-ERS_DECLARE_ISSUE_BASE(dfmodules,
-                       UnexpectedTRBCompletion,
-                       appfwk::GeneralDAQModuleIssue,
-                       "Received TRBCompletion message for trigger number "
-                         << trigger_number << " in run " << run_number << " which is not in the building list",
-                       ((std::string)name),
-                       ((uint32_t)trigger_number)((uint32_t)run_number)) // NOLINT
-
-ERS_DECLARE_ISSUE_BASE(dfmodules,
-                       UnexpectedTriggerDecisionToken,
-                       appfwk::GeneralDAQModuleIssue,
-                       "Received TriggerDecisionToken message for trigger number "
-                         << trigger_number << " in run " << run_number << " which is not in the writing list",
-                       ((std::string)name),
-                       ((uint32_t)trigger_number)((uint32_t)run_number)) // NOLINT
+ERS_DECLARE_ISSUE_BASE(
+  dfmodules,
+  UnexpectedTRBCompletion,
+  appfwk::GeneralDAQModuleIssue,
+  "Received TRBCompletion message for trigger number " << trigger_number << "." << sequence_number << "/" << run_number
+                                                       << " which is not in the building list (current run is "
+                                                       << current_run << ")",
+  ((std::string)name),
+  ((uint32_t)trigger_number)((uint32_t)sequence_number)((uint32_t)run_number)((uint32_t)current_run)) // NOLINT
+
+ERS_DECLARE_ISSUE_BASE(
+  dfmodules,
+  UnexpectedTriggerDecisionToken,
+  appfwk::GeneralDAQModuleIssue,
+  "Received TriggerDecisionToken message for trigger number "
+    << trigger_number << "." << sequence_number << "/" << run_number
+    << " which is not in the writing list (current run is " << current_run << ")",
+  ((std::string)name),
+  ((uint32_t)trigger_number)((uint32_t)sequence_number)((uint32_t)run_number)((uint32_t)current_run)) // NOLINT
 
 ERS_DECLARE_ISSUE_BASE(dfmodules,
                        SnapshotNotFound,
                        appfwk::GeneralDAQModuleIssue,
-                       "No snapshot found for trigger number " << trigger_number << ". Sending current status instead.",
+                       "No snapshot found for trigger number " << trigger_number << " in run " << current_run
+                                                               << ". Sending current status instead.",
                        ((std::string)name),
-                       ((uint32_t)trigger_number)) // NOLINT
+                       ((uint32_t)trigger_number)((uint32_t)current_run)) // NOLINT
 // Re-enable coverage checking LCOV_EXCL_STOP
 namespace dfmodules {
 

unittest/DataflowStatusModule_test.cxx

@@ -840,10 +840,10 @@ BOOST_AUTO_TEST_CASE(MultipleSequences)
   ConnectionFixture::send_trb_completion(100, 0, 2);
   std::this_thread::sleep_for(std::chrono::milliseconds(150));
 
-  // Should still be building, waiting for all sequences to complete
+  // Should now be both building and writing, since at least one sequence has been completed
   BOOST_REQUIRE(ConnectionFixture::s_received_statuses.back().triggers_building.size() == 1);
   BOOST_REQUIRE(ConnectionFixture::s_received_statuses.back().triggers_building.count({ 1, 100 }) == 1);
-  BOOST_REQUIRE(ConnectionFixture::s_received_statuses.back().triggers_writing.size() == 0);
+  BOOST_REQUIRE(ConnectionFixture::s_received_statuses.back().triggers_writing.size() == 1);
   BOOST_REQUIRE(ConnectionFixture::s_received_statuses.back().recently_completed_triggers.size() == 0);
   BOOST_REQUIRE(ConnectionFixture::s_received_statuses.back().trigger_records_processed == 0);
 
@@ -858,7 +858,7 @@ BOOST_AUTO_TEST_CASE(MultipleSequences)
   // Still building, waiting for sequence 2
   BOOST_REQUIRE(ConnectionFixture::s_received_statuses.back().triggers_building.size() == 1);
   BOOST_REQUIRE(ConnectionFixture::s_received_statuses.back().triggers_building.count({ 1, 100 }) == 1);
-  BOOST_REQUIRE(ConnectionFixture::s_received_statuses.back().triggers_writing.size() == 0);
+  BOOST_REQUIRE(ConnectionFixture::s_received_statuses.back().triggers_writing.size() == 1);
   BOOST_REQUIRE(ConnectionFixture::s_received_statuses.back().recently_completed_triggers.size() == 0);
   BOOST_REQUIRE(ConnectionFixture::s_received_statuses.back().trigger_records_processed == 0);