DPL: optionally fork data processors without exec

Framework/Core/CMakeLists.txt

@@ -268,6 +268,7 @@ add_executable(o2-test-framework-core
               test/test_TopologyPolicies.cxx
               test/test_Variants.cxx
               test/test_WorkflowHelpers.cxx
+              test/test_WorkflowDefinition.cxx
               test/test_WorkflowSerialization.cxx
               test/test_DataOutputDirector.cxx
               test/unittest_SimpleOptionsRetriever.cxx
@@ -447,6 +448,35 @@ o2_add_test(
     "--global-config consumer-config --local-option hello-aliceo2 --a-boolean3 --an-int2 20 --a-double2 22. --an-int64-2 50000000000000"
   )
 
+# runs the workflows of two plugins built in DSOs, loaded with --workflow-plugin
+o2_add_test(
+  WorkflowPlugins NAME test_Framework_test_WorkflowPlugins
+  SOURCES test/test_WorkflowPlugins.cxx
+  COMPONENT_NAME Framework
+  LABELS framework workflow
+  TIMEOUT 30
+  PUBLIC_LINK_LIBRARIES O2::Framework
+  TARGETVARNAME workflowPluginsTest
+  NO_BOOST_TEST
+  COMMAND_LINE_ARGS
+    --workflow-plugin O2TestFrameworkWorkflowPluginAPlugin:TestFrameworkWorkflowPluginA,O2TestFrameworkWorkflowPluginBPlugin:TestFrameworkWorkflowPluginB
+    --shared 3 --run --shm-segment-size 20000000 ${DPL_WORKFLOW_TESTS_EXTRA_OPTIONS}
+  ENVIRONMENT "DYLD_LIBRARY_PATH=${CMAKE_LIBRARY_OUTPUT_DIRECTORY}:$ENV{DYLD_LIBRARY_PATH};LD_LIBRARY_PATH=${CMAKE_LIBRARY_OUTPUT_DIRECTORY}:$ENV{LD_LIBRARY_PATH}"
+  )
+if(workflowPluginsTest)
+  foreach(p A B)
+    o2_add_executable(WorkflowPlugin${p} SOURCES test/test_WorkflowPlugin${p}.cxx
+                      COMPONENT_NAME Framework PUBLIC_LINK_LIBRARIES O2::Framework
+                      IS_TEST NO_INSTALL WORKFLOW_PLUGIN TARGETVARNAME workflowPlugin)
+    add_dependencies(${workflowPluginsTest} ${workflowPlugin})
+  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
 # command line option to include the output proxy
 o2_add_test(

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/Plugins.h

@@ -44,6 +44,9 @@ enum struct DplPluginKind : int {
   // using the arrow dataset API
   RootObjectReadingImplementation,
 
+  // A plugin which defines a whole workflow. This will be used to separate
+  // workflows in shared libraries and run them via a separate loader.
+  Workflow,
   // A plugin which was not initialised properly.
   Unknown
 };

Framework/Core/include/Framework/WorkflowDefinitionContext.h

@@ -0,0 +1,75 @@
+// Copyright 2019-2025 CERN and copyright holders of ALICE O2.
+// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders.
+// All rights not expressly granted are reserved.
+//
+// This software is distributed under the terms of the GNU General Public
+// License v3 (GPL Version 3), copied verbatim in the file "COPYING".
+//
+// In applying this license CERN does not waive the privileges and immunities
+// granted to it by virtue of its status as an Intergovernmental Organization
+// or submit itself to any jurisdiction.
+#ifndef O2_FRAMEWORK_WORKFLOWDEFINITIONCONTEXT_H_
+#define O2_FRAMEWORK_WORKFLOWDEFINITIONCONTEXT_H_
+
+#include "Framework/ConfigContext.h"
+#include "Framework/ConfigParamRegistry.h"
+#include "Framework/ConfigParamSpec.h"
+#include "Framework/CompletionPolicy.h"
+#include "Framework/DispatchPolicy.h"
+#include "Framework/ResourcePolicy.h"
+#include "Framework/CallbacksPolicy.h"
+#include "Framework/SendingPolicy.h"
+#include "Framework/WorkflowSpec.h"
+#include "Framework/ChannelConfigurationPolicy.h"
+#include "Framework/ServiceRegistry.h"
+#include <functional>
+#include <map>
+#include <memory>
+#include <vector>
+
+namespace o2::framework
+{
+
+struct WorkflowDefinitionContext {
+  std::vector<ConfigParamSpec> workflowOptions;
+  std::vector<CompletionPolicy> completionPolicies;
+  std::vector<DispatchPolicy> dispatchPolicies;
+  std::vector<ResourcePolicy> resourcePolicies;
+  std::vector<CallbacksPolicy> callbacksPolicies;
+  std::vector<SendingPolicy> sendingPolicies;
+  std::vector<ConfigParamSpec> extraOptions;
+  std::vector<ChannelConfigurationPolicy> channelPolicies;
+  std::unique_ptr<ConfigContext> configContext;
+
+  // For the moment, let's put them here. We should
+  // probably move them to a different place, since these are not really part
+  // of the workflow definition but will be there also at runtime.
+  std::unique_ptr<ServiceRegistry> configRegistry{nullptr};
+  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 {
+  /// Fills the workflow options and the policies the plugin customises.
+  std::function<void(o2::framework::WorkflowDefinitionContext& context)> customize;
+  /// Fills the data processors and the channel policies of the plugin,
+  /// using the workflow options parsed in @a configContext.
+  std::function<void(o2::framework::WorkflowDefinitionContext& context, o2::framework::ConfigContext& configContext)> defineWorkflow;
+};
+
+struct WorkflowPlugin {
+  virtual o2::framework::WorkflowDefinition* create() = 0;
+};
+
+/// Defines the workflow made of all the @a definitions, called @a names in errors.
+/// The workflow options of all of them are parsed once from @a argc / @a argv, then
+/// their data processors and policies are merged, in the order of @a definitions.
+/// An option may be declared by several definitions only with the same type and default,
+/// and a data processor name may not be used by more than one definition.
+WorkflowDefinitionContext defineWorkflow(std::vector<WorkflowDefinition> const& definitions, std::vector<std::string> const& names, int argc, char** argv);
+
+} // namespace o2::framework
+#endif // O2_FRAMEWORK_WORKFLOWDEFINITIONCONTEXT_H_

Framework/Core/include/Framework/runDataProcessing.h

@@ -27,6 +27,8 @@
 #include "Framework/CommonServices.h"
 #include "Framework/WorkflowCustomizationHelpers.h"
 #include "Framework/Logger.h" // IWYU pragma: export
+#include "Framework/WorkflowDefinitionContext.h"
+#include "Framework/Plugins.h"
 #include "Framework/CheckTypes.h"
 #include "Framework/StructToTuple.h"
 #include "ResourcePolicy.h"
@@ -124,18 +126,6 @@ void overrideCloning(o2::framework::ConfigContext& ctx, std::vector<o2::framewor
 /// Helper used to add labels to Data Processors
 void overrideLabels(o2::framework::ConfigContext& ctx, std::vector<o2::framework::DataProcessorSpec>& workflow);
 
-// This comes from the framework itself. This way we avoid code duplication.
-int doMain(int argc, char** argv, o2::framework::WorkflowSpec const& specs,
-           std::vector<o2::framework::ChannelConfigurationPolicy> const& channelPolicies,
-           std::vector<o2::framework::CompletionPolicy> const& completionPolicies,
-           std::vector<o2::framework::DispatchPolicy> const& dispatchPolicies,
-           std::vector<o2::framework::ResourcePolicy> const& resourcePolicies,
-           std::vector<o2::framework::CallbacksPolicy> const& callbacksPolicies,
-           std::vector<o2::framework::SendingPolicy> const& sendingPolicies,
-           std::vector<o2::framework::ConfigParamSpec> const& workflowOptions,
-           std::vector<o2::framework::ConfigParamSpec> const& detectedOptions,
-           o2::framework::ConfigContext& configContext);
-
 void doDefaultWorkflowTerminationHook();
 
 template <typename T>
@@ -167,55 +157,71 @@ void callWorkflowTermination(T&, char const* idstring)
 
 void overrideAll(o2::framework::ConfigContext& ctx, std::vector<o2::framework::DataProcessorSpec>& workflow);
 
-o2::framework::ConfigContext createConfigContext(std::unique_ptr<o2::framework::ConfigParamRegistry>& workflowOptionsRegistry,
-                                                 o2::framework::ServiceRegistry& configRegistry,
-                                                 std::vector<o2::framework::ConfigParamSpec>& workflowOptions,
-                                                 std::vector<o2::framework::ConfigParamSpec>& extraOptions, int argc, char** argv);
+std::unique_ptr<o2::framework::ConfigContext> createConfigContext(std::unique_ptr<o2::framework::ConfigParamRegistry>& workflowOptionsRegistry,
+                                                                  o2::framework::ServiceRegistry& configRegistry,
+                                                                  std::vector<o2::framework::ConfigParamSpec>& workflowOptions,
+                                                                  std::vector<o2::framework::ConfigParamSpec>& extraOptions, int argc, char** argv);
 
 std::unique_ptr<o2::framework::ServiceRegistry> createRegistry();
 
-int mainNoCatch(int argc, char** argv)
-{
-  using namespace o2::framework;
+char* getIdString(int argc, char** argv);
+
+#define DPL_STRINGIZE_NX(A) #A
+#define DPL_STRINGIZE(A) DPL_STRINGIZE_NX(A)
+
+// The workflow definition is a plugin. By default it is embedded in the
+// executable as DPL_WORKFLOW_PLUGIN_NAME, and main() runs it. A workflow built
+// as a plugin library gets DPL_WORKFLOW_PLUGIN_NAME at build time: it has no
+// main(), and the executable which links it calls dpl_workflow_main() instead.
+#ifndef DPL_WORKFLOW_PLUGIN_NAME
+#define DPL_WORKFLOW_PLUGIN_NAME InternalCustomWorkflow
+#define DPL_WORKFLOW_PLUGIN_EMBEDDED
+#endif
 
-  std::vector<o2::framework::ConfigParamSpec> workflowOptions;
-  UserCustomizationsHelper::userDefinedCustomization(workflowOptions);
-  auto requiredWorkflowOptions = WorkflowCustomizationHelpers::requiredWorkflowOptions();
-  workflowOptions.insert(std::end(workflowOptions), std::begin(requiredWorkflowOptions), std::end(requiredWorkflowOptions));
-
-  std::vector<CompletionPolicy> completionPolicies = injectCustomizations<CompletionPolicy>();
-  std::vector<DispatchPolicy> dispatchPolicies = injectCustomizations<DispatchPolicy>();
-  std::vector<ResourcePolicy> resourcePolicies = injectCustomizations<ResourcePolicy>();
-  std::vector<CallbacksPolicy> callbacksPolicies = injectCustomizations<CallbacksPolicy>();
-  std::vector<SendingPolicy> sendingPolicies = injectCustomizations<SendingPolicy>();
-
-  std::unique_ptr<ServiceRegistry> configRegistry = createRegistry();
-  std::vector<ConfigParamSpec> extraOptions;
-  std::unique_ptr<ConfigParamRegistry> workflowOptionsRegistry{nullptr};
-  auto configContext = createConfigContext(workflowOptionsRegistry, *configRegistry, workflowOptions, extraOptions, argc, argv);
-
-  o2::framework::WorkflowSpec specs = defineDataProcessing(configContext);
-  overrideAll(configContext, specs);
-  for (auto& spec : specs) {
-    UserCustomizationsHelper::userDefinedCustomization(spec.requiredServices);
+struct DPL_WORKFLOW_PLUGIN_NAME : o2::framework::WorkflowPlugin {
+  o2::framework::WorkflowDefinition* create() override
+  {
+    return new o2::framework::WorkflowDefinition{
+      .customize = [](o2::framework::WorkflowDefinitionContext& workflowContext) {
+        using namespace o2::framework;
+        UserCustomizationsHelper::userDefinedCustomization(workflowContext.workflowOptions);
+        UserCustomizationsHelper::userDefinedCustomization(workflowContext.completionPolicies);
+        UserCustomizationsHelper::userDefinedCustomization(workflowContext.dispatchPolicies);
+        UserCustomizationsHelper::userDefinedCustomization(workflowContext.resourcePolicies);
+        UserCustomizationsHelper::userDefinedCustomization(workflowContext.callbacksPolicies);
+        UserCustomizationsHelper::userDefinedCustomization(workflowContext.sendingPolicies); },
+      .defineWorkflow = [](o2::framework::WorkflowDefinitionContext& workflowContext, o2::framework::ConfigContext& configContext) {
+        using namespace o2::framework;
+        workflowContext.specs = defineDataProcessing(configContext);
+        overrideAll(configContext, workflowContext.specs);
+        for (auto& spec : workflowContext.specs) {
+          UserCustomizationsHelper::userDefinedCustomization(spec.requiredServices);
+        }
+        UserCustomizationsHelper::userDefinedCustomization(workflowContext.channelPolicies); }};
   }
-  std::vector<ChannelConfigurationPolicy> channelPolicies;
-  UserCustomizationsHelper::userDefinedCustomization(channelPolicies);
-  auto defaultChannelPolicies = ChannelConfigurationPolicy::createDefaultPolicies(configContext);
-  channelPolicies.insert(std::end(channelPolicies), std::begin(defaultChannelPolicies), std::end(defaultChannelPolicies));
-  return doMain(argc, argv, specs,
-                channelPolicies, completionPolicies, dispatchPolicies,
-                resourcePolicies, callbacksPolicies, sendingPolicies, workflowOptions, extraOptions, configContext);
+};
+
+// This is like the plugin macros, we simply do it explicitly to avoid macro inside macro expansion
+extern "C" {
+DPLPluginHandle* dpl_plugin_callback(DPLPluginHandle* previous)
+{
+  return new DPLPluginHandle{new DPL_WORKFLOW_PLUGIN_NAME{}, strdup(DPL_STRINGIZE(DPL_WORKFLOW_PLUGIN_NAME)), o2::framework::DplPluginKind::Workflow, previous};
+}
 }
 
-int callMain(int argc, char** argv, int (*)(int, char**));
-char* getIdString(int argc, char** argv);
+// Loads the workflow plugin @a pluginSpec, i.e. "<library>:<plugin>", and runs it.
+// An empty library means the plugin embedded in the executable, found via @a embeddedPlugins.
+int callMain(int argc, char** argv, char const* pluginSpec, DPLPluginHandle* (*embeddedPlugins)(DPLPluginHandle*));
 
+#ifdef DPL_WORKFLOW_PLUGIN_EMBEDDED
 int main(int argc, char** argv)
+#else
+extern "C" int dpl_workflow_main(int argc, char** argv)
+#endif
 {
   using namespace o2::framework;
 
-  int result = callMain(argc, argv, mainNoCatch);
+  int result = callMain(argc, argv, ":" DPL_STRINGIZE(DPL_WORKFLOW_PLUGIN_NAME), dpl_plugin_callback);
 
   char* idstring = getIdString(argc, argv);
   o2::framework::OnWorkflowTerminationHook onWorkflowTerminationHook;

Framework/Core/src/DeviceSpecHelpers.cxx

@@ -1490,6 +1490,8 @@ void DeviceSpecHelpers::prepareArguments(bool defaultQuiet, bool defaultStopped,
     auto forwardedOptions = getForwardedDeviceOptions();
     /// Add to foDesc the options which are not already there
     foDesc.add(forwardedOptions);
+    // Children need the same workflow plugins to define the same workflow.
+    foDesc.add_options()("workflow-plugin", bpo::value<std::string>());
 
     // has option --session been specified on the command line?
     bool haveSessionArg = false;
@@ -1747,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) {