Revise DFO protocol to enhance resilience and consensus

.github/workflows/dunedaq-develop-cpp-ci.yml

@@ -30,4 +30,4 @@ jobs:
     name: Build against the development release
     uses: DUNE-DAQ/.github/.github/workflows/dunedaq-develop-cpp-ci.yml@develop
     with:
-      caller_event_name: ${{ github.event.inputs.caller_event_name || github.event_name }}
+      caller_event_name: ${{ github.event.inputs.caller_event_name || github.event_name }}

CMakeLists.txt

@@ -22,16 +22,16 @@ find_package(Boost COMPONENTS iostreams unit_test_framework REQUIRED)
 daq_protobuf_codegen( opmon/*.proto )
 
 ##############################################################################
-daq_add_library( TriggerInhibitAgent.cpp TriggerRecordBuilderData.cpp TPBundleHandler.cpp
-                 LINK_LIBRARIES 
+daq_add_library( TriggerInhibitAgent.cpp TriggerRecordBuilderData.cpp TPBundleHandler.cpp DFOCore.cpp
+                 LINK_LIBRARIES
                  opmonlib::opmonlib ers::ers HighFive appfwk::appfwk logging::logging stdc++fs dfmessages::dfmessages utilities::utilities trigger::trigger detdataformats::detdataformats trgdataformats::trgdataformats)
 
-      
+
 daq_add_plugin( HDF5DataStore     duneDataStore LINK_LIBRARIES dfmodules hdf5libs::hdf5libs stdc++fs)
 
 daq_add_plugin( FragmentAggregatorModule    duneDAQModule LINK_LIBRARIES dfmodules iomanager::iomanager )
 daq_add_plugin( DataWriterModule            duneDAQModule LINK_LIBRARIES dfmodules hdf5libs::hdf5libs iomanager::iomanager )
-daq_add_plugin( DFOModule  duneDAQModule LINK_LIBRARIES dfmodules iomanager::iomanager )
+daq_add_plugin( DFOModule          duneDAQModule LINK_LIBRARIES dfmodules iomanager::iomanager )
 daq_add_plugin( TRBModule  duneDAQModule LINK_LIBRARIES dfmodules iomanager::iomanager )
 daq_add_plugin( TRMonRequestorModule  duneDAQModule LINK_LIBRARIES dfmodules iomanager::iomanager )
 daq_add_plugin( FakeDataProdModule          duneDAQModule LINK_LIBRARIES dfmodules iomanager::iomanager)
@@ -46,6 +46,8 @@ add_dependencies(  HDF5Write_test dfmodules_HDF5DataStore_duneDataStore )
 daq_add_unit_test( DFOModule_test LINK_LIBRARIES dfmodules )
 add_dependencies( DFOModule_test dfmodules_DFOModule_duneDAQModule)
 
+daq_add_unit_test( DFOCore_test LINK_LIBRARIES dfmodules )
+
 daq_add_unit_test( TriggerRecordBuilderData_test LINK_LIBRARIES dfmodules)
 daq_add_unit_test( DataStoreFactory_test    LINK_LIBRARIES dfmodules)
 

include/dfmodules/DFODecision.hpp

@@ -0,0 +1,63 @@
+/**
+ * @file DFODecision.hpp  DFODecision message used by DFOConsensusModule.
+ *
+ * A DFODecision is broadcast by the responsible DFOConsensusModule to all
+ * peer DFOs after each TRBModule state change (trigger assignment or
+ * completion).  This allows every DFO in the ensemble to maintain an accurate
+ * per-TRBModule slot-usage view and issue correct TriggerInhibit messages to
+ * the MLT.
+ *
+ * Fields
+ * ------
+ * run_number         – run in which this decision was made
+ * trigger_number     – the TriggerDecision that triggered the state change
+ * trb_connection_name– connection-ID of the TRBModule whose slot count changed
+ * trb_slot_count     – absolute slot count for that TRB *after* the change
+ * source_dfo_name    – name() of the DFO that generated this message
+ * is_completion      – false = new assignment; true = TRB completed a trigger
+ *
+ * Serialisation
+ * -------------
+ * nlohmann/json (to_json / from_json) is provided via
+ * NLOHMANN_DEFINE_TYPE_NON_INTRUSIVE.  For in-process Queue transport this is
+ * not required; for network (ZMQ / kSendRecv) connections production
+ * deployments should additionally register a msgpack serialiser in dfmessages.
+ *
+ * This is part of the DUNE DAQ Software Suite, copyright 2020.
+ * Licensing/copyright details are in the COPYING file that you should have
+ * received with this code.
+ */
+
+#ifndef DFMODULES_INCLUDE_DFMODULES_DFODECISION_HPP_
+#define DFMODULES_INCLUDE_DFMODULES_DFODECISION_HPP_
+
+#include "daqdataformats/Types.hpp"
+#include "nlohmann/json.hpp"
+
+#include <string>
+
+namespace dunedaq {
+namespace dfmodules {
+
+struct DFODecision
+{
+  daqdataformats::run_number_t run_number{ 0 };
+  daqdataformats::trigger_number_t trigger_number{ 0 };
+  std::string trb_connection_name;  ///< Connection-ID of the TRBModule involved
+  size_t trb_slot_count{ 0 };       ///< TRB slot count after this event
+  std::string source_dfo_name;      ///< Name of the DFO that generated this message
+  bool is_completion{ false };      ///< true = completion; false = new assignment
+};
+
+NLOHMANN_DEFINE_TYPE_NON_INTRUSIVE(DFODecision,
+                                   run_number,
+                                   trigger_number,
+                                   trb_connection_name,
+                                   trb_slot_count,
+                                   source_dfo_name,
+                                   is_completion)
+
+} // namespace dfmodules
+} // namespace dunedaq
+
+#endif // DFMODULES_INCLUDE_DFMODULES_DFODECISION_HPP_

integtest/disabled_output_test.py

@@ -7,7 +7,14 @@
 
 import integrationtest.data_file_checks as data_file_checks
 import integrationtest.log_file_checks as log_file_checks
+import integrationtest.basic_checks as basic_checks
 import integrationtest.data_classes as data_classes
+import integrationtest.resource_validation as resource_validation
+from integrationtest.get_pytest_tmpdir import get_pytest_tmpdir
+from integrationtest.verbosity_helper import IntegtestVerbosityLevels
+
+import functools
+print = functools.partial(print, flush=True)  # always flush print() output
 
 pytest_plugins = "integrationtest.integrationtest_drunc"
 
@@ -78,6 +85,14 @@
     ],
 }
 
+# Determine if the conditions are right for these tests
+resource_validator = resource_validation.ResourceValidator()
+resource_validator.cpu_count_needs(6, 12)  # two for each data source plus two more for everything else
+resource_validator.free_memory_needs(10, 20)  # 25% more than what we observe being used ('free -h')
+resource_validator.total_memory_needs()  # no specific request, but it's useful to see how much is available
+actual_output_path = get_pytest_tmpdir()
+resource_validator.free_disk_space_needs(actual_output_path, 1)  # what we observe
+
 # The next three variable declarations *must* be present as globals in the test
 # file. They're read by the "fixtures" in conftest.py to determine how
 # to run the config generation and dunerc
@@ -113,22 +128,22 @@
     )
 )
 
-swtpg_conf = copy.deepcopy(conf_dict)
-swtpg_conf.tpg_enabled = True
-swtpg_conf.config_substitutions.append(
+tpg_conf = copy.deepcopy(conf_dict)
+tpg_conf.tpg_enabled = True
+tpg_conf.config_substitutions.append(
     data_classes.attribute_substitution(
         obj_class="TAMakerPrescaleAlgorithm",
         obj_id="dummy-ta-maker",
         updates={"prescale": 25},
     )
 )
-swtpg_conf.frame_file = (
+tpg_conf.frame_file = (
     "asset://?checksum=dd156b4895f1b06a06b6ff38e37bd798"  # WIBEth All Zeros
 )
 
 confgen_arguments = {
     "WIBEth_System": conf_dict,
-    "Software_TPG_System": swtpg_conf,
+    "WIBEth_TPG_System": tpg_conf,
 }
 
 # The commands to run in dunerc, as a list
@@ -158,26 +173,17 @@
 # The tests themselves
 
 
-def test_dunerc_success(run_dunerc):
-    # print the name of the current test
-    current_test = os.environ.get("PYTEST_CURRENT_TEST")
-    match_obj = re.search(r".*\[(.+)-run_.*rc.*\d].*", current_test)
-    if match_obj:
-        current_test = match_obj.group(1)
-    banner_line = re.sub(".", "=", current_test)
-    print(banner_line)
-    print(current_test)
-    print(banner_line)
-
-    # Check that dunerc completed correctly
-    assert run_dunerc.completed_process.returncode == 0
+def test_dunerc_success(run_dunerc, caplog):
+    # checks for run control success, problems during pytest setup, etc.
+    basic_checks.basic_checks(run_dunerc, caplog, print_test_name=True)
 
 
 def test_log_files(run_dunerc):
     if check_for_logfile_errors:
         # Check that there are no warnings or errors in the log files
         assert log_file_checks.logs_are_error_free(
-            run_dunerc.log_files, True, True, ignored_logfile_problems
+            run_dunerc.log_files, True, True, ignored_logfile_problems,
+            verbosity_helper=run_dunerc.verbosity_helper
         )
 
 
@@ -204,7 +210,7 @@ def test_data_files(run_dunerc):
 
     all_ok = True
     for idx in range(len(run_dunerc.data_files)):
-        data_file = data_file_checks.DataFile(run_dunerc.data_files[idx])
+        data_file = data_file_checks.DataFile(run_dunerc.data_files[idx], run_dunerc.verbosity_helper)
         all_ok &= data_file_checks.sanity_check(data_file)
         all_ok &= data_file_checks.check_file_attributes(data_file)
         all_ok &= data_file_checks.check_event_count(