Merge pull request #274 from DUNE-DAQ/eflumerf/DMCRConfiguration

Configuration objects for DataMoveCallbackRegistry

Author: ShyamB97

schema/appmodel/application.schema.xml

@@ -80,7 +80,7 @@
 
 <oks-schema>
 
-<info name="" type="" num-of-items="66" oks-format="schema" oks-version="862f2957270" created-by="gjc" created-on="thinkpad" creation-time="20230616T091343" last-modified-by="eflumerf" last-modified-on="ironvirt9.mshome.net" last-modification-time="20251104T211957"/>
+<info name="" type="" num-of-items="69" oks-format="schema" oks-version="862f2957270" created-by="gjc" created-on="thinkpad" creation-time="20230616T091343" last-modified-by="eflumerf" last-modified-on="ironvirt9.mshome.net" last-modification-time="20260203T131505"/>
 
 <include>
  <file path="schema/confmodel/dunedaq.schema.xml"/>
@@ -94,6 +94,7 @@
   <relationship name="data_writers" class-type="SocketDataWriterConf" low-cc="one" high-cc="many" is-composite="no" is-exclusive="no" is-dependent="no"/>
   <relationship name="data_reader" class-type="DataReaderConf" low-cc="one" high-cc="one" is-composite="no" is-exclusive="no" is-dependent="no"/>
   <relationship name="detector_connections" class-type="DetectorToDaqConnection" low-cc="zero" high-cc="many" is-composite="no" is-exclusive="no" is-dependent="no"/>
+  <relationship name="callback_desc" class-type="DataMoveCallbackDescriptor" low-cc="one" high-cc="one" is-composite="no" is-exclusive="no" is-dependent="no"/>
   <method name="generate_modules" description="Generate daq module dal objects for streams of the CRTReaderApplication on the fly">
    <method-implementation language="c++" prototype="void generate_modules(const confmodel::Session*) const override" body=""/>
   </method>
@@ -167,6 +168,7 @@
   <attribute name="emulation_mode" type="bool" init-value="false" is-not-null="yes"/>
   <attribute name="post_processing_enabled" type="bool" init-value="true"/>
   <relationship name="geo_id" class-type="GeoId" low-cc="zero" high-cc="one" is-composite="no" is-exclusive="no" is-dependent="no"/>
+  <relationship name="raw_data_callback" description="Configuration for raw data callback" class-type="DataMoveCallbackConf" low-cc="zero" high-cc="one" is-composite="no" is-exclusive="no" is-dependent="no"/>
   <relationship name="module_configuration" class-type="DataHandlerConf" low-cc="one" high-cc="one" is-composite="no" is-exclusive="no" is-dependent="no"/>
  </class>
 
@@ -187,6 +189,7 @@
   <superclass name="DaqModule"/>
   <relationship name="configuration" class-type="DataReaderConf" low-cc="one" high-cc="one" is-composite="no" is-exclusive="no" is-dependent="no"/>
   <relationship name="connections" class-type="DetectorToDaqConnection" low-cc="one" high-cc="many" is-composite="yes" is-exclusive="no" is-dependent="no"/>
+  <relationship name="raw_data_callbacks" description="Configurations for raw data callbacks" class-type="DataMoveCallbackConf" low-cc="one" high-cc="many" is-composite="no" is-exclusive="no" is-dependent="no"/>
  </class>
 
  <class name="DataRecorderConf">
@@ -403,6 +406,16 @@
   <attribute name="data_type" description="string identifying type of data transferred through this queue" type="string" is-not-null="yes"/>
  </class>
 
+ <class name="DataMoveCallbackConf">
+  <attribute name="source_id" type="u32" init-value="0" is-not-null="yes"/>
+  <attribute name="data_type" description="Name of the output data type. Should be defined via a call to DUNE_DAQ_TYPESTRING" type="string" init-value="WIBEthFrame"/>
+ </class>
+
+ <class name="DataMoveCallbackDescriptor">
+  <attribute name="uid_base" description="Base for callback UID string. May be combined with a source id" type="string" is-not-null="yes"/>
+  <attribute name="data_type" description="string identifying type of data passed as a callback parameter" type="string" is-not-null="yes"/>
+ </class>
+
  <class name="ReadoutApplication">
   <superclass name="SmartDaqApplication"/>
   <superclass name="ResourceSetDisableAND"/>
@@ -417,6 +430,7 @@
   <relationship name="data_recorder" class-type="DataRecorderConf" low-cc="zero" high-cc="one" is-composite="no" is-exclusive="no" is-dependent="no"/>
   <relationship name="fragment_aggregator" class-type="FragmentAggregatorConf" low-cc="one" high-cc="one" is-composite="no" is-exclusive="no" is-dependent="no"/>
   <relationship name="detector_connections" description="The list of detector channels to be read out by this readout application." class-type="DetectorToDaqConnection" low-cc="zero" high-cc="many" is-composite="no" is-exclusive="no" is-dependent="no"/>
+  <relationship name="callback_desc" class-type="DataMoveCallbackDescriptor" low-cc="one" high-cc="one" is-composite="no" is-exclusive="no" is-dependent="no"/>
   <method name="generate_modules" description="Generate daq module dal objects for streams of thie ReadoutApplication on the fly">
    <method-implementation language="c++" prototype="void generate_modules(const confmodel::Session*) const override" body=""/>
   </method>
@@ -514,6 +528,7 @@
   <superclass name="DaqModule"/>
   <relationship name="configuration" class-type="SocketDataWriterConf" low-cc="one" high-cc="one" is-composite="no" is-exclusive="no" is-dependent="no"/>
   <relationship name="connections" class-type="NetworkDetectorToDaqConnection" low-cc="one" high-cc="many" is-composite="yes" is-exclusive="no" is-dependent="no"/>
+  <relationship name="raw_data_callback" description="Configuration for raw data callback" class-type="DataMoveCallbackConf" low-cc="one" high-cc="one" is-composite="no" is-exclusive="no" is-dependent="no"/>
  </class>
 
  <class name="SourceIDConf">

src/CRTReaderApplication.cpp

@@ -15,6 +15,7 @@
 #include "appmodel/DataReaderConf.hpp"
 #include "appmodel/SocketWriterConf.hpp"
 #include "appmodel/SocketWriterModule.hpp"
+#include "appmodel/DataMoveCallbackConf.hpp"
 #include "appmodel/QueueConnectionRule.hpp"
 #include "appmodel/QueueDescriptor.hpp"
 
@@ -66,24 +67,12 @@ CRTReaderApplication::generate_modules(const confmodel::Session* session) const
   }    
 
   //
-  // Process the queue rules looking for inputs to our DL/TP handler modules
+  // Get the callback descriptor
   //
-  const QueueDescriptor* dlh_input_qdesc = nullptr;
-
-  for (auto rule : get_queue_rules()) {
-    auto destination_class = rule->get_destination_class();
-    auto data_type = rule->get_descriptor()->get_data_type();
-    // Why datahander here? It is the base class for several DataHandler types (e.g. FDDataHandlerModule,
-    // SNBDataHandlerModule)
-    if (destination_class == "DataHandlerModule") {
-      if (data_type != "DataRequest") {
-        dlh_input_qdesc = rule->get_descriptor();
-      }
-    }
-  }
+  const DataMoveCallbackDescriptor* raw_data_callback_desc = get_callback_desc();
 
-  if (dlh_input_qdesc == nullptr) {
-    throw(BadConf(ERS_HERE, "No data link handler input queue descriptor given"));
+  if (raw_data_callback_desc == nullptr) {
+    throw(BadConf(ERS_HERE, "No Raw Data Callback descriptor given"));
   }
 
   //
@@ -93,7 +82,7 @@ CRTReaderApplication::generate_modules(const confmodel::Session* session) const
   // Loop over the detector to daq connections and generate one data reader per connection
 
   // Collect all streams
-  std::map<uint32_t, const confmodel::Connection*> data_queues_by_sid;
+  std::map<uint32_t, const appmodel::DataMoveCallbackConf*> callback_confs_by_sid;
 
   uint16_t conn_idx = 0;
 
@@ -121,16 +110,15 @@ CRTReaderApplication::generate_modules(const confmodel::Session* session) const
       enabled_det_streams.push_back(stream);
     }
 
-    // Create the raw data queues
-    std::vector<const conffwk::ConfigObject*> data_queue_objs;
-    // keep a map for convenience
+    // Create the raw data callbacks
+    std::vector<const conffwk::ConfigObject*> raw_data_callback_objs;
 
     // Create data queues
     for (auto ds : enabled_det_streams) {
-      conffwk::ConfigObject queue_obj = obj_fac.create_queue_sid_obj(dlh_input_qdesc, ds);
-      const auto* data_queue = obj_fac.get_dal<confmodel::Connection>(queue_obj.UID());
-      data_queue_objs.push_back(&data_queue->config_object());
-      data_queues_by_sid[ds->get_source_id()] = data_queue;
+      conffwk::ConfigObject callback_obj = obj_fac.create_callback_sid_obj(raw_data_callback_desc, ds->get_source_id());
+      const auto* callback_conf = obj_fac.get_dal<DataMoveCallbackConf>(callback_obj.UID());
+      raw_data_callback_objs.push_back(&callback_conf->config_object());
+      callback_confs_by_sid[ds->get_source_id()] = callback_conf;
     }
 
     //-----------------------------------------------------------------
@@ -150,8 +138,8 @@ CRTReaderApplication::generate_modules(const confmodel::Session* session) const
 
     // Populate configuration and interfaces (leave output queues for later)
     reader_obj.set_obj("configuration", &reader_conf->config_object());
-    reader_obj.set_objs("connections", {&d2d_conn->config_object()});
-    reader_obj.set_objs("outputs", data_queue_objs);
+    reader_obj.set_objs("connections", { &d2d_conn->config_object() });
+    reader_obj.set_objs("raw_data_callbacks", raw_data_callback_objs);
 
     modules.push_back(obj_fac.get_dal<confmodel::DaqModule>(reader_obj.UID()));
 
@@ -180,7 +168,6 @@ CRTReaderApplication::generate_modules(const confmodel::Session* session) const
       // Populate configuration and interfaces
       writer_obj.set_obj("configuration", &writer_conf->config_object());
       writer_obj.set_objs("connections", {&d2d_conn->config_object()});
-      writer_obj.set_objs("inputs", data_queue_objs);
 
       modules.push_back(obj_fac.get_dal<confmodel::DaqModule>(writer_obj.UID()));
     }

src/ConfigObjectFactory.cpp

@@ -22,7 +22,7 @@ ConfigObjectFactory::~ConfigObjectFactory() {
     oks::OksFile::set_nolock_mode(false);
 }
 
-conffwk::ConfigObject 
+conffwk::ConfigObject
 ConfigObjectFactory::create(const std::string& class_name,
                                      const std::string& id) const {
     conffwk::ConfigObject cfg_obj;
@@ -58,12 +58,24 @@ ConfigObjectFactory::create_queue_sid_obj(const QueueDescriptor* qdesc, uint32_t
 }
 
 //---
-conffwk::ConfigObject 
+conffwk::ConfigObject
 ConfigObjectFactory::create_queue_sid_obj(const QueueDescriptor* qdesc,
                                      const confmodel::DetectorStream* stream) const {
 return create_queue_sid_obj(qdesc, stream->get_source_id());
 }
 
+//---
+conffwk::ConfigObject
+ConfigObjectFactory::create_callback_sid_obj(const DataMoveCallbackDescriptor* cdesc, uint32_t src_id) const
+{
+  std::string rdc_uid(fmt::format("{}{}", cdesc->get_uid_base(), src_id));
+  auto rdc_obj = create("DataMoveCallbackConf", rdc_uid);
+
+  rdc_obj.set_by_val<std::string>("data_type", cdesc->get_data_type());
+  rdc_obj.set_by_val<uint32_t>("source_id", src_id);
+
+  return rdc_obj;
+}
 //---
 
 /**

src/ConfigObjectFactory.hpp

@@ -13,6 +13,7 @@
 #include "appmodel/appmodelIssues.hpp"
 #include "appmodel/NetworkConnectionDescriptor.hpp"
 #include "appmodel/QueueDescriptor.hpp"
+#include "appmodel/DataMoveCallbackDescriptor.hpp"
 #include "appmodel/SmartDaqApplication.hpp"
 
 #include "conffwk/ConfigObject.hpp"
@@ -53,7 +54,10 @@ class ConfigObjectFactory
   //---
   [[nodiscard]] conffwk::ConfigObject
   create_queue_sid_obj(const QueueDescriptor* qdesc, const confmodel::DetectorStream* stream) const;
+
   //---
+  [[nodiscard]] conffwk::ConfigObject create_callback_sid_obj(const DataMoveCallbackDescriptor* cdesc,
+                                                              uint32_t src_id) const;
 
   /**
    * \brief Helper function that gets a network connection config

src/NP02ReadoutApplication.cpp

@@ -36,6 +36,7 @@
 #include "confmodel/Service.hpp"
 
 #include "appmodel/SourceIDConf.hpp"
+#include "appmodel/DataMoveCallbackConf.hpp"
 #include "appmodel/DataReaderModule.hpp"
 #include "appmodel/DataReaderConf.hpp"
 #include "appmodel/DataRecorderModule.hpp"
@@ -76,7 +77,7 @@ NP02ReadoutApplication::generate_modules(const confmodel::Session* session) cons
   TLOG_DEBUG(6) << "Generating modules for application " << this->UID();
 
   ConfigObjectFactory obj_fac(this);
-  
+
   //
   // Extract basic configuration objects
   //
@@ -106,7 +107,6 @@ NP02ReadoutApplication::generate_modules(const confmodel::Session* session) cons
   //
   // Process the queue rules looking for inputs to our DL/TP handler modules
   //
-  const QueueDescriptor* dlh_input_qdesc = nullptr;
   const QueueDescriptor* dlh_reqinput_qdesc = nullptr;
   const QueueDescriptor* tp_input_qdesc = nullptr;
   // const QueueDescriptor* tpReqInputQDesc = nullptr;
@@ -121,8 +121,6 @@ NP02ReadoutApplication::generate_modules(const confmodel::Session* session) cons
         dlh_reqinput_qdesc = rule->get_descriptor();
       } else if ((data_type == "TriggerPrimitive" || data_type == "TriggerPrimitiveVector") && get_tp_generation_enabled()) {
         tp_input_qdesc = rule->get_descriptor();
-      } else {
-        dlh_input_qdesc = rule->get_descriptor();
       }
     } else if (destination_class == "FragmentAggregatorModule") {
       fa_output_qdesc = rule->get_descriptor();
@@ -159,6 +157,15 @@ NP02ReadoutApplication::generate_modules(const confmodel::Session* session) cons
   std::vector<const confmodel::Connection*> req_queues;
   conffwk::ConfigObject frag_queue_obj = obj_fac.create_queue_obj(fa_output_qdesc);
 
+  //
+  // Get the callback descriptor
+  //
+  const DataMoveCallbackDescriptor* raw_data_callback_desc = get_callback_desc();
+
+  if (raw_data_callback_desc == nullptr) {
+    throw(BadConf(ERS_HERE, "No Raw Data Callback descriptor given"));
+  }
+
   //
   // Scan Detector 2 DAQ connections to extract sender, receiver and stream information
   //
@@ -170,11 +177,11 @@ NP02ReadoutApplication::generate_modules(const confmodel::Session* session) cons
 
   // Collect all streams
   std::vector<std::pair<int16_t, const confmodel::DetectorStream*>> all_enabled_det_streams;
-  std::map<uint32_t, const confmodel::Connection*> data_queues_by_sid;
+  std::map<uint32_t, const appmodel::DataMoveCallbackConf*> callback_confs_by_sid;
 
   std::vector<const conffwk::ConfigObject*> d2d_conn_objs;
   uint16_t conn_idx = 0;
-  
+
 
   std::set<int16_t> numas;
   for (auto d2d_conn : get_detector_connections()) {
@@ -264,19 +271,18 @@ NP02ReadoutApplication::generate_modules(const confmodel::Session* session) cons
     reader_obj.set_obj("configuration", &reader_conf->config_object());
     reader_obj.set_objs("connections", d2d_conn_objs);
 
-    // Create the raw data queues
-    std::vector<const conffwk::ConfigObject*> data_queue_objs;
-    // keep a map for convenience
+    // Create the raw data callbacks
+    std::vector<const conffwk::ConfigObject*> raw_data_callback_objs;
 
     // Create data queues
     for (auto& [numa, ds] : all_enabled_det_streams) {
-      conffwk::ConfigObject queue_obj = obj_fac.create_queue_sid_obj(dlh_input_qdesc, ds);
-      const auto* data_queue = obj_fac.get_dal<confmodel::Connection>(queue_obj.UID());
-      data_queue_objs.push_back(&data_queue->config_object());
-      data_queues_by_sid[ds->get_source_id()] = data_queue;
+      conffwk::ConfigObject callback_obj = obj_fac.create_callback_sid_obj(raw_data_callback_desc, ds->get_source_id());
+      const auto* callback_conf = obj_fac.get_dal<DataMoveCallbackConf>(callback_obj.UID());
+      raw_data_callback_objs.push_back(&callback_conf->config_object());
+      callback_confs_by_sid[ds->get_source_id()] = callback_conf;
     }
 
-    reader_obj.set_objs("outputs", data_queue_objs);
+    reader_obj.set_objs("raw_data_callbacks", raw_data_callback_objs);
 
     modules.push_back(obj_fac.get_dal<confmodel::DaqModule>(reader_obj.UID()));
 
@@ -384,10 +390,9 @@ NP02ReadoutApplication::generate_modules(const confmodel::Session* session) cons
     dlh_obj.set_by_val<bool>("emulation_mode", emulation_mode);
     dlh_obj.set_obj("geo_id", &ds->get_geo_id()->config_object());
     dlh_obj.set_obj("module_configuration", &numa_dhlconf_map[numa]);
-    std::vector<const conffwk::ConfigObject*> dlh_ins, dlh_outs;
+    dlh_obj.set_obj("raw_data_callback", &callback_confs_by_sid[sid]->config_object());
 
-    // Add datalink-handler queue to the inputs
-    dlh_ins.push_back(&data_queues_by_sid.at(sid)->config_object());
+    std::vector<const conffwk::ConfigObject*> dlh_ins, dlh_outs;
 
     // Create request queue
     conffwk::ConfigObject req_queue_obj = obj_fac.create_queue_sid_obj(dlh_reqinput_qdesc, ds);
@@ -478,6 +483,6 @@ NP02ReadoutApplication::generate_modules(const confmodel::Session* session) cons
   obj_fac.update_modules(modules);
 }
 
-  
+
 }
 }