Skip to content

Commit 0857c8a

Browse files
ktfktf
committed
DPL: first step to make workflow definition plugins
This is the first step towards having the workflow definition inside plugins rather than in executables. This will allow accumulating plugins which are needed to instantiate a topology and do the option parsing / topology building only once, simplifying the current case. The end goal is to allow the driver to preload certain common services (e.g. ROOT) and share it among the different tasks (which at the moment it's not allowed because different tasks are in different executables). Moreover this will allow us to coalesce strictly coupled dataprocessors and reduce the number of running processes. For now the plugins are embedded in the executables and behave exactly like before: runDataProcessing.h wraps what used to be mainNoCatch in a WorkflowPlugin of kind DplPluginKind::Workflow, which returns the whole WorkflowDefinitionContext (options, policies, config context and specs), and callMain picks it by plugin spec ("<library>:<plugin>", with an empty library for the embedded one) before handing it to doMain. The embedded plugin is looked up through the callback passed by main, not with dlsym on the executable, which would need exported symbols.
1 parent d670d3a commit 0857c8a

4 files changed

Lines changed: 165 additions & 65 deletions

File tree

‎Framework/Core/include/Framework/Plugins.h‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -44,6 +44,9 @@ enum struct DplPluginKind : int {
4444
// using the arrow dataset API
4545
RootObjectReadingImplementation,
4646

47+
// A plugin which defines a whole workflow. This will be used to separate
48+
// workflows in shared libraries and run them via a separate loader.
49+
Workflow,
4750
// A plugin which was not initialised properly.
4851
Unknown
4952
};

‎Framework/Core/include/Framework/WorkflowDefinitionContext.h‎

Lines changed: 61 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,61 @@
1+
// Copyright 2019-2025 CERN and copyright holders of ALICE O2.
2+
// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders.
3+
// All rights not expressly granted are reserved.
4+
//
5+
// This software is distributed under the terms of the GNU General Public
6+
// License v3 (GPL Version 3), copied verbatim in the file "COPYING".
7+
//
8+
// In applying this license CERN does not waive the privileges and immunities
9+
// granted to it by virtue of its status as an Intergovernmental Organization
10+
// or submit itself to any jurisdiction.
11+
#ifndef O2_FRAMEWORK_WORKFLOWDEFINITIONCONTEXT_H_
12+
#define O2_FRAMEWORK_WORKFLOWDEFINITIONCONTEXT_H_
13+
14+
#include "Framework/ConfigContext.h"
15+
#include "Framework/ConfigParamRegistry.h"
16+
#include "Framework/ConfigParamSpec.h"
17+
#include "Framework/CompletionPolicy.h"
18+
#include "Framework/DispatchPolicy.h"
19+
#include "Framework/ResourcePolicy.h"
20+
#include "Framework/CallbacksPolicy.h"
21+
#include "Framework/SendingPolicy.h"
22+
#include "Framework/WorkflowSpec.h"
23+
#include "Framework/ChannelConfigurationPolicy.h"
24+
#include "Framework/ServiceRegistry.h"
25+
#include <functional>
26+
#include <memory>
27+
#include <vector>
28+
29+
namespace o2::framework
30+
{
31+
32+
struct WorkflowDefinitionContext {
33+
std::vector<ConfigParamSpec> workflowOptions;
34+
std::vector<CompletionPolicy> completionPolicies;
35+
std::vector<DispatchPolicy> dispatchPolicies;
36+
std::vector<ResourcePolicy> resourcePolicies;
37+
std::vector<CallbacksPolicy> callbacksPolicies;
38+
std::vector<SendingPolicy> sendingPolicies;
39+
std::vector<ConfigParamSpec> extraOptions;
40+
std::vector<ChannelConfigurationPolicy> channelPolicies;
41+
std::unique_ptr<ConfigContext> configContext;
42+
43+
// For the moment, let's put them here. We should
44+
// probably move them to a different place, since these are not really part
45+
// of the workflow definition but will be there also at runtime.
46+
std::unique_ptr<ServiceRegistry> configRegistry{nullptr};
47+
std::unique_ptr<ConfigParamRegistry> workflowOptionsRegistry{nullptr};
48+
49+
o2::framework::WorkflowSpec specs;
50+
};
51+
52+
struct WorkflowDefinition {
53+
std::function<o2::framework::WorkflowDefinitionContext(int argc, char** argv)> defineWorkflow;
54+
};
55+
56+
struct WorkflowPlugin {
57+
virtual o2::framework::WorkflowDefinition* create() = 0;
58+
};
59+
60+
} // namespace o2::framework
61+
#endif // O2_FRAMEWORK_WORKFLOWDEFINITIONCONTEXT_H_

‎Framework/Core/include/Framework/runDataProcessing.h‎

Lines changed: 65 additions & 49 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,8 @@
2727
#include "Framework/CommonServices.h"
2828
#include "Framework/WorkflowCustomizationHelpers.h"
2929
#include "Framework/Logger.h" // IWYU pragma: export
30+
#include "Framework/WorkflowDefinitionContext.h"
31+
#include "Framework/Plugins.h"
3032
#include "Framework/CheckTypes.h"
3133
#include "Framework/StructToTuple.h"
3234
#include "ResourcePolicy.h"
@@ -124,18 +126,6 @@ void overrideCloning(o2::framework::ConfigContext& ctx, std::vector<o2::framewor
124126
/// Helper used to add labels to Data Processors
125127
void overrideLabels(o2::framework::ConfigContext& ctx, std::vector<o2::framework::DataProcessorSpec>& workflow);
126128

127-
// This comes from the framework itself. This way we avoid code duplication.
128-
int doMain(int argc, char** argv, o2::framework::WorkflowSpec const& specs,
129-
std::vector<o2::framework::ChannelConfigurationPolicy> const& channelPolicies,
130-
std::vector<o2::framework::CompletionPolicy> const& completionPolicies,
131-
std::vector<o2::framework::DispatchPolicy> const& dispatchPolicies,
132-
std::vector<o2::framework::ResourcePolicy> const& resourcePolicies,
133-
std::vector<o2::framework::CallbacksPolicy> const& callbacksPolicies,
134-
std::vector<o2::framework::SendingPolicy> const& sendingPolicies,
135-
std::vector<o2::framework::ConfigParamSpec> const& workflowOptions,
136-
std::vector<o2::framework::ConfigParamSpec> const& detectedOptions,
137-
o2::framework::ConfigContext& configContext);
138-
139129
void doDefaultWorkflowTerminationHook();
140130

141131
template <typename T>
@@ -167,55 +157,81 @@ void callWorkflowTermination(T&, char const* idstring)
167157

168158
void overrideAll(o2::framework::ConfigContext& ctx, std::vector<o2::framework::DataProcessorSpec>& workflow);
169159

170-
o2::framework::ConfigContext createConfigContext(std::unique_ptr<o2::framework::ConfigParamRegistry>& workflowOptionsRegistry,
171-
o2::framework::ServiceRegistry& configRegistry,
172-
std::vector<o2::framework::ConfigParamSpec>& workflowOptions,
173-
std::vector<o2::framework::ConfigParamSpec>& extraOptions, int argc, char** argv);
160+
std::unique_ptr<o2::framework::ConfigContext> createConfigContext(std::unique_ptr<o2::framework::ConfigParamRegistry>& workflowOptionsRegistry,
161+
o2::framework::ServiceRegistry& configRegistry,
162+
std::vector<o2::framework::ConfigParamSpec>& workflowOptions,
163+
std::vector<o2::framework::ConfigParamSpec>& extraOptions, int argc, char** argv);
174164

175165
std::unique_ptr<o2::framework::ServiceRegistry> createRegistry();
176166

177-
int mainNoCatch(int argc, char** argv)
178-
{
179-
using namespace o2::framework;
167+
char* getIdString(int argc, char** argv);
168+
169+
#define DPL_STRINGIZE_NX(A) #A
170+
#define DPL_STRINGIZE(A) DPL_STRINGIZE_NX(A)
171+
172+
// The workflow definition is a plugin. By default it is embedded in the
173+
// executable as DPL_WORKFLOW_PLUGIN_NAME, with an empty library name.
174+
// To build it in a DSO instead, both DPL_WORKFLOW_PLUGIN_NAME and
175+
// DPL_WORKFLOW_PLUGIN_LIBRARY must be specified at build time.
176+
#ifndef DPL_WORKFLOW_PLUGIN_NAME
177+
#define DPL_WORKFLOW_PLUGIN_NAME InternalCustomWorkflow
178+
#ifdef DPL_WORKFLOW_PLUGIN_LIBRARY
179+
#error Missing DPL_WORKFLOW_PLUGIN_NAME
180+
#endif
181+
#define DPL_WORKFLOW_PLUGIN_LIBRARY
182+
#endif
180183

181-
std::vector<o2::framework::ConfigParamSpec> workflowOptions;
182-
UserCustomizationsHelper::userDefinedCustomization(workflowOptions);
183-
auto requiredWorkflowOptions = WorkflowCustomizationHelpers::requiredWorkflowOptions();
184-
workflowOptions.insert(std::end(workflowOptions), std::begin(requiredWorkflowOptions), std::end(requiredWorkflowOptions));
185-
186-
std::vector<CompletionPolicy> completionPolicies = injectCustomizations<CompletionPolicy>();
187-
std::vector<DispatchPolicy> dispatchPolicies = injectCustomizations<DispatchPolicy>();
188-
std::vector<ResourcePolicy> resourcePolicies = injectCustomizations<ResourcePolicy>();
189-
std::vector<CallbacksPolicy> callbacksPolicies = injectCustomizations<CallbacksPolicy>();
190-
std::vector<SendingPolicy> sendingPolicies = injectCustomizations<SendingPolicy>();
191-
192-
std::unique_ptr<ServiceRegistry> configRegistry = createRegistry();
193-
std::vector<ConfigParamSpec> extraOptions;
194-
std::unique_ptr<ConfigParamRegistry> workflowOptionsRegistry{nullptr};
195-
auto configContext = createConfigContext(workflowOptionsRegistry, *configRegistry, workflowOptions, extraOptions, argc, argv);
196-
197-
o2::framework::WorkflowSpec specs = defineDataProcessing(configContext);
198-
overrideAll(configContext, specs);
199-
for (auto& spec : specs) {
200-
UserCustomizationsHelper::userDefinedCustomization(spec.requiredServices);
184+
struct DPL_WORKFLOW_PLUGIN_NAME : o2::framework::WorkflowPlugin {
185+
o2::framework::WorkflowDefinition* create() override
186+
{
187+
return new o2::framework::WorkflowDefinition{
188+
.defineWorkflow = [](int argc, char** argv) -> o2::framework::WorkflowDefinitionContext {
189+
using namespace o2::framework;
190+
WorkflowDefinitionContext workflowContext;
191+
192+
UserCustomizationsHelper::userDefinedCustomization(workflowContext.workflowOptions);
193+
auto requiredWorkflowOptions = WorkflowCustomizationHelpers::requiredWorkflowOptions();
194+
workflowContext.workflowOptions.insert(std::end(workflowContext.workflowOptions), std::begin(requiredWorkflowOptions), std::end(requiredWorkflowOptions));
195+
196+
workflowContext.completionPolicies = injectCustomizations<CompletionPolicy>();
197+
workflowContext.dispatchPolicies = injectCustomizations<DispatchPolicy>();
198+
workflowContext.resourcePolicies = injectCustomizations<ResourcePolicy>();
199+
workflowContext.callbacksPolicies = injectCustomizations<CallbacksPolicy>();
200+
workflowContext.sendingPolicies = injectCustomizations<SendingPolicy>();
201+
202+
workflowContext.configRegistry = createRegistry();
203+
workflowContext.configContext = createConfigContext(workflowContext.workflowOptionsRegistry, *workflowContext.configRegistry, workflowContext.workflowOptions, workflowContext.extraOptions, argc, argv);
204+
205+
workflowContext.specs = defineDataProcessing(*workflowContext.configContext);
206+
overrideAll(*workflowContext.configContext, workflowContext.specs);
207+
for (auto& spec : workflowContext.specs) {
208+
UserCustomizationsHelper::userDefinedCustomization(spec.requiredServices);
209+
}
210+
UserCustomizationsHelper::userDefinedCustomization(workflowContext.channelPolicies);
211+
auto defaultChannelPolicies = ChannelConfigurationPolicy::createDefaultPolicies(*workflowContext.configContext);
212+
workflowContext.channelPolicies.insert(std::end(workflowContext.channelPolicies), std::begin(defaultChannelPolicies), std::end(defaultChannelPolicies));
213+
return workflowContext;
214+
}};
201215
}
202-
std::vector<ChannelConfigurationPolicy> channelPolicies;
203-
UserCustomizationsHelper::userDefinedCustomization(channelPolicies);
204-
auto defaultChannelPolicies = ChannelConfigurationPolicy::createDefaultPolicies(configContext);
205-
channelPolicies.insert(std::end(channelPolicies), std::begin(defaultChannelPolicies), std::end(defaultChannelPolicies));
206-
return doMain(argc, argv, specs,
207-
channelPolicies, completionPolicies, dispatchPolicies,
208-
resourcePolicies, callbacksPolicies, sendingPolicies, workflowOptions, extraOptions, configContext);
216+
};
217+
218+
// This is like the plugin macros, we simply do it explicitly to avoid macro inside macro expansion
219+
extern "C" {
220+
DPLPluginHandle* dpl_plugin_callback(DPLPluginHandle* previous)
221+
{
222+
return new DPLPluginHandle{new DPL_WORKFLOW_PLUGIN_NAME{}, strdup(DPL_STRINGIZE(DPL_WORKFLOW_PLUGIN_NAME)), o2::framework::DplPluginKind::Workflow, previous};
223+
}
209224
}
210225

211-
int callMain(int argc, char** argv, int (*)(int, char**));
212-
char* getIdString(int argc, char** argv);
226+
// Loads the workflow plugin @a pluginSpec, i.e. "<library>:<plugin>", and runs it.
227+
// An empty library means the plugin embedded in the executable, found via @a embeddedPlugins.
228+
int callMain(int argc, char** argv, char const* pluginSpec, DPLPluginHandle* (*embeddedPlugins)(DPLPluginHandle*));
213229

214230
int main(int argc, char** argv)
215231
{
216232
using namespace o2::framework;
217233

218-
int result = callMain(argc, argv, mainNoCatch);
234+
int result = callMain(argc, argv, DPL_STRINGIZE(DPL_WORKFLOW_PLUGIN_LIBRARY) ":" DPL_STRINGIZE(DPL_WORKFLOW_PLUGIN_NAME), dpl_plugin_callback);
219235

220236
char* idstring = getIdString(argc, argv);
221237
o2::framework::OnWorkflowTerminationHook onWorkflowTerminationHook;

‎Framework/Core/src/runDataProcessing.cxx‎

Lines changed: 36 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -65,6 +65,7 @@
6565
#include "Framework/DeviceContext.h"
6666
#include "Framework/ServiceMetricsInfo.h"
6767
#include "Framework/DataTakingContext.h"
68+
#include "Framework/WorkflowDefinitionContext.h"
6869
#include "Framework/CommonServices.h"
6970
#include "Framework/DefaultsHelpers.h"
7071
#include "ProcessingPoliciesHelpers.h"
@@ -196,8 +197,26 @@ char* getIdString(int argc, char** argv)
196197
return nullptr;
197198
}
198199

199-
int callMain(int argc, char** argv, int (*mainNoCatch)(int, char**))
200+
int doMain(int argc, char** argv, o2::framework::WorkflowDefinitionContext& workflowContext);
201+
202+
int callMain(int argc, char** argv, char const* pluginSpec, DPLPluginHandle* (*embeddedPlugins)(DPLPluginHandle*))
200203
{
204+
auto mainNoCatch = [pluginSpec, embeddedPlugins](int argc, char** argv) {
205+
auto plugins = PluginManager::parsePluginSpecString(pluginSpec);
206+
std::vector<WorkflowDefinition> workflows;
207+
if (plugins.size() == 1 && plugins[0].library.empty()) {
208+
if (auto* plugin = PluginManager::getByName<WorkflowPlugin>(embeddedPlugins(nullptr), plugins[0].name.c_str())) {
209+
workflows.push_back(*plugin->create());
210+
}
211+
} else {
212+
PluginManager::loadFromPlugin<WorkflowDefinition, WorkflowPlugin>(plugins, workflows);
213+
}
214+
if (workflows.size() != 1) {
215+
throw runtime_error_f("Expected exactly one workflow plugin in %s, found %zu.", pluginSpec, workflows.size());
216+
}
217+
auto workflowContext = workflows.back().defineWorkflow(argc, argv);
218+
return doMain(argc, argv, workflowContext);
219+
};
201220
static bool noCatch = getenv("O2_NO_CATCHALL_EXCEPTIONS") && strcmp(getenv("O2_NO_CATCHALL_EXCEPTIONS"), "0");
202221
int result = 1;
203222
if (noCatch) {
@@ -2882,10 +2901,10 @@ void overrideAll(o2::framework::ConfigContext& ctx, std::vector<o2::framework::D
28822901
overrideLabels(ctx, workflow);
28832902
}
28842903

2885-
o2::framework::ConfigContext createConfigContext(std::unique_ptr<ConfigParamRegistry>& workflowOptionsRegistry,
2886-
o2::framework::ServiceRegistry& configRegistry,
2887-
std::vector<o2::framework::ConfigParamSpec>& workflowOptions,
2888-
std::vector<o2::framework::ConfigParamSpec>& extraOptions, int argc, char** argv)
2904+
std::unique_ptr<o2::framework::ConfigContext> createConfigContext(std::unique_ptr<ConfigParamRegistry>& workflowOptionsRegistry,
2905+
o2::framework::ServiceRegistry& configRegistry,
2906+
std::vector<o2::framework::ConfigParamSpec>& workflowOptions,
2907+
std::vector<o2::framework::ConfigParamSpec>& extraOptions, int argc, char** argv)
28892908
{
28902909
std::vector<std::unique_ptr<o2::framework::ParamRetriever>> retrievers;
28912910
std::unique_ptr<o2::framework::ParamRetriever> retriever{new o2::framework::BoostOptionsRetriever(true, argc, argv)};
@@ -2899,7 +2918,7 @@ o2::framework::ConfigContext createConfigContext(std::unique_ptr<ConfigParamRegi
28992918
workflowOptions.push_back(extra);
29002919
}
29012920

2902-
return o2::framework::ConfigContext(*workflowOptionsRegistry, o2::framework::ServiceRegistryRef{configRegistry}, argc, argv);
2921+
return std::make_unique<o2::framework::ConfigContext>(*workflowOptionsRegistry, o2::framework::ServiceRegistryRef{configRegistry}, argc, argv);
29032922
}
29042923

29052924
std::unique_ptr<o2::framework::ServiceRegistry> createRegistry()
@@ -2916,17 +2935,18 @@ std::unique_ptr<o2::framework::ServiceRegistry> createRegistry()
29162935
// killing them all on ctrl-c).
29172936
// - Child, pick the data-processor ID and start a O2DataProcessorDevice for
29182937
// each DataProcessorSpec
2919-
int doMain(int argc, char** argv, o2::framework::WorkflowSpec const& workflow,
2920-
std::vector<ChannelConfigurationPolicy> const& channelPolicies,
2921-
std::vector<CompletionPolicy> const& completionPolicies,
2922-
std::vector<DispatchPolicy> const& dispatchPolicies,
2923-
std::vector<ResourcePolicy> const& resourcePolicies,
2924-
std::vector<CallbacksPolicy> const& callbacksPolicies,
2925-
std::vector<SendingPolicy> const& sendingPolicies,
2926-
std::vector<ConfigParamSpec> const& currentWorkflowOptions,
2927-
std::vector<ConfigParamSpec> const& detectedParams,
2928-
o2::framework::ConfigContext& configContext)
2938+
int doMain(int argc, char** argv, o2::framework::WorkflowDefinitionContext& workflowContext)
29292939
{
2940+
auto const& workflow = workflowContext.specs;
2941+
auto const& channelPolicies = workflowContext.channelPolicies;
2942+
auto const& completionPolicies = workflowContext.completionPolicies;
2943+
auto const& dispatchPolicies = workflowContext.dispatchPolicies;
2944+
auto const& resourcePolicies = workflowContext.resourcePolicies;
2945+
auto const& callbacksPolicies = workflowContext.callbacksPolicies;
2946+
auto const& sendingPolicies = workflowContext.sendingPolicies;
2947+
auto const& currentWorkflowOptions = workflowContext.workflowOptions;
2948+
auto const& detectedParams = workflowContext.extraOptions;
2949+
auto& configContext = *workflowContext.configContext;
29302950
// Peek very early in the driver options and look for
29312951
// signposts, so the we can enable it without going through the whole dance
29322952
if (getenv("DPL_DRIVER_SIGNPOSTS")) {

0 commit comments

Comments
 (0)