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": [ diff --git a/plugins/DFOModule.cpp b/plugins/DFOModule.cpp index f6b9ec07..b4f9b14c 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 { @@ -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(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(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,7 +473,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); @@ -469,8 +481,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); } 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/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; diff --git a/plugins/DataflowStatusModule.cpp b/plugins/DataflowStatusModule.cpp index 37a940c9..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"; @@ -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(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; } } diff --git a/plugins/DataflowStatusModule.hpp b/plugins/DataflowStatusModule.hpp index 0db32283..72933393 100644 --- a/plugins/DataflowStatusModule.hpp +++ b/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 { 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);