Skip to content

Commit 91a803b

Browse files
committed
DPL: allow a workflow made of several plugins
--workflow-plugin <library>:<plugin>[,<library>:<plugin>...] replaces the workflow plugin embedded in the executable with the given ones, which are merged in a single workflow. An empty library means the executable itself, so :InternalCustomWorkflow adds the executable's own workflow. A plugin now only provides two steps, customize and defineWorkflow, and defineWorkflow() in the framework does the rest once for all of them: - the workflow options of all plugins are merged and parsed once, into a single ConfigContext and service registry, shared by all plugins, so that e.g. what analysis tasks request lands in the same registry. An option declared by more than one plugin is shared, provided they agree on its type and default; otherwise defining the workflow fails. The options required by DPL are added once, after those of the plugins. - the data processors are concatenated in plugin order. A data processor name defined by more than one plugin is an error. Within a single plugin nothing is checked, as before. - the policies customised by each plugin are concatenated in plugin order, followed by the default policies once, so that the catch-all defaults of one plugin do not hide the policies of the next one. For a single plugin, which is what every executable does by default, the sequence of calls is the same as before, so behaviour, help and dumps are unchanged. --workflow-plugin is a hidden driver option, it is forwarded to the children, which need it to define the same workflow, and accepted by the devices. Tested by test_WorkflowDefinition (merging, conflicting options, duplicate data processors) and test_Framework_test_WorkflowPlugins, which runs a producer and a consumer from two DSO plugins and checks that both see the same value of their shared option.
1 parent 0857c8a commit 91a803b

9 files changed

Lines changed: 334 additions & 32 deletions

‎Framework/Core/CMakeLists.txt‎

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -268,6 +268,7 @@ add_executable(o2-test-framework-core
268268
test/test_TopologyPolicies.cxx
269269
test/test_Variants.cxx
270270
test/test_WorkflowHelpers.cxx
271+
test/test_WorkflowDefinition.cxx
271272
test/test_WorkflowSerialization.cxx
272273
test/test_DataOutputDirector.cxx
273274
test/unittest_SimpleOptionsRetriever.cxx
@@ -447,6 +448,32 @@ o2_add_test(
447448
"--global-config consumer-config --local-option hello-aliceo2 --a-boolean3 --an-int2 20 --a-double2 22. --an-int64-2 50000000000000"
448449
)
449450

451+
# runs the workflows of two plugins built in DSOs, loaded with --workflow-plugin
452+
o2_add_test(
453+
WorkflowPlugins NAME test_Framework_test_WorkflowPlugins
454+
SOURCES test/test_WorkflowPlugins.cxx
455+
COMPONENT_NAME Framework
456+
LABELS framework workflow
457+
TIMEOUT 30
458+
PUBLIC_LINK_LIBRARIES O2::Framework
459+
TARGETVARNAME workflowPluginsTest
460+
NO_BOOST_TEST
461+
COMMAND_LINE_ARGS
462+
--workflow-plugin O2FrameworkTestWorkflowPluginA:TestWorkflowPluginA,O2FrameworkTestWorkflowPluginB:TestWorkflowPluginB
463+
--shared 3 --run --shm-segment-size 20000000 ${DPL_WORKFLOW_TESTS_EXTRA_OPTIONS}
464+
ENVIRONMENT "DYLD_LIBRARY_PATH=${CMAKE_LIBRARY_OUTPUT_DIRECTORY}:$ENV{DYLD_LIBRARY_PATH};LD_LIBRARY_PATH=${CMAKE_LIBRARY_OUTPUT_DIRECTORY}:$ENV{LD_LIBRARY_PATH}"
465+
)
466+
if(workflowPluginsTest)
467+
foreach(p A B)
468+
add_library(O2FrameworkTestWorkflowPlugin${p} SHARED test/test_WorkflowPlugin${p}.cxx)
469+
target_link_libraries(O2FrameworkTestWorkflowPlugin${p} PRIVATE O2::Framework)
470+
target_compile_definitions(O2FrameworkTestWorkflowPlugin${p} PRIVATE
471+
DPL_WORKFLOW_PLUGIN_NAME=TestWorkflowPlugin${p}
472+
DPL_WORKFLOW_PLUGIN_LIBRARY=O2FrameworkTestWorkflowPlugin${p})
473+
add_dependencies(${workflowPluginsTest} O2FrameworkTestWorkflowPlugin${p})
474+
endforeach()
475+
endif()
476+
450477
# the test is compiled from the ExternalFairMQDeviceWorkflow test and run with
451478
# command line option to include the output proxy
452479
o2_add_test(

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

Lines changed: 12 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -50,12 +50,23 @@ struct WorkflowDefinitionContext {
5050
};
5151

5252
struct WorkflowDefinition {
53-
std::function<o2::framework::WorkflowDefinitionContext(int argc, char** argv)> defineWorkflow;
53+
/// Fills the workflow options and the policies the plugin customises.
54+
std::function<void(o2::framework::WorkflowDefinitionContext& context)> customize;
55+
/// Fills the data processors and the channel policies of the plugin,
56+
/// using the workflow options parsed in @a configContext.
57+
std::function<void(o2::framework::WorkflowDefinitionContext& context, o2::framework::ConfigContext& configContext)> defineWorkflow;
5458
};
5559

5660
struct WorkflowPlugin {
5761
virtual o2::framework::WorkflowDefinition* create() = 0;
5862
};
5963

64+
/// Defines the workflow made of all the @a definitions, called @a names in errors.
65+
/// The workflow options of all of them are parsed once from @a argc / @a argv, then
66+
/// their data processors and policies are merged, in the order of @a definitions.
67+
/// An option may be declared by several definitions only with the same type and default,
68+
/// and a data processor name may not be used by more than one definition.
69+
WorkflowDefinitionContext defineWorkflow(std::vector<WorkflowDefinition> const& definitions, std::vector<std::string> const& names, int argc, char** argv);
70+
6071
} // namespace o2::framework
6172
#endif // O2_FRAMEWORK_WORKFLOWDEFINITIONCONTEXT_H_

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

Lines changed: 11 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -185,33 +185,22 @@ struct DPL_WORKFLOW_PLUGIN_NAME : o2::framework::WorkflowPlugin {
185185
o2::framework::WorkflowDefinition* create() override
186186
{
187187
return new o2::framework::WorkflowDefinition{
188-
.defineWorkflow = [](int argc, char** argv) -> o2::framework::WorkflowDefinitionContext {
188+
.customize = [](o2::framework::WorkflowDefinitionContext& workflowContext) {
189189
using namespace o2::framework;
190-
WorkflowDefinitionContext workflowContext;
191-
192190
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);
191+
UserCustomizationsHelper::userDefinedCustomization(workflowContext.completionPolicies);
192+
UserCustomizationsHelper::userDefinedCustomization(workflowContext.dispatchPolicies);
193+
UserCustomizationsHelper::userDefinedCustomization(workflowContext.resourcePolicies);
194+
UserCustomizationsHelper::userDefinedCustomization(workflowContext.callbacksPolicies);
195+
UserCustomizationsHelper::userDefinedCustomization(workflowContext.sendingPolicies); },
196+
.defineWorkflow = [](o2::framework::WorkflowDefinitionContext& workflowContext, o2::framework::ConfigContext& configContext) {
197+
using namespace o2::framework;
198+
workflowContext.specs = defineDataProcessing(configContext);
199+
overrideAll(configContext, workflowContext.specs);
207200
for (auto& spec : workflowContext.specs) {
208201
UserCustomizationsHelper::userDefinedCustomization(spec.requiredServices);
209202
}
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-
}};
203+
UserCustomizationsHelper::userDefinedCustomization(workflowContext.channelPolicies); }};
215204
}
216205
};
217206

‎Framework/Core/src/DeviceSpecHelpers.cxx‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1490,6 +1490,8 @@ void DeviceSpecHelpers::prepareArguments(bool defaultQuiet, bool defaultStopped,
14901490
auto forwardedOptions = getForwardedDeviceOptions();
14911491
/// Add to foDesc the options which are not already there
14921492
foDesc.add(forwardedOptions);
1493+
// Children need the same workflow plugins to define the same workflow.
1494+
foDesc.add_options()("workflow-plugin", bpo::value<std::string>());
14931495

14941496
// has option --session been specified on the command line?
14951497
bool haveSessionArg = false;

‎Framework/Core/src/runDataProcessing.cxx‎

Lines changed: 81 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -66,6 +66,7 @@
6666
#include "Framework/ServiceMetricsInfo.h"
6767
#include "Framework/DataTakingContext.h"
6868
#include "Framework/WorkflowDefinitionContext.h"
69+
#include "Framework/WorkflowCustomizationHelpers.h"
6970
#include "Framework/CommonServices.h"
7071
#include "Framework/DefaultsHelpers.h"
7172
#include "ProcessingPoliciesHelpers.h"
@@ -202,19 +203,35 @@ int doMain(int argc, char** argv, o2::framework::WorkflowDefinitionContext& work
202203
int callMain(int argc, char** argv, char const* pluginSpec, DPLPluginHandle* (*embeddedPlugins)(DPLPluginHandle*))
203204
{
204205
auto mainNoCatch = [pluginSpec, embeddedPlugins](int argc, char** argv) {
205-
auto plugins = PluginManager::parsePluginSpecString(pluginSpec);
206+
// --workflow-plugin replaces the plugin embedded in the executable.
207+
char const* spec = pluginSpec;
208+
for (int ai = 1; ai < argc; ++ai) {
209+
if (strcmp(argv[ai], "--workflow-plugin") == 0 && ai + 1 < argc) {
210+
spec = argv[ai + 1];
211+
} else if (strncmp(argv[ai], "--workflow-plugin=", 18) == 0) {
212+
spec = argv[ai] + 18;
213+
}
214+
}
215+
auto plugins = PluginManager::parsePluginSpecString(spec);
206216
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());
217+
std::vector<std::string> names;
218+
for (auto& plugin : plugins) {
219+
if (plugin.library.empty()) {
220+
if (auto* embedded = PluginManager::getByName<WorkflowPlugin>(embeddedPlugins(nullptr), plugin.name.c_str())) {
221+
workflows.push_back(*embedded->create());
222+
}
223+
} else {
224+
PluginManager::loadFromPlugin<WorkflowDefinition, WorkflowPlugin>({plugin}, workflows);
225+
}
226+
names.push_back(plugin.library + ":" + plugin.name);
227+
if (workflows.size() != names.size()) {
228+
throw runtime_error_f("Could not load the workflow plugin %s.", names.back().c_str());
210229
}
211-
} else {
212-
PluginManager::loadFromPlugin<WorkflowDefinition, WorkflowPlugin>(plugins, workflows);
213230
}
214-
if (workflows.size() != 1) {
215-
throw runtime_error_f("Expected exactly one workflow plugin in %s, found %zu.", pluginSpec, workflows.size());
231+
if (workflows.empty()) {
232+
throw runtime_error_f("No workflow plugin in %s.", spec);
216233
}
217-
auto workflowContext = workflows.back().defineWorkflow(argc, argv);
234+
auto workflowContext = defineWorkflow(workflows, names, argc, argv);
218235
return doMain(argc, argv, workflowContext);
219236
};
220237
static bool noCatch = getenv("O2_NO_CATCHALL_EXCEPTIONS") && strcmp(getenv("O2_NO_CATCHALL_EXCEPTIONS"), "0");
@@ -1085,6 +1102,7 @@ int doChild(int argc, char** argv, ServiceRegistry& serviceRegistry,
10851102
("data-processing-timeout", bpo::value<std::string>()->default_value(defaultDataProcessingTimeout), "how many second to wait before stopping data processing and allowing data calibration") //
10861103
("timeframes-rate-limit", bpo::value<std::string>()->default_value("0"), "how many timeframe can be in flight at the same moment (0 disables)") //
10871104
("configuration,cfg", bpo::value<std::string>()->default_value("command-line"), "configuration backend") //
1105+
("workflow-plugin", bpo::value<std::string>(), "workflow plugins to load") //
10881106
("infologger-mode", bpo::value<std::string>()->default_value(defaultInfologgerMode), "O2_INFOLOGGER_MODE override") //
10891107
("log-timestamp-us", bpo::value<bool>()->zero_tokens()->default_value(false), "enable microsecond timestamps in log messages");
10901108
r.fConfig.AddToCmdLineOptions(optsDesc, true);
@@ -2926,6 +2944,59 @@ std::unique_ptr<o2::framework::ServiceRegistry> createRegistry()
29262944
return std::make_unique<o2::framework::ServiceRegistry>();
29272945
}
29282946

2947+
o2::framework::WorkflowDefinitionContext o2::framework::defineWorkflow(std::vector<WorkflowDefinition> const& definitions, std::vector<std::string> const& names, int argc, char** argv)
2948+
{
2949+
auto append = [](auto& to, auto const& from) { to.insert(to.end(), from.begin(), from.end()); };
2950+
auto toString = [](Variant const& value) { std::ostringstream oss; oss << value; return oss.str(); };
2951+
WorkflowDefinitionContext result;
2952+
std::vector<WorkflowDefinitionContext> contexts(definitions.size());
2953+
for (size_t di = 0; di < definitions.size(); ++di) {
2954+
auto& context = contexts[di];
2955+
definitions[di].customize(context);
2956+
// Options already declared by a previous definition are shared, if they agree.
2957+
auto previous = result.workflowOptions.size();
2958+
for (auto& option : context.workflowOptions) {
2959+
auto end = result.workflowOptions.begin() + previous;
2960+
auto found = std::find_if(result.workflowOptions.begin(), end, [&option](auto const& other) { return other.name == option.name; });
2961+
if (found == end) {
2962+
result.workflowOptions.push_back(option);
2963+
} else if (found->type != option.type || toString(found->defaultValue) != toString(option.defaultValue)) {
2964+
throw runtime_error_f("Workflow option %s of %s has a different type or default than in a previous plugin.", option.name.c_str(), names[di].c_str());
2965+
}
2966+
}
2967+
append(result.completionPolicies, context.completionPolicies);
2968+
append(result.dispatchPolicies, context.dispatchPolicies);
2969+
append(result.resourcePolicies, context.resourcePolicies);
2970+
append(result.callbacksPolicies, context.callbacksPolicies);
2971+
append(result.sendingPolicies, context.sendingPolicies);
2972+
}
2973+
append(result.workflowOptions, WorkflowCustomizationHelpers::requiredWorkflowOptions());
2974+
append(result.completionPolicies, CompletionPolicy::createDefaultPolicies());
2975+
append(result.dispatchPolicies, DispatchPolicy::createDefaultPolicies());
2976+
append(result.resourcePolicies, ResourcePolicy::createDefaultPolicies());
2977+
append(result.callbacksPolicies, CallbacksPolicy::createDefaultPolicies());
2978+
append(result.sendingPolicies, SendingPolicy::createDefaultPolicies());
2979+
2980+
result.configRegistry = createRegistry();
2981+
result.configContext = createConfigContext(result.workflowOptionsRegistry, *result.configRegistry, result.workflowOptions, result.extraOptions, argc, argv);
2982+
2983+
std::map<std::string, size_t> definedBy;
2984+
for (size_t di = 0; di < definitions.size(); ++di) {
2985+
auto& context = contexts[di];
2986+
definitions[di].defineWorkflow(context, *result.configContext);
2987+
for (auto& spec : context.specs) {
2988+
auto [it, inserted] = definedBy.try_emplace(spec.name, di);
2989+
if (!inserted && it->second != di) {
2990+
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());
2991+
}
2992+
}
2993+
append(result.specs, context.specs);
2994+
append(result.channelPolicies, context.channelPolicies);
2995+
}
2996+
append(result.channelPolicies, ChannelConfigurationPolicy::createDefaultPolicies(*result.configContext));
2997+
return result;
2998+
}
2999+
29293000
// This is a toy executor for the workflow spec
29303001
// What it needs to do is:
29313002
//
@@ -3010,6 +3081,7 @@ int doMain(int argc, char** argv, o2::framework::WorkflowDefinitionContext& work
30103081
("id,i", bpo::value<std::string>(), "device id for child spawning") //
30113082
("channel-config", bpo::value<std::vector<std::string>>(), "channel configuration") //
30123083
("control", "control plugin") //
3084+
("workflow-plugin", bpo::value<std::string>(), "workflow plugins to load") //
30133085
("log-color", "logging color scheme")("color", "logging color scheme");
30143086

30153087
bpo::options_description visibleOptions;
Lines changed: 95 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,95 @@
1+
// Copyright 2019-2026 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+
12+
#include "Framework/WorkflowDefinitionContext.h"
13+
#include "Framework/CompletionPolicyHelpers.h"
14+
#include <catch_amalgamated.hpp>
15+
#include <algorithm>
16+
17+
using namespace o2::framework;
18+
19+
namespace
20+
{
21+
// A plugin with a processor called @a processor, which declares the
22+
// workflow options @a options and records the value of "shared" it sees.
23+
WorkflowDefinition makeDefinition(std::string processor, std::vector<ConfigParamSpec> options, int& shared)
24+
{
25+
return WorkflowDefinition{
26+
.customize = [processor, options](WorkflowDefinitionContext& context) {
27+
context.workflowOptions = options;
28+
context.completionPolicies.push_back(CompletionPolicyHelpers::consumeWhenAll(processor.c_str(), [](auto const&) { return false; })); },
29+
.defineWorkflow = [processor, &shared](WorkflowDefinitionContext& context, ConfigContext& configContext) {
30+
shared = configContext.options().get<int>("shared");
31+
context.specs.push_back(DataProcessorSpec{.name = processor}); }};
32+
}
33+
34+
size_t countOption(WorkflowDefinitionContext const& context, std::string const& name)
35+
{
36+
return std::count_if(context.workflowOptions.begin(), context.workflowOptions.end(), [&name](auto const& option) { return option.name == name; });
37+
}
38+
} // namespace
39+
40+
TEST_CASE("TestMergeWorkflowDefinitions")
41+
{
42+
int sharedA = 0;
43+
int sharedB = 0;
44+
std::vector<WorkflowDefinition> definitions{
45+
makeDefinition("a", {{"shared", VariantType::Int, 1, {"shared option"}}, {"a-only", VariantType::Int, 2, {"option of a"}}}, sharedA),
46+
makeDefinition("b", {{"shared", VariantType::Int, 1, {"shared option"}}, {"b-only", VariantType::String, "x", {"option of b"}}}, sharedB)};
47+
char const* args[] = {"test", "--shared", "3", "--b-only", "y"};
48+
auto context = defineWorkflow(definitions, {"libA:a", "libB:b"}, 5, const_cast<char**>(args));
49+
50+
// Options are parsed once and each plugin sees the value of the shared one.
51+
REQUIRE(countOption(context, "shared") == 1);
52+
REQUIRE(countOption(context, "a-only") == 1);
53+
REQUIRE(countOption(context, "b-only") == 1);
54+
REQUIRE(sharedA == 3);
55+
REQUIRE(sharedB == 3);
56+
REQUIRE(context.configContext->options().get<std::string>("b-only") == "y");
57+
// Options of the plugins come first, then those required by DPL.
58+
REQUIRE(context.workflowOptions[0].name == "shared");
59+
REQUIRE(context.workflowOptions[1].name == "a-only");
60+
REQUIRE(context.workflowOptions[2].name == "b-only");
61+
REQUIRE(countOption(context, "pipeline") == 1);
62+
63+
// Data processors and policies are merged in plugin order, the default policies last.
64+
REQUIRE(context.specs.size() == 2);
65+
REQUIRE(context.specs[0].name == "a");
66+
REQUIRE(context.specs[1].name == "b");
67+
REQUIRE(context.completionPolicies.size() == 2 + CompletionPolicy::createDefaultPolicies().size());
68+
REQUIRE(context.completionPolicies[0].name == "a");
69+
REQUIRE(context.completionPolicies[1].name == "b");
70+
REQUIRE(context.completionPolicies[2].name == CompletionPolicy::createDefaultPolicies()[0].name);
71+
}
72+
73+
TEST_CASE("TestConflictingWorkflowOptions")
74+
{
75+
int shared = 0;
76+
char const* args[] = {"test"};
77+
// Same option with a different default.
78+
std::vector<WorkflowDefinition> definitions{
79+
makeDefinition("a", {{"shared", VariantType::Int, 1, {"shared option"}}}, shared),
80+
makeDefinition("b", {{"shared", VariantType::Int, 2, {"shared option"}}}, shared)};
81+
REQUIRE_THROWS(defineWorkflow(definitions, {"libA:a", "libB:b"}, 1, const_cast<char**>(args)));
82+
// Same option with a different type.
83+
definitions[1] = makeDefinition("b", {{"shared", VariantType::Float, 1.f, {"shared option"}}}, shared);
84+
REQUIRE_THROWS(defineWorkflow(definitions, {"libA:a", "libB:b"}, 1, const_cast<char**>(args)));
85+
}
86+
87+
TEST_CASE("TestDuplicateDataProcessor")
88+
{
89+
int shared = 0;
90+
char const* args[] = {"test"};
91+
std::vector<WorkflowDefinition> definitions{
92+
makeDefinition("same", {{"shared", VariantType::Int, 1, {"shared option"}}}, shared),
93+
makeDefinition("same", {{"shared", VariantType::Int, 1, {"shared option"}}}, shared)};
94+
REQUIRE_THROWS(defineWorkflow(definitions, {"libA:a", "libB:b"}, 1, const_cast<char**>(args)));
95+
}

0 commit comments

Comments
 (0)