From cef892555bf28b86e78f0135f59b1b4eac963e53 Mon Sep 17 00:00:00 2001 From: Eric Flumerfelt Date: Thu, 5 Nov 2026 07:38:23 -0600 Subject: [PATCH 01/11] Improve handling of sequenced triggers...they are allowed to be both in the building and writing states. --- plugins/DFOModule.cpp | 7 ++++--- plugins/DataflowStatusModule.cpp | 15 +++++++++------ plugins/DataflowStatusModule.hpp | 16 ++++++++-------- unittest/DataflowStatusModule_test.cxx | 6 +++--- 4 files changed, 24 insertions(+), 20 deletions(-) diff --git a/plugins/DFOModule.cpp b/plugins/DFOModule.cpp index f6b9ec07..b837423c 100644 --- a/plugins/DFOModule.cpp +++ b/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 { @@ -362,7 +363,7 @@ 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 + TLOG(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 diff --git a/plugins/DataflowStatusModule.cpp b/plugins/DataflowStatusModule.cpp index 37a940c9..4e400315 100644 --- a/plugins/DataflowStatusModule.cpp +++ b/plugins/DataflowStatusModule.cpp @@ -425,7 +425,7 @@ DataflowStatusModule::process_trb_completion(const dfmessages::TRBCompletion& co 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_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 +434,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 +452,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(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(); @@ -481,11 +484,11 @@ DataflowStatusModule::process_trigger_decision_token(const dfmessages::TriggerDe ++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_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_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 +567,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; } } diff --git a/plugins/DataflowStatusModule.hpp b/plugins/DataflowStatusModule.hpp index 0db32283..41c2aae3 100644 --- a/plugins/DataflowStatusModule.hpp +++ b/plugins/DataflowStatusModule.hpp @@ -33,8 +33,8 @@ namespace dunedaq { ERS_DECLARE_ISSUE_BASE(dfmodules, TriggerDecisionIncorrectRun, appfwk::GeneralDAQModuleIssue, - "Received TriggerDecision message for trigger number " - << trigger_number << " in run " << run_number << " which is not the current run (" + "Received TriggerDecision message for trigger number " << + trigger_number << " in run " << run_number << " which is not the current run (" << current_run << ")", ((std::string)name), ((uint32_t)trigger_number)((uint32_t)run_number)((uint32_t)current_run)) // NOLINT @@ -43,24 +43,24 @@ 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", + << 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)run_number)) // NOLINT + ((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 << " in run " << run_number << " which is not in the writing list", + << 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)run_number)) // NOLINT + ((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 { diff --git a/unittest/DataflowStatusModule_test.cxx b/unittest/DataflowStatusModule_test.cxx index 9a7e6f1f..550bf52b 100644 --- a/unittest/DataflowStatusModule_test.cxx +++ b/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); From 894b4007c38b7ca628b10fb930faecfe6a06dbae Mon Sep 17 00:00:00 2001 From: Eric Flumerfelt Date: Thu, 5 Nov 2026 07:42:57 -0600 Subject: [PATCH 02/11] Fix DFOModule::is_busy logging --- plugins/DFOModule.cpp | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/plugins/DFOModule.cpp b/plugins/DFOModule.cpp index b837423c..503f9ced 100644 --- a/plugins/DFOModule.cpp +++ b/plugins/DFOModule.cpp @@ -352,7 +352,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 } @@ -363,14 +363,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_VERBOSE) << get_name() << " is not busy: DF app " << dfapp.first + 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 } From f51d33e4d14b6cd9841c2416ff0a72c3abe7e193 Mon Sep 17 00:00:00 2001 From: Eric Flumerfelt Date: Thu, 5 Nov 2026 08:53:46 -0600 Subject: [PATCH 03/11] Identify when an entire status message is from a mismatched run --- plugins/DFOModule.cpp | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/plugins/DFOModule.cpp b/plugins/DFOModule.cpp index 503f9ced..a75f517c 100644 --- a/plugins/DFOModule.cpp +++ b/plugins/DFOModule.cpp @@ -464,6 +464,16 @@ DFOModule::receive_dataflow_status(const dfmessages::DataflowStatus& status) << " for trigger_number " << status.trigger_id.trigger_number << " in run " << status.trigger_id.run_number; + if(status.trigger_id.run_number != m_run_number) { + ers::warning(DFOModuleRunNumberMismatch(ERS_HERE, + get_name(), + status.trigger_id.run_number, + m_run_number, + status.decision_destination, + status.trigger_id.trigger_number)); + return; + } + { std::lock_guard guard(m_status_mutex); m_received_statuses.insert(status.decision_destination); From bd692fb17106d364bb127e0273076df9a2d95547 Mon Sep 17 00:00:00 2001 From: Eric Flumerfelt Date: Thu, 5 Nov 2026 08:54:12 -0600 Subject: [PATCH 04/11] Run clang-format --- include/dfmodules/FileDataStoreImpl.hpp | 18 +++++------ plugins/DFOModule.cpp | 18 +++++------ plugins/DataflowStatusModule.cpp | 26 +++++++++++----- plugins/DataflowStatusModule.hpp | 41 ++++++++++++++----------- 4 files changed, 60 insertions(+), 43 deletions(-) diff --git a/include/dfmodules/FileDataStoreImpl.hpp b/include/dfmodules/FileDataStoreImpl.hpp index 6a983683..42658b47 100644 --- a/include/dfmodules/FileDataStoreImpl.hpp +++ b/include/dfmodules/FileDataStoreImpl.hpp @@ -496,7 +496,7 @@ class FileDataStoreImpl : public DataStore daqdataformats::run_number_t m_run_number; // Total number of generated files - std::atomic m_file_index{0}; + std::atomic m_file_index{ 0 }; const std::string m_writer_identifier; const appmodel::DataStoreConf* m_config_params; @@ -512,31 +512,31 @@ class FileDataStoreImpl : public DataStore std::string m_basic_name_of_open_file; // Size of data being written, excluding metadata - std::atomic m_recorded_size{0}; + std::atomic m_recorded_size{ 0 }; // Theoretical, "uncompressed" size of data being written, excluding metadata - std::atomic m_uncompressed_raw_data_size{0}; + std::atomic m_uncompressed_raw_data_size{ 0 }; // Used for tracking the "delta" of the current write - std::atomic m_previous_file_size{0}; + std::atomic m_previous_file_size{ 0 }; // Total size of the file, including raw data, metadata, and free space - std::atomic m_total_file_size{0}; + std::atomic m_total_file_size{ 0 }; // Record number for the record that is currently being written out // This is only useful for long-readout windows, in which there may // be multiple calls to write() - size_t m_current_record_number{0}; + size_t m_current_record_number{ 0 }; // incremental written data - std::atomic m_new_bytes{0}; // NOLINT(build/unsigned) - std::atomic m_new_objects{0}; // NOLINT(build/unsigned) + std::atomic m_new_bytes{ 0 }; // NOLINT(build/unsigned) + std::atomic m_new_objects{ 0 }; // NOLINT(build/unsigned) const std::string m_operation_mode; const std::string m_path; const size_t m_max_file_size; const bool m_disable_unique_suffix; - float m_free_space_safety_factor_for_write{0.0f}; + float m_free_space_safety_factor_for_write{ 0.0f }; }; } // namespace dfmodules diff --git a/plugins/DFOModule.cpp b/plugins/DFOModule.cpp index a75f517c..1864d1dc 100644 --- a/plugins/DFOModule.cpp +++ b/plugins/DFOModule.cpp @@ -363,9 +363,9 @@ 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_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 << ")"; + 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 } } @@ -464,13 +464,13 @@ DFOModule::receive_dataflow_status(const dfmessages::DataflowStatus& status) << " for trigger_number " << status.trigger_id.trigger_number << " in run " << status.trigger_id.run_number; - if(status.trigger_id.run_number != m_run_number) { + if (status.trigger_id.run_number != m_run_number) { ers::warning(DFOModuleRunNumberMismatch(ERS_HERE, - get_name(), - status.trigger_id.run_number, - m_run_number, - status.decision_destination, - status.trigger_id.trigger_number)); + get_name(), + status.trigger_id.run_number, + m_run_number, + status.decision_destination, + status.trigger_id.trigger_number)); return; } diff --git a/plugins/DataflowStatusModule.cpp b/plugins/DataflowStatusModule.cpp index 4e400315..b2d85cc1 100644 --- a/plugins/DataflowStatusModule.cpp +++ b/plugins/DataflowStatusModule.cpp @@ -424,8 +424,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.sequence_number, completion.trigger_id.run_number, m_current_status.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 } @@ -457,7 +461,7 @@ DataflowStatusModule::process_trb_completion(const dfmessages::TRBCompletion& co } m_pending_trb_completion_time_us += std::chrono::duration_cast(now - received).count(); - if(remove_from_building_list) { + if (remove_from_building_list) { m_current_status.triggers_building.erase(completion.trigger_id); } m_current_status.triggers_writing.insert(completion.trigger_id); @@ -483,12 +487,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 , token.sequence_number, token.trigger_id.run_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, token.sequence_number, token.trigger_id.run_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; diff --git a/plugins/DataflowStatusModule.hpp b/plugins/DataflowStatusModule.hpp index 41c2aae3..72933393 100644 --- a/plugins/DataflowStatusModule.hpp +++ b/plugins/DataflowStatusModule.hpp @@ -33,32 +33,37 @@ namespace dunedaq { ERS_DECLARE_ISSUE_BASE(dfmodules, TriggerDecisionIncorrectRun, appfwk::GeneralDAQModuleIssue, - "Received TriggerDecision message for trigger number " << - trigger_number << " in run " << run_number << " which is not the current run (" + "Received TriggerDecision message for trigger number " + << trigger_number << " in run " << run_number << " which is not the current run (" << current_run << ")", ((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 << "." << 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, + 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 << " in run " << current_run << ". 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)((uint32_t)current_run)) // NOLINT // Re-enable coverage checking LCOV_EXCL_STOP From 80dc1f5547c6be77a42c893fc588c14df6798467 Mon Sep 17 00:00:00 2001 From: Eric Flumerfelt Date: Thu, 5 Nov 2026 09:13:29 -0600 Subject: [PATCH 05/11] Reduce severity of messages, since DFO knows to ignore stale information --- plugins/DFOModule.cpp | 14 ++------------ 1 file changed, 2 insertions(+), 12 deletions(-) diff --git a/plugins/DFOModule.cpp b/plugins/DFOModule.cpp index 1864d1dc..fde6d5d2 100644 --- a/plugins/DFOModule.cpp +++ b/plugins/DFOModule.cpp @@ -464,16 +464,6 @@ DFOModule::receive_dataflow_status(const dfmessages::DataflowStatus& status) << " for trigger_number " << status.trigger_id.trigger_number << " in run " << status.trigger_id.run_number; - if (status.trigger_id.run_number != m_run_number) { - ers::warning(DFOModuleRunNumberMismatch(ERS_HERE, - get_name(), - status.trigger_id.run_number, - m_run_number, - status.decision_destination, - status.trigger_id.trigger_number)); - return; - } - { std::lock_guard guard(m_status_mutex); m_received_statuses.insert(status.decision_destination); @@ -498,7 +488,7 @@ DFOModule::receive_dataflow_status(const dfmessages::DataflowStatus& status) for (auto& trigger : status.triggers_building) { if (!m_assigned_trigger_decisions.count(trigger)) { if (trigger.run_number != m_run_number) { - ers::error(DFOModuleRunNumberMismatch(ERS_HERE, + ers::info(DFOModuleRunNumberMismatch(ERS_HERE, get_name(), trigger.run_number, m_run_number, @@ -517,7 +507,7 @@ DFOModule::receive_dataflow_status(const dfmessages::DataflowStatus& status) for (auto& trigger : status.triggers_writing) { if (!m_assigned_trigger_decisions.count(trigger)) { if (trigger.run_number != m_run_number) { - ers::error(DFOModuleRunNumberMismatch(ERS_HERE, + ers::info(DFOModuleRunNumberMismatch(ERS_HERE, get_name(), trigger.run_number, m_run_number, From 242e76b2bc13cc3af4cd710c45c3b959ad8b6d07 Mon Sep 17 00:00:00 2001 From: Eric Flumerfelt Date: Thu, 5 Nov 2026 09:24:25 -0600 Subject: [PATCH 06/11] Add comment detailing why DataflowStatus messages with incorrect run numbers are okay to ignore/not raise errors or warnings --- plugins/DFOModule.cpp | 24 ++++++++++++++---------- 1 file changed, 14 insertions(+), 10 deletions(-) diff --git a/plugins/DFOModule.cpp b/plugins/DFOModule.cpp index fde6d5d2..febc125b 100644 --- a/plugins/DFOModule.cpp +++ b/plugins/DFOModule.cpp @@ -460,6 +460,10 @@ DFOModule::get_trigger_counter(trgdataformats::TriggerCandidateData::Type type) void DFOModule::receive_dataflow_status(const dfmessages::DataflowStatus& status) { + // DataflowStatusModule sends messages from conf through scrap. The run number in the message is only updated at run + // start, so when this callback starts, there may be queued messages with either uninitalized run number or the + // previous run number. DFOModule is only interested in status updates for triggers in the current run, so it can just + // log and ignore others. 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; @@ -489,11 +493,11 @@ DFOModule::receive_dataflow_status(const dfmessages::DataflowStatus& status) if (!m_assigned_trigger_decisions.count(trigger)) { if (trigger.run_number != m_run_number) { ers::info(DFOModuleRunNumberMismatch(ERS_HERE, - get_name(), - trigger.run_number, - m_run_number, - status.decision_destination, - trigger.trigger_number)); + get_name(), + trigger.run_number, + m_run_number, + status.decision_destination, + trigger.trigger_number)); continue; } ers::info(UnexpectedTriggerDecision(ERS_HERE, get_name(), trigger.trigger_number, status.decision_destination)); @@ -508,11 +512,11 @@ DFOModule::receive_dataflow_status(const dfmessages::DataflowStatus& status) if (!m_assigned_trigger_decisions.count(trigger)) { if (trigger.run_number != m_run_number) { ers::info(DFOModuleRunNumberMismatch(ERS_HERE, - get_name(), - trigger.run_number, - m_run_number, - status.decision_destination, - trigger.trigger_number)); + get_name(), + trigger.run_number, + m_run_number, + status.decision_destination, + trigger.trigger_number)); continue; } ers::warning( From 350490def5a848b1f7440dbae25520797dddf5f2 Mon Sep 17 00:00:00 2001 From: Eric Flumerfelt Date: Thu, 5 Nov 2026 11:42:57 -0600 Subject: [PATCH 07/11] Remove note and restore error messages. Move heartbeat thread to start/stop instead of conf/scrap --- plugins/DFOModule.cpp | 8 ++------ plugins/DataflowStatusModule.cpp | 5 ++++- 2 files changed, 6 insertions(+), 7 deletions(-) diff --git a/plugins/DFOModule.cpp b/plugins/DFOModule.cpp index febc125b..2aa32507 100644 --- a/plugins/DFOModule.cpp +++ b/plugins/DFOModule.cpp @@ -460,10 +460,6 @@ DFOModule::get_trigger_counter(trgdataformats::TriggerCandidateData::Type type) void DFOModule::receive_dataflow_status(const dfmessages::DataflowStatus& status) { - // DataflowStatusModule sends messages from conf through scrap. The run number in the message is only updated at run - // start, so when this callback starts, there may be queued messages with either uninitalized run number or the - // previous run number. DFOModule is only interested in status updates for triggers in the current run, so it can just - // log and ignore others. 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; @@ -492,7 +488,7 @@ DFOModule::receive_dataflow_status(const dfmessages::DataflowStatus& status) for (auto& trigger : status.triggers_building) { if (!m_assigned_trigger_decisions.count(trigger)) { if (trigger.run_number != m_run_number) { - ers::info(DFOModuleRunNumberMismatch(ERS_HERE, + ers::error(DFOModuleRunNumberMismatch(ERS_HERE, get_name(), trigger.run_number, m_run_number, @@ -511,7 +507,7 @@ DFOModule::receive_dataflow_status(const dfmessages::DataflowStatus& status) for (auto& trigger : status.triggers_writing) { if (!m_assigned_trigger_decisions.count(trigger)) { if (trigger.run_number != m_run_number) { - ers::info(DFOModuleRunNumberMismatch(ERS_HERE, + ers::error(DFOModuleRunNumberMismatch(ERS_HERE, get_name(), trigger.run_number, m_run_number, diff --git a/plugins/DataflowStatusModule.cpp b/plugins/DataflowStatusModule.cpp index b2d85cc1..c224e1ae 100644 --- a/plugins/DataflowStatusModule.cpp +++ b/plugins/DataflowStatusModule.cpp @@ -198,7 +198,6 @@ DataflowStatusModule::do_conf(const CommandData_t&) iom->add_callback( 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( 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(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"; From 38ca4771ddaea664d521db90c7dffaedd9a49a76 Mon Sep 17 00:00:00 2001 From: Eric Flumerfelt Date: Thu, 5 Nov 2026 14:49:37 -0600 Subject: [PATCH 08/11] Call DataflowStatusModule::do_stop at drain_dataflow. Implement check for stale run DataflowStatus message, silently allowing it if it is from the previous run and there are no open triggers --- plugins/DFOModule.cpp | 47 ++++++++++++++++++++++++-------- plugins/DataflowStatusModule.cpp | 2 +- 2 files changed, 37 insertions(+), 12 deletions(-) diff --git a/plugins/DFOModule.cpp b/plugins/DFOModule.cpp index 2aa32507..cd27c0cb 100644 --- a/plugins/DFOModule.cpp +++ b/plugins/DFOModule.cpp @@ -462,7 +462,7 @@ 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 guard(m_status_mutex); @@ -470,8 +470,33 @@ DFOModule::receive_dataflow_status(const dfmessages::DataflowStatus& status) // 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(status, m_dataflow_status_timeout); } @@ -489,11 +514,11 @@ DFOModule::receive_dataflow_status(const dfmessages::DataflowStatus& status) if (!m_assigned_trigger_decisions.count(trigger)) { if (trigger.run_number != m_run_number) { ers::error(DFOModuleRunNumberMismatch(ERS_HERE, - get_name(), - trigger.run_number, - m_run_number, - status.decision_destination, - trigger.trigger_number)); + get_name(), + trigger.run_number, + m_run_number, + status.decision_destination, + trigger.trigger_number)); continue; } ers::info(UnexpectedTriggerDecision(ERS_HERE, get_name(), trigger.trigger_number, status.decision_destination)); @@ -508,11 +533,11 @@ DFOModule::receive_dataflow_status(const dfmessages::DataflowStatus& status) if (!m_assigned_trigger_decisions.count(trigger)) { if (trigger.run_number != m_run_number) { ers::error(DFOModuleRunNumberMismatch(ERS_HERE, - get_name(), - trigger.run_number, - m_run_number, - status.decision_destination, - trigger.trigger_number)); + get_name(), + trigger.run_number, + m_run_number, + status.decision_destination, + trigger.trigger_number)); continue; } ers::warning( diff --git a/plugins/DataflowStatusModule.cpp b/plugins/DataflowStatusModule.cpp index c224e1ae..e502db29 100644 --- a/plugins/DataflowStatusModule.cpp +++ b/plugins/DataflowStatusModule.cpp @@ -39,7 +39,7 @@ DataflowStatusModule::DataflowStatusModule(const std::string& name) { register_command("conf", &DataflowStatusModule::do_conf); register_command("start", &DataflowStatusModule::do_start); - register_command("stop", &DataflowStatusModule::do_stop); + register_command("drain_dataflow", &DataflowStatusModule::do_stop); register_command("scrap", &DataflowStatusModule::do_scrap); } From 146aa5d09ac47c3389ceac9a508e0f55fe01d450 Mon Sep 17 00:00:00 2001 From: Eric Flumerfelt Date: Fri, 6 Nov 2026 09:11:09 -0600 Subject: [PATCH 09/11] Split drain_dataflow and stop in DFOModule --- plugins/DFOModule.cpp | 17 ++++++++++++++--- plugins/DFOModule.hpp | 1 + plugins/DataflowStatusModule.cpp | 2 +- 3 files changed, 16 insertions(+), 4 deletions(-) diff --git a/plugins/DFOModule.cpp b/plugins/DFOModule.cpp index cd27c0cb..b4f9b14c 100644 --- a/plugins/DFOModule.cpp +++ b/plugins/DFOModule.cpp @@ -56,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); } @@ -171,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(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; @@ -192,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(m_status_connection); { diff --git a/plugins/DFOModule.hpp b/plugins/DFOModule.hpp index 0ca6f5cf..962a6043 100644 --- a/plugins/DFOModule.hpp +++ b/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&); diff --git a/plugins/DataflowStatusModule.cpp b/plugins/DataflowStatusModule.cpp index e502db29..c224e1ae 100644 --- a/plugins/DataflowStatusModule.cpp +++ b/plugins/DataflowStatusModule.cpp @@ -39,7 +39,7 @@ DataflowStatusModule::DataflowStatusModule(const std::string& name) { register_command("conf", &DataflowStatusModule::do_conf); register_command("start", &DataflowStatusModule::do_start); - register_command("drain_dataflow", &DataflowStatusModule::do_stop); + register_command("stop", &DataflowStatusModule::do_stop); register_command("scrap", &DataflowStatusModule::do_scrap); } From 95a198feb941823ede078309a61b240cf5279b60 Mon Sep 17 00:00:00 2001 From: Eric Flumerfelt Date: Fri, 6 Nov 2026 10:30:58 -0600 Subject: [PATCH 10/11] Add do_drain method for DataWriterModule, which instructs the module to stop retrying RetryableDataStoreProblem exceptions --- plugins/DataWriterModule.cpp | 13 ++++++++++++- plugins/DataWriterModule.hpp | 2 ++ 2 files changed, 14 insertions(+), 1 deletion(-) diff --git a/plugins/DataWriterModule.cpp b/plugins/DataWriterModule.cpp index 5fdedb5e..33051d19 100644 --- a/plugins/DataWriterModule.cpp +++ b/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_ptrget_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(), diff --git a/plugins/DataWriterModule.hpp b/plugins/DataWriterModule.hpp index d81cbbb8..7abadd0d 100644 --- a/plugins/DataWriterModule.hpp +++ b/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&); std::atomic m_running = false; + std::atomic m_draining = false; // Configuration std::shared_ptr m_module_configuration; From d4b292b1ef96f0e6c6ddf4f327a8d578973d3ca0 Mon Sep 17 00:00:00 2001 From: Eric Flumerfelt Date: Fri, 6 Nov 2026 10:57:26 -0600 Subject: [PATCH 11/11] Temporarily comment out message requirement from DFO --- integtest/insufficient_disk_space_test.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/integtest/insufficient_disk_space_test.py b/integtest/insufficient_disk_space_test.py index 132c8f67..6c2aa064 100644 --- a/integtest/insufficient_disk_space_test.py +++ b/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 , 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": [