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

Configuration objects for DataMoveCallbackRegistry

Author: ShyamB97

apps/GraphBuilder.cpp

@@ -23,9 +23,15 @@
 #include "appmodel/TriggerApplication.hpp"
 #include "appmodel/appmodelIssues.hpp"
 
+#include "appmodel/DataHandlerModule.hpp"
+#include "appmodel/DataReaderModule.hpp"
+#include "appmodel/DataMoveCallbackConf.hpp"
+#include "appmodel/SocketDataWriterModule.hpp"
+
 #include "conffwk/Configuration.hpp"
 #include "conffwk/Schema.hpp"
 #include "confmodel/Connection.hpp"
+#include "confmodel/NetworkConnection.hpp"
 #include "confmodel/DaqModule.hpp"
 #include "confmodel/Session.hpp"
 #include "ers/ers.hpp"
@@ -425,7 +431,7 @@ GraphBuilder::find_objects_and_connections(const ConfigObject& object)
       if (m_root_object_kind == ObjectKind::kSession || m_root_object_kind == ObjectKind::kSegment) {
         allowed_conns = { "NetworkConnection" };
       } else if (m_root_object_kind == ObjectKind::kApplication || m_root_object_kind == ObjectKind::kModule) {
-        allowed_conns = { "NetworkConnection", "Queue", "QueueWithSourceId" };
+        allowed_conns = { "NetworkConnection", "Queue", "QueueWithSourceId", "DataMoveCallbackConf" };
       }
 
       for (const auto& module : modules) {
@@ -436,7 +442,12 @@ GraphBuilder::find_objects_and_connections(const ConfigObject& object)
           // connection is a network or a queue, so include the
           // class name in the std::string key
 
-          const std::string key = in->config_object().UID() + "@" + in->config_object().class_name();
+          std::string key = in->config_object().UID() + "@" + in->config_object().class_name();
+
+          if (in->config_object().class_name() == "NetworkConnection") {
+            auto innc = in->cast<dunedaq::confmodel::NetworkConnection>();
+            key += "@" +  innc->get_connection_type();
+          }
 
           if (std::ranges::find(allowed_conns, in->config_object().class_name()) != allowed_conns.end()) {
             m_incoming_connections[key].push_back(object.UID());
@@ -446,14 +457,56 @@ GraphBuilder::find_objects_and_connections(const ConfigObject& object)
 
         for (auto out : module->get_outputs()) {
 
-          const std::string key = out->config_object().UID() + "@" + out->config_object().class_name();
+          std::string key = out->config_object().UID() + "@" + out->config_object().class_name();
+
+          if (out->config_object().class_name() == "NetworkConnection") {
+            auto outnc = out->cast<dunedaq::confmodel::NetworkConnection>();
+            key += "@" + outnc->get_connection_type();
+          }
 
           if (std::ranges::find(allowed_conns, out->config_object().class_name()) != allowed_conns.end()) {
             m_outgoing_connections[key].push_back(object.UID());
             m_outgoing_connections[key].push_back(module->UID());
           }
         }
 
+        // Look for DataMoveCallbackConfs
+        auto datareader = module->cast<dunedaq::appmodel::DataReaderModule>();
+        auto datahandler = module->cast<dunedaq::appmodel::DataHandlerModule>();
+        auto socketwriter = module->cast<dunedaq::appmodel::SocketDataWriterModule>();
+
+        if (datareader != nullptr) {
+          for (auto& out : datareader->get_raw_data_callbacks()) {
+            const std::string key = out->config_object().UID() + "@" + out->config_object().class_name();
+            if (std::ranges::find(allowed_conns, out->config_object().class_name()) != allowed_conns.end()) {
+              m_outgoing_connections[key].push_back(object.UID());
+              m_outgoing_connections[key].push_back(module->UID());
+            }
+          }
+        }
+        if (datahandler != nullptr) {
+          auto in = datahandler->get_raw_data_callback();
+          if (in != nullptr) {
+            const std::string key = in->config_object().UID() + "@" + in->config_object().class_name();
+
+            if (std::ranges::find(allowed_conns, in->config_object().class_name()) != allowed_conns.end()) {
+              m_incoming_connections[key].push_back(object.UID());
+              m_incoming_connections[key].push_back(module->UID());
+            }
+          }
+        }
+        if (socketwriter != nullptr) {
+          auto in = socketwriter->get_raw_data_callback();
+          if (in != nullptr) {
+            const std::string key = in->config_object().UID() + "@" + in->config_object().class_name();
+
+            if (std::ranges::find(allowed_conns, in->config_object().class_name()) != allowed_conns.end()) {
+              m_incoming_connections[key].push_back(object.UID());
+              m_incoming_connections[key].push_back(module->UID());
+            }
+          }
+        }
+
         if (std::ranges::find(m_included_classes.at(m_root_object_kind), "Module") !=
             m_included_classes.at(m_root_object_kind).end()) {
           find_objects_and_connections(module->config_object());
@@ -518,9 +571,6 @@ GraphBuilder::construct_graph(std::string root_obj_uid)
   for (auto& possible_sender_object : m_objects_for_graph | std::views::values) {
     for (auto& receiver_info : possible_sender_object.receiving_object_infos) {
 
-      auto at_pos = receiver_info.connection_name.find("@");
-      const std::string connection_label = receiver_info.connection_name.substr(0, at_pos);
-
       // If we're plotting at the level of a session or segment,
       // show the connections as between applications; if we're
       // doing this for a single application, show them entering and
@@ -540,7 +590,7 @@ GraphBuilder::construct_graph(std::string root_obj_uid)
 
       boost::add_edge(possible_sender_object.vertex_in_graph,
                       m_objects_for_graph.at(receiver_info.receiver_label).vertex_in_graph,
-                      { connection_label },
+                      { receiver_info.connection_name },
                       m_graph)
         .first;
     }
@@ -608,7 +658,16 @@ GraphBuilder::write_graph(const std::string& outputfilename) const
                                                                    { ObjectKind::kModule, { "rectangle", "red" } } };
 
   std::string dotfile_slurped = outputstream.str();
-  std::vector<std::string> legend_entries{};
+  std::vector<std::string> legend_entries{
+    "legendGA [label=<<font color=\"black\"><b><i>&#10230; Network Connection</i></b></font>>, shape=plaintext];",
+    "legendGB [label=<<font color=\"blue\"><b><i>&#10230; Pub/Sub Network</i></b></font>>, shape=plaintext];"
+  };
+  std::vector<std::string> internal_legend_entries{
+    "legendGC [label=<<font color=\"green\"><b><i>&#10230; Data Move Callback</i></b></font>>, shape=plaintext];",
+    "legendGD [label=<<font color=\"red\"><b><i>&#10230; Queue</i></b></font>>, shape=plaintext];",
+    "legendGE [label=<<font color=\"orange\"><b><i>&#10230; Queue w/ Source ID</i></b></font>>, shape=plaintext];"
+  };
+  bool internal_legend_added = false;
   std::vector<std::string> legend_ordering_code{};
 
   for (auto& eo : m_objects_for_graph | std::views::values) {
@@ -665,6 +724,12 @@ GraphBuilder::write_graph(const std::string& outputfilename) const
       case ObjectKind::kModule:
         add_vertex_info();
         add_legend_entry('D', "DAQModule");
+        if (!internal_legend_added) {
+          legend_entries.insert(legend_entries.end(),
+                                internal_legend_entries.begin(),
+                                internal_legend_entries.end());
+          internal_legend_added = true;
+        }
         break;
       case ObjectKind::kIncomingExternal:
         legendstr
@@ -741,6 +806,23 @@ GraphBuilder::write_graph(const std::string& outputfilename) const
     pos += (unlabeled_edge + edge_modifier).length();
   }
 
+  // Replace the connection types with color information
+  std::vector<std::pair<std::string, std::string>> connection_colors = { { "@NetworkConnection@kSendRecv\"", "\", color=black" },
+                                                                         { "@NetworkConnection@kPubSub\"", "\", color=blue" },
+                                                                         { "@QueueWithSourceId\"", "\", color=orange" },
+                                                                         { "@Queue\"", "\", color=red" },
+                                                                         { "@DataMoveCallbackConf\"",
+                                                                           "\", color=green" } };
+  for (auto& color_pair : connection_colors) {
+    auto conn_type = color_pair.first;
+    auto color_info = color_pair.second;
+    pos = 0;
+    while ((pos = dotfile_slurped.find(conn_type, pos)) != std::string::npos) {
+      dotfile_slurped.replace(pos, conn_type.length(), color_info);
+      pos += color_info.length();
+    }
+  }
+
   // And now with all the edits made to the contents of the DOT code, write it to file
 
   std::ofstream outputfile;

python/daqconf/generate.py

@@ -548,44 +548,44 @@ def generate_readout(
                 linkhandler = db.get_dal(
                     class_name="DataHandlerConf", uid="def-pds-stream-link-handler"
                 )
-                det_q = db.get_dal(
-                    class_name="QueueConnectionRule", uid="pds-stream-raw-data-rule"
+                cb_desc = db.get_dal(
+                    class_name="DataMoveCallbackDescriptor", uid="pds-stream-raw-input"
                 )
             else:
                 linkhandler = db.get_dal(
                     class_name="DataHandlerConf", uid="def-pds-link-handler"
                 )
-                det_q = db.get_dal(
-                    class_name="QueueConnectionRule", uid="pds-raw-data-rule"
+                cb_desc = db.get_dal(
+                    class_name="DataMoveCallbackDescriptor", uid="pds-raw-input"
                 )
 
         elif det_id == 3 or det_id == 10:
             linkhandler = db.get_dal(
                 class_name="DataHandlerConf", uid="def-link-handler"
             )
-            det_q = db.get_dal(
-                class_name="QueueConnectionRule", uid="wib-eth-raw-data-rule"
+            cb_desc = db.get_dal(
+                class_name="DataMoveCallbackDescriptor", uid="wib-eth-raw-input"
             )
         elif det_id == 11:
             linkhandler = db.get_dal(
                 class_name="DataHandlerConf", uid="def-tde-link-handler"
             )
-            det_q = db.get_dal(
-                class_name="QueueConnectionRule", uid="tde-raw-data-rule"
+            cb_desc = db.get_dal(
+                class_name="DataMoveCallbackDescriptor", uid="tde-raw-input"
             )
         elif det_id == 12:
             linkhandler = db.get_dal(
                 class_name="DataHandlerConf", uid="def-crt-bern-link-handler"
             )
-            det_q = db.get_dal(
-                class_name="QueueConnectionRule", uid="crt-bern-raw-data-rule"
+            cb_desc = db.get_dal(
+                class_name="DataMoveCallbackDescriptor", uid="crt-bern-raw-input"
             )
         elif det_id == 13:
             linkhandler = db.get_dal(
                 class_name="DataHandlerConf", uid="def-crt-grenoble-link-handler"
             )
-            det_q = db.get_dal(
-                class_name="QueueConnectionRule", uid="crt-grenoble-raw-data-rule"
+            cb_desc = db.get_dal(
+                class_name="DataMoveCallbackDescriptor", uid="crt-grenoble-raw-input"
             )
 
         hostnum = appnum % len(hosts)
@@ -729,7 +729,7 @@ def generate_readout(
             runs_on=host,
             detector_connections=[connection],
             network_rules=netrules,
-            queue_rules=qrules + [det_q],
+            queue_rules=qrules,
             link_handler=linkhandler,
             data_reader=datareader,
             fragment_aggregator=fragagg,
@@ -739,6 +739,7 @@ def generate_readout(
             uses=rohw,
             exposes_service=[daqapp_control, dataRequests, timeSyncs],
             action_plans=[readout_start, readout_stop],
+            callback_desc=cb_desc,
         )
         if tpg_enabled:
             ru.tp_handler = tphandler