DPL: record the workflow plugin of each data processor

DataProcessorInfo and DeviceExecution get the plugin which defines the
data processor, as <library>:<plugin>, and the workflow JSON gets a
"plugin" key in the metadata of the data processor. This is what a
driver needs to load the plugins of a workflow itself.

Plugins embedded in the executable, i.e. with an empty library, are not
recorded: their plugin stays empty and the key is not written, so the
dumps of all existing executables are unchanged. Internal data
processors added by DPL have no plugin either. Data processors imported
from an upstream workflow keep the plugin they had there.

Tested by TestDataProcessorInfoPlugin, a round trip of a workflow with
and without plugins, and test_Framework_test_WorkflowPluginsDump, which
dumps the workflow of the two test plugins.

Author: ktf

Framework/Core/CMakeLists.txt

@@ -472,6 +472,11 @@ if(workflowPluginsTest)
                                DPL_WORKFLOW_PLUGIN_LIBRARY=O2FrameworkTestWorkflowPlugin${p})
     add_dependencies(${workflowPluginsTest} O2FrameworkTestWorkflowPlugin${p})
   endforeach()
+  add_test(NAME test_Framework_test_WorkflowPluginsDump
+           COMMAND ${CMAKE_CURRENT_LIST_DIR}/test/test_WorkflowPluginsDump.sh $<TARGET_FILE:${workflowPluginsTest}>)
+  set_property(TEST test_Framework_test_WorkflowPluginsDump PROPERTY LABELS framework)
+  set_property(TEST test_Framework_test_WorkflowPluginsDump PROPERTY ENVIRONMENT
+               "DYLD_LIBRARY_PATH=${CMAKE_LIBRARY_OUTPUT_DIRECTORY}:$ENV{DYLD_LIBRARY_PATH};LD_LIBRARY_PATH=${CMAKE_LIBRARY_OUTPUT_DIRECTORY}:$ENV{LD_LIBRARY_PATH}")
 endif()
 
 # the test is compiled from the ExternalFairMQDeviceWorkflow test and run with

Framework/Core/include/Framework/DataProcessorInfo.h

@@ -32,6 +32,9 @@ struct DataProcessorInfo {
   std::vector<ConfigParamSpec> workflowOptions = {};
   /// The channels for a given dataprocessor
   std::vector<std::string> channels = {};
+  /// The workflow plugin which defines the DataProcessorSpec, as <library>:<plugin>.
+  /// Empty when it is the one embedded in the executable.
+  std::string plugin = "";
 };
 
 } // namespace o2

Framework/Core/include/Framework/DeviceExecution.h

@@ -11,6 +11,7 @@
 #ifndef FRAMEWORK_DEVICEEXECUTION_H
 #define FRAMEWORK_DEVICEEXECUTION_H
 
+#include <string>
 #include <vector>
 
 namespace o2
@@ -26,6 +27,8 @@ struct DeviceExecution {
   std::vector<char*> args;
   /// The environment to be passed to a given device
   std::vector<char*> environ;
+  /// The workflow plugin which defines the device, see DataProcessorInfo::plugin
+  std::string plugin;
 };
 
 } // namespace framework

Framework/Core/include/Framework/WorkflowDefinitionContext.h

@@ -23,6 +23,7 @@
 #include "Framework/ChannelConfigurationPolicy.h"
 #include "Framework/ServiceRegistry.h"
 #include <functional>
+#include <map>
 #include <memory>
 #include <vector>
 
@@ -47,6 +48,8 @@ struct WorkflowDefinitionContext {
   std::unique_ptr<ConfigParamRegistry> workflowOptionsRegistry{nullptr};
 
   o2::framework::WorkflowSpec specs;
+  /// The plugin which defines each data processor, if not embedded in the executable.
+  std::map<std::string, std::string> processorPlugins;
 };
 
 struct WorkflowDefinition {

Framework/Core/src/DeviceSpecHelpers.cxx

@@ -1749,6 +1749,7 @@ void DeviceSpecHelpers::prepareArguments(bool defaultQuiet, bool defaultStopped,
     }
     O2_SIGNPOST_END(device_spec_helpers, poid, "prepareArguments", "The following options are being forwarded to %{public}s: %{public}s",
                     spec.id.c_str(), str.str().c_str());
+    execution.plugin = pi->plugin;
   }
 }
 

Framework/Core/src/WorkflowSerializationHelpers.cxx

@@ -92,6 +92,7 @@ struct WorkflowImporter : public rapidjson::BaseReaderHandler<rapidjson::UTF8<>,
     IN_DATAPROCESSOR_INFO,
     IN_DATAPROCESSOR_INFO_NAME,
     IN_DATAPROCESSOR_INFO_EXECUTABLE,
+    IN_DATAPROCESSOR_INFO_PLUGIN,
     IN_DATAPROCESSOR_INFO_ARGS,
     IN_DATAPROCESSOR_INFO_ARG,
     IN_DATAPROCESSOR_INFO_CHANNELS,
@@ -264,6 +265,9 @@ struct WorkflowImporter : public rapidjson::BaseReaderHandler<rapidjson::UTF8<>,
       case State::IN_DATAPROCESSOR_INFO_EXECUTABLE:
         s << "IN_DATAPROCESSOR_INFO_EXECUTABLE";
         break;
+      case State::IN_DATAPROCESSOR_INFO_PLUGIN:
+        s << "IN_DATAPROCESSOR_INFO_PLUGIN";
+        break;
       case State::IN_DATAPROCESSOR_INFO_ARGS:
         s << "IN_DATAPROCESSOR_INFO_ARGS";
         break;
@@ -708,6 +712,8 @@ struct WorkflowImporter : public rapidjson::BaseReaderHandler<rapidjson::UTF8<>,
       push(State::IN_DATAPROCESSOR_INFO_NAME);
     } else if (in(State::IN_DATAPROCESSOR_INFO) && strncmp(str, "executable", length) == 0) {
       push(State::IN_DATAPROCESSOR_INFO_EXECUTABLE);
+    } else if (in(State::IN_DATAPROCESSOR_INFO) && strncmp(str, "plugin", length) == 0) {
+      push(State::IN_DATAPROCESSOR_INFO_PLUGIN);
     } else if (in(State::IN_DATAPROCESSOR_INFO) && strncmp(str, "cmdLineArgs", length) == 0) {
       push(State::IN_DATAPROCESSOR_INFO_ARGS);
     } else if (in(State::IN_DATAPROCESSOR_INFO) && strncmp(str, "workflowOptions", length) == 0) {
@@ -734,6 +740,9 @@ struct WorkflowImporter : public rapidjson::BaseReaderHandler<rapidjson::UTF8<>,
     } else if (in(State::IN_DATAPROCESSOR_INFO_EXECUTABLE)) {
       assert(metadata.size());
       metadata.back().executable = s;
+    } else if (in(State::IN_DATAPROCESSOR_INFO_PLUGIN)) {
+      assert(metadata.size());
+      metadata.back().plugin = s;
     } else if (in(State::IN_INPUT_BINDING)) {
       binding = s;
     } else if (in(State::IN_INPUT_ORIGIN)) {
@@ -1270,6 +1279,10 @@ void WorkflowSerializationHelpers::dump(std::ostream& out,
     w.String(info.name.c_str());
     w.Key("executable");
     w.String(info.executable.c_str());
+    if (!info.plugin.empty()) {
+      w.Key("plugin");
+      w.String(info.plugin.c_str());
+    }
     w.Key("cmdLineArgs");
     w.StartArray();
     for (auto& arg : info.cmdLineArgs) {

Framework/Core/src/runDataProcessing.cxx

@@ -1183,6 +1183,7 @@ struct WorkflowInfo {
   std::string executable;
   std::vector<std::string> args;
   std::vector<ConfigParamSpec> options;
+  std::map<std::string, std::string> plugins;
 };
 
 void gui_callback(uv_timer_s* ctx)
@@ -1696,7 +1697,8 @@ int runStateMachine(DataProcessorSpecs const& workflow,
               workflowInfo.executable,
               workflowInfo.args,
               workflowInfo.options,
-              channels});
+              channels,
+              workflowInfo.plugins.contains(device.name) ? workflowInfo.plugins.at(device.name) : ""});
         }
         break;
       case DriverState::MATERIALISE_WORKFLOW:
@@ -2989,6 +2991,9 @@ o2::framework::WorkflowDefinitionContext o2::framework::defineWorkflow(std::vect
       if (!inserted && it->second != di) {
         throw runtime_error_f("Data processor %s is defined by both %s and %s.", spec.name.c_str(), names[it->second].c_str(), names[di].c_str());
       }
+      if (!names[di].starts_with(":")) {
+        result.processorPlugins[spec.name] = names[di];
+      }
     }
     append(result.specs, context.specs);
     append(result.channelPolicies, context.channelPolicies);
@@ -3035,7 +3040,8 @@ int doMain(int argc, char** argv, o2::framework::WorkflowDefinitionContext& work
   WorkflowInfo currentWorkflow{
     argv[0],
     currentArgs,
-    currentWorkflowOptions};
+    currentWorkflowOptions,
+    workflowContext.processorPlugins};
 
   ProcessingPolicies processingPolicies;
   enum LogParsingHelpers::LogLevel minFailureLevel;

Framework/Core/test/test_WorkflowPluginsDump.sh

@@ -0,0 +1,10 @@
+#!/bin/sh -e
+# Dumps the workflow of the two test plugins and checks that it records,
+# for each data processor, the plugin which defines it.
+# usage: test_WorkflowPluginsDump.sh <o2-test-framework-WorkflowPlugins>
+A=O2FrameworkTestWorkflowPluginA:TestWorkflowPluginA
+B=O2FrameworkTestWorkflowPluginB:TestWorkflowPluginB
+dump=$("$1" -b --workflow-plugin $A,$B --dump-workflow < /dev/null)
+echo "$dump" | grep -q "\"plugin\": \"$A\""
+echo "$dump" | grep -q "\"plugin\": \"$B\""
+test "$(echo "$dump" | grep -c '"plugin"')" -eq 2

Framework/Core/test/test_WorkflowSerialization.cxx

@@ -169,3 +169,42 @@ TEST_CASE("TestInputOutputSpecMetadata")
   REQUIRE(wso[0].outputs[0].metadata[0] == wsi[0].outputs[0].metadata[0]);
   REQUIRE(wso[0].outputs[0].metadata[1] == wsi[0].outputs[0].metadata[1]);
 }
+
+/// The plugin of a data processor is only written when there is one,
+/// so that dumps of executables with an embedded plugin do not change.
+TEST_CASE("TestDataProcessorInfoPlugin")
+{
+  using namespace o2::framework;
+  WorkflowSpec w0{
+    DataProcessorSpec{.name = "A", .outputs = {OutputSpec{"TST", "A"}}},
+    DataProcessorSpec{.name = "B", .inputs = {InputSpec{"a", "TST", "A"}}}};
+
+  std::vector<DataProcessorInfo> dataProcessorInfoOut{
+    {.name = "A", .executable = "o2-dpl-workflow", .plugin = "O2SomePlugins:A"},
+    {.name = "B", .executable = "o2-dpl-workflow"},
+  };
+  CommandInfo commandInfoOut{"o2-dpl-workflow -b"};
+
+  std::ostringstream firstDump;
+  WorkflowSerializationHelpers::dump(firstDump, w0, dataProcessorInfoOut, commandInfoOut);
+  auto json = firstDump.str();
+  REQUIRE(json.find(R"("plugin": "O2SomePlugins:A")") != std::string::npos);
+  // Only A has a plugin.
+  REQUIRE(json.find(R"("plugin")") == json.rfind(R"("plugin")"));
+
+  std::vector<DataProcessorInfo> dataProcessorInfoIn{};
+  CommandInfo commandInfoIn;
+  std::istringstream is;
+  is.str(json);
+  WorkflowSpec w1;
+  WorkflowSerializationHelpers::import(is, w1, dataProcessorInfoIn, commandInfoIn);
+  REQUIRE(dataProcessorInfoIn.size() == 2);
+  REQUIRE(dataProcessorInfoIn[0].plugin == "O2SomePlugins:A");
+  REQUIRE(dataProcessorInfoIn[0].executable == "o2-dpl-workflow");
+  REQUIRE(dataProcessorInfoIn[1].plugin.empty());
+  REQUIRE(dataProcessorInfoIn[1].executable == "o2-dpl-workflow");
+
+  std::ostringstream secondDump;
+  WorkflowSerializationHelpers::dump(secondDump, w1, dataProcessorInfoIn, commandInfoIn);
+  REQUIRE(json == secondDump.str());
+}