Support Multiple DFOs through an update to the DFO Protocol

CMakeLists.txt

@@ -13,6 +13,8 @@ find_package(Boost COMPONENTS unit_test_framework REQUIRED)
 daq_add_library(LINK_LIBRARIES daqdataformats::daqdataformats serialization::serialization)
 
 daq_add_unit_test(DataRequest_test LINK_LIBRARIES daqdataformats::daqdataformats serialization::serialization)
+daq_add_unit_test(DataflowHeartbeat_test LINK_LIBRARIES daqdataformats::daqdataformats serialization::serialization)
+daq_add_unit_test(DFODecision_test LINK_LIBRARIES daqdataformats::daqdataformats serialization::serialization)
 daq_add_unit_test(HSIEvent_test LINK_LIBRARIES daqdataformats::daqdataformats serialization::serialization)
 daq_add_unit_test(TimeSync_test LINK_LIBRARIES daqdataformats::daqdataformats serialization::serialization)
 daq_add_unit_test(TriggerDecision_test LINK_LIBRARIES daqdataformats::daqdataformats serialization::serialization)

include/dfmessages/DFODecision.hpp

@@ -0,0 +1,41 @@
+/**
+ * @file dfmessages/DFODecision.hpp DFODecision Message Declaration
+ *
+ * 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.
+ */
+
+#ifndef DFMESSAGES_INCLUDE_DFMESSAGES_DFODECISION_HPP_
+#define DFMESSAGES_INCLUDE_DFMESSAGES_DFODECISION_HPP_
+
+#include "dfmessages/TriggerDecision.hpp"
+#include "dfmessages/Types.hpp"
+
+#include "serialization/Serialization.hpp"
+
+#include <limits>
+#include <map>
+#include <vector>
+
+namespace dunedaq {
+namespace dfmessages {
+/**
+ * @brief A message containing information about a Trigger from Data Selection (or a DFODecisionEmulator)
+ */
+struct DFODecision
+{
+  std::string dfo_id{ "UNKNOWN_DFO" };
+  TriggerDecision trigger_decision{};
+  std::vector<trigger_number_t> acknowledged_completions{};
+
+  DUNE_DAQ_SERIALIZE(DFODecision, dfo_id, trigger_decision,
+                     acknowledged_completions);
+};
+} // namespace dfmessages
+
+DUNE_DAQ_SERIALIZABLE(dfmessages::DFODecision, "DFODecision");
+
+} // namespace dunedaq
+
+#endif // DFMESSAGES_INCLUDE_DFMESSAGES_DFODECISION_HPP_

include/dfmessages/DataflowHeartbeat.hpp

@@ -0,0 +1,38 @@
+/**
+ * @file dfmessages/DataflowHeartbeat.hpp DataflowHeartbeat Message Declaration
+ *
+ * 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.
+ */
+
+#ifndef DFMESSAGES_INCLUDE_DFMESSAGES_DATAFLOWHEARTBEAT_HPP_
+#define DFMESSAGES_INCLUDE_DFMESSAGES_DATAFLOWHEARTBEAT_HPP_
+
+#include "dfmessages/Types.hpp"
+
+#include "serialization/Serialization.hpp"
+
+#include <string>
+
+namespace dunedaq {
+namespace dfmessages {
+/**
+ * @brief Represents a message indicating the current status of a DataFlow application
+ */
+struct DataflowHeartbeat
+{
+  run_number_t run_number{ TypeDefaults::s_invalid_run_number }; ///< The run number that this heartbeat corresponds to
+  std::vector<trigger_number_t> recent_completed_triggers{}; ///< A list of recently-completed triggers
+  std::vector<trigger_number_t> outstanding_decisions{}; ///< A list of trigger decisions that the app is currently working on
+
+  std::string
+    decision_destination; ///< Connection name for TriggerDecisions sent to the Dataflow process originating this heartbeat
+
+  DUNE_DAQ_SERIALIZE(DataflowHeartbeat, run_number, recent_completed_triggers, outstanding_decisions, decision_destination);
+};
+} // namespace dfmessages
+DUNE_DAQ_SERIALIZABLE(dfmessages::DataflowHeartbeat, "DataflowHeartbeat");
+} // namespace dunedaq
+
+#endif // DFMESSAGES_INCLUDE_DFMESSAGES_DATAFLOWHEARTBEAT_HPP_

include/dfmessages/TriggerInhibit.hpp

@@ -23,10 +23,10 @@ struct TriggerInhibit
 {
   bool busy = false;           ///< Whether the system is busy
   run_number_t run_number = 0; ///< Current run number
+  std::string dfo_id{ "UNKNOWN_DFO" };
 
-  DUNE_DAQ_SERIALIZE(TriggerInhibit, busy, run_number);
+  DUNE_DAQ_SERIALIZE(TriggerInhibit, busy, run_number, dfo_id);
 };
-static_assert(sizeof(TriggerInhibit) == 8, "TriggerInhibit size unexpected!");
 static_assert(offsetof(TriggerInhibit, busy) == 0, "TriggerInhibit busy field not at expected offset!");
 static_assert(offsetof(TriggerInhibit, run_number) == 4, "TriggerInhibit run_number field not at expected offset!");
 } // namespace dfmessages

unittest/DFODecision_test.cxx

@@ -0,0 +1,164 @@
+/**
+ * @file DFODecision_test.cxx DFODecision class Unit Tests
+ *
+ * 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.
+ */
+
+#include "dfmessages/DFODecision.hpp"
+
+/**
+ * @brief Name of this test module
+ */
+#define BOOST_TEST_MODULE DFODecision_test // NOLINT
+
+#include "TRACE/trace.h"
+#include "boost/test/unit_test.hpp"
+
+using namespace dunedaq::dfmessages;
+
+BOOST_AUTO_TEST_SUITE(DFODecision_test)
+
+/**
+ * @brief Check that DFODecisions have appropriate Copy/Move semantics
+ */
+BOOST_AUTO_TEST_CASE(CopyAndMoveSemantics)
+{
+  BOOST_REQUIRE(std::is_copy_constructible_v<DFODecision>);
+  BOOST_REQUIRE(std::is_copy_assignable_v<DFODecision>);
+  BOOST_REQUIRE(std::is_move_constructible_v<DFODecision>);
+  BOOST_REQUIRE(std::is_move_assignable_v<DFODecision>);
+}
+
+BOOST_AUTO_TEST_CASE(SerDes_JSON)
+{
+  DFODecision dd;
+  dd.dfo_id = "TEST";
+  dd.acknowledged_completions = { 1, 2, 3, 4, 5 };
+
+  TriggerDecision td;
+  td.trigger_number = 1;
+  td.run_number = 2;
+  td.trigger_timestamp = 3;
+  td.trigger_type = 4;
+  td.readout_type = ReadoutType::kLocalized;
+
+  SourceID sid;
+  sid.subsystem = SourceID::Subsystem::kDetectorReadout;
+  sid.id= 1;
+  ComponentRequest cr;
+  cr.component = sid;
+  cr.window_begin = 5;
+  cr.window_end = 6;
+
+  SourceID another_sid;
+  another_sid.subsystem = SourceID::Subsystem::kDetectorReadout;
+  another_sid.id= 2;
+  ComponentRequest another_cr;
+  another_cr.component = another_sid;
+  another_cr.window_begin = 7;
+  another_cr.window_end = 8;
+
+  td.components.push_back(cr);
+  td.components.push_back(another_cr);
+  dd.trigger_decision = td;
+
+  auto bytes = dunedaq::serialization::serialize(dd, dunedaq::serialization::kJSON);
+
+  std::ostringstream ostr;
+  for (auto& b : bytes) {
+    ostr << static_cast<char>(b);
+  }
+  TLOG(TLVL_INFO) << "Serialized string: " << ostr.str();
+
+  DFODecision dd_deserialized = dunedaq::serialization::deserialize<DFODecision>(bytes);
+  TriggerDecision td_deserialized = dd_deserialized.trigger_decision;
+
+  BOOST_REQUIRE_EQUAL(dd.dfo_id, dd_deserialized.dfo_id);
+  BOOST_REQUIRE_EQUAL(dd.acknowledged_completions.size(), dd_deserialized.acknowledged_completions.size());
+  BOOST_REQUIRE_EQUAL(dd.acknowledged_completions[0], dd_deserialized.acknowledged_completions[0]);
+
+  BOOST_REQUIRE_EQUAL(td.trigger_number, td_deserialized.trigger_number);
+  BOOST_REQUIRE_EQUAL(td.run_number, td_deserialized.run_number);
+  BOOST_REQUIRE_EQUAL(td.trigger_timestamp, td_deserialized.trigger_timestamp);
+  BOOST_REQUIRE_EQUAL(td.trigger_type, td_deserialized.trigger_type);
+  BOOST_REQUIRE_EQUAL(static_cast<uint16_t>(td.readout_type),               // NOLINT(build/unsigned)
+                      static_cast<uint16_t>(td_deserialized.readout_type)); // NOLINT(build/unsigned)
+
+  BOOST_REQUIRE_EQUAL(td.components.size(), td_deserialized.components.size());
+
+  BOOST_REQUIRE_EQUAL(td.components[0].component.subsystem, td_deserialized.components[0].component.subsystem);
+  BOOST_REQUIRE_EQUAL(td.components[0].component.id, td_deserialized.components[0].component.id);
+  BOOST_REQUIRE_EQUAL(td.components[0].window_begin, td_deserialized.components[0].window_begin);
+  BOOST_REQUIRE_EQUAL(td.components[0].window_end, td_deserialized.components[0].window_end);
+
+  BOOST_REQUIRE_EQUAL(td.components[1].component.subsystem, td_deserialized.components[1].component.subsystem);
+  BOOST_REQUIRE_EQUAL(td.components[1].component.id, td_deserialized.components[1].component.id);
+  BOOST_REQUIRE_EQUAL(td.components[1].window_begin, td_deserialized.components[1].window_begin);
+  BOOST_REQUIRE_EQUAL(td.components[1].window_end, td_deserialized.components[1].window_end);
+}
+
+BOOST_AUTO_TEST_CASE(SerDes_MsgPack)
+{
+  DFODecision dd;
+  dd.dfo_id = "TEST";
+  dd.acknowledged_completions = { 1, 2, 3, 4, 5 };
+
+  TriggerDecision td;
+  td.trigger_number = 1;
+  td.run_number = 2;
+  td.trigger_timestamp = 3;
+  td.trigger_type = 4;
+  td.readout_type = ReadoutType::kLocalized;
+
+  SourceID sid;
+  sid.subsystem = SourceID::Subsystem::kDetectorReadout;
+  sid.id= 1;
+  ComponentRequest cr;
+  cr.component = sid;
+  cr.window_begin = 5;
+  cr.window_end = 6;
+
+  SourceID another_sid;
+  another_sid.subsystem = SourceID::Subsystem::kDetectorReadout;
+  another_sid.id= 2;
+  ComponentRequest another_cr;
+  another_cr.component = another_sid;
+  another_cr.window_begin = 7;
+  another_cr.window_end = 8;
+
+  td.components.push_back(cr);
+  td.components.push_back(another_cr);
+  dd.trigger_decision = td;
+
+  auto bytes = dunedaq::serialization::serialize(dd, dunedaq::serialization::kMsgPack);
+  TLOG(TLVL_INFO) << "MsgPack message size: " << bytes.size() << " bytes";
+  DFODecision dd_deserialized = dunedaq::serialization::deserialize<DFODecision>(bytes);
+  TriggerDecision td_deserialized = dd_deserialized.trigger_decision;
+
+  BOOST_REQUIRE_EQUAL(dd.dfo_id, dd_deserialized.dfo_id);
+  BOOST_REQUIRE_EQUAL(dd.acknowledged_completions.size(), dd_deserialized.acknowledged_completions.size());
+  BOOST_REQUIRE_EQUAL(dd.acknowledged_completions[0], dd_deserialized.acknowledged_completions[0]);
+
+  BOOST_REQUIRE_EQUAL(td.trigger_number, td_deserialized.trigger_number);
+  BOOST_REQUIRE_EQUAL(td.run_number, td_deserialized.run_number);
+  BOOST_REQUIRE_EQUAL(td.trigger_timestamp, td_deserialized.trigger_timestamp);
+  BOOST_REQUIRE_EQUAL(td.trigger_type, td_deserialized.trigger_type);
+  BOOST_REQUIRE_EQUAL(static_cast<uint16_t>(td.readout_type),               // NOLINT(build/unsigned)
+                      static_cast<uint16_t>(td_deserialized.readout_type)); // NOLINT(build/unsigned)
+
+  BOOST_REQUIRE_EQUAL(td.components.size(), td_deserialized.components.size());
+
+  BOOST_REQUIRE_EQUAL(td.components[0].component.subsystem, td_deserialized.components[0].component.subsystem);
+  BOOST_REQUIRE_EQUAL(td.components[0].component.id, td_deserialized.components[0].component.id);
+  BOOST_REQUIRE_EQUAL(td.components[0].window_begin, td_deserialized.components[0].window_begin);
+  BOOST_REQUIRE_EQUAL(td.components[0].window_end, td_deserialized.components[0].window_end);
+
+  BOOST_REQUIRE_EQUAL(td.components[1].component.subsystem, td_deserialized.components[1].component.subsystem);
+  BOOST_REQUIRE_EQUAL(td.components[1].component.id, td_deserialized.components[1].component.id);
+  BOOST_REQUIRE_EQUAL(td.components[1].window_begin, td_deserialized.components[1].window_begin);
+  BOOST_REQUIRE_EQUAL(td.components[1].window_end, td_deserialized.components[1].window_end);
+}
+
+BOOST_AUTO_TEST_SUITE_END()

unittest/DataflowHeartbeat_test.cxx

@@ -0,0 +1,87 @@
+/**
+ * @file DataflowHeartbeat_test.cxx DataflowHeartbeat class Unit Tests
+ *
+ * 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.
+ */
+
+#include "dfmessages/DataflowHeartbeat.hpp"
+
+/**
+ * @brief Name of this test module
+ */
+#define BOOST_TEST_MODULE DataflowHeartbeat_test // NOLINT
+
+#include "TRACE/trace.h"
+#include "boost/test/unit_test.hpp"
+
+#include <string>
+#include <vector>
+
+using namespace dunedaq::dfmessages;
+
+BOOST_AUTO_TEST_SUITE(DataflowHeartbeat_test)
+
+/**
+ * @brief Check that DataflowHeartbeats have appropriate Copy/Move semantics
+ */
+BOOST_AUTO_TEST_CASE(CopyAndMoveSemantics)
+{
+  BOOST_REQUIRE(std::is_copy_constructible_v<DataflowHeartbeat>);
+  BOOST_REQUIRE(std::is_copy_assignable_v<DataflowHeartbeat>);
+  BOOST_REQUIRE(std::is_move_constructible_v<DataflowHeartbeat>);
+  BOOST_REQUIRE(std::is_move_assignable_v<DataflowHeartbeat>);
+}
+
+BOOST_AUTO_TEST_CASE(DefaultConstruction)
+{
+  DataflowHeartbeat tdt;
+  BOOST_REQUIRE_EQUAL(tdt.run_number, TypeDefaults::s_invalid_run_number);
+  BOOST_REQUIRE_EQUAL(tdt.recent_completed_triggers.size(), 0);
+  BOOST_REQUIRE_EQUAL(tdt.outstanding_decisions.size(), 0);
+}
+
+BOOST_AUTO_TEST_CASE(SerDes_JSON)
+{
+  DataflowHeartbeat tdt;
+  tdt.run_number = 1;
+  tdt.recent_completed_triggers = {1,2,3};
+  tdt.outstanding_decisions = { 4, 5 };
+
+  auto bytes = dunedaq::serialization::serialize(tdt, dunedaq::serialization::kJSON);
+
+  std::ostringstream ostr;
+  for (auto& b : bytes) {
+    ostr << static_cast<char>(b);
+  }
+  TLOG(TLVL_INFO) << "Serialized string: " << ostr.str();
+
+  DataflowHeartbeat tdt_deserialized = dunedaq::serialization::deserialize<DataflowHeartbeat>(bytes);
+
+  BOOST_REQUIRE_EQUAL(tdt.run_number, tdt_deserialized.run_number);
+  BOOST_REQUIRE_EQUAL(tdt.recent_completed_triggers.size(), tdt_deserialized.recent_completed_triggers.size());
+  BOOST_REQUIRE_EQUAL(tdt.recent_completed_triggers[0], tdt_deserialized.recent_completed_triggers[0]);
+  BOOST_REQUIRE_EQUAL(tdt.outstanding_decisions.size(), tdt_deserialized.outstanding_decisions.size());
+  BOOST_REQUIRE_EQUAL(tdt.outstanding_decisions[0], tdt_deserialized.outstanding_decisions[0]);
+}
+
+BOOST_AUTO_TEST_CASE(SerDes_MsgPack)
+{
+  DataflowHeartbeat tdt;
+  tdt.run_number = 1;
+  tdt.recent_completed_triggers = { 1, 2, 3 };
+  tdt.outstanding_decisions = { 4, 5 };
+
+  auto bytes = dunedaq::serialization::serialize(tdt, dunedaq::serialization::kMsgPack);
+  TLOG(TLVL_INFO) << "MsgPack message size: " << bytes.size() << " bytes";
+  DataflowHeartbeat tdt_deserialized = dunedaq::serialization::deserialize<DataflowHeartbeat>(bytes);
+
+  BOOST_REQUIRE_EQUAL(tdt.run_number, tdt_deserialized.run_number);
+  BOOST_REQUIRE_EQUAL(tdt.recent_completed_triggers.size(), tdt_deserialized.recent_completed_triggers.size());
+  BOOST_REQUIRE_EQUAL(tdt.recent_completed_triggers[0], tdt_deserialized.recent_completed_triggers[0]);
+  BOOST_REQUIRE_EQUAL(tdt.outstanding_decisions.size(), tdt_deserialized.outstanding_decisions.size());
+  BOOST_REQUIRE_EQUAL(tdt.outstanding_decisions[0], tdt_deserialized.outstanding_decisions[0]);
+}
+
+BOOST_AUTO_TEST_SUITE_END()