Skip to content

Commit

Permalink
DPL: first step to make workflow definition plugins
Browse files Browse the repository at this point in the history 
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.
  • Loading branch information
@ktf
ktf committed Feb 18, 2025
1 parent 02090ec commit c381e69
Show file tree
Hide file tree
Showing 3 changed files with 122 additions and 85 deletions.
3 changes: 3 additions & 0 deletions Framework/Core/include/Framework/Plugins.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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
};
Expand Down
126 changes: 78 additions & 48 deletions Framework/Core/include/Framework/runDataProcessing.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,9 @@
#include "Framework/CustomWorkflowTerminationHook.h"
#include "Framework/CommonServices.h"
#include "Framework/WorkflowCustomizationHelpers.h"
#include "Framework/WorkflowDefinitionContext.h"
#include "Framework/Logger.h"
#include "Framework/Plugins.h"
#include "Framework/CheckTypes.h"
#include "Framework/StructToTuple.h"
#include "ResourcePolicy.h"
Expand Down Expand Up @@ -125,16 +127,7 @@ void overrideCloning(o2::framework::ConfigContext& ctx, std::vector<o2::framewor
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);
int doMain(int argc, char** argv, o2::framework::WorkflowDefinitionContext& context, o2::framework::ConfigContext& configContext);

void doDefaultWorkflowTerminationHook();

Expand Down Expand Up @@ -167,60 +160,97 @@ 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);

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);
}
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);
#define STRINGIZE_NX(A) #A
#define STRINGIZE(A) STRINGIZE_NX(A)

// This is to allow the old "executable" based behavior
// Each executable will contain a plugin called InternalWorkflow
// In case one wants to use the new DSO based approach, the
// name of the plugin an the library name where it is located
// will have to be specified at build time.
#ifndef DPL_WORKFLOW_PLUGIN_NAME
#define DPL_WORKFLOW_PLUGIN_NAME InternalCustomWorkflow
#ifdef DPL_WORKFLOW_PLUGIN_LIBRARY
#error Missing DPL_WORKFLOW_PLUGIN_NAME
#endif
#define DPL_WORKFLOW_PLUGIN_LIBRARY ""
#endif

consteval char const* pluginName()
{
return DPL_WORKFLOW_PLUGIN_LIBRARY ":" STRINGIZE(DPL_WORKFLOW_PLUGIN_NAME);
}

int callMain(int argc, char** argv, int (*)(int, char**));
char* getIdString(int argc, char** argv);
// Executables behave this way
int callMain(int argc, char** argv, char const* pluginName);

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

int result = callMain(argc, argv, mainNoCatch);
int result = callMain(argc, argv, pluginName());

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

return result;
}

struct WorkflowDefinition {
std::function<o2::framework::WorkflowDefinitionContext(int argc, char** argv)> defineWorkflow;
};

struct DPL_WORKFLOW_PLUGIN_NAME : o2::framework::WorkflowPlugin {
o2::framework::WorkflowDefinition* create() override
{
return new o2::framework::WorkflowDefinition{
.defineWorkflow = [](int argc, char** argv) -> o2::framework::WorkflowDefinitionContext {
using namespace o2::framework;
WorkflowDefinitionContext workflowContext;

UserCustomizationsHelper::userDefinedCustomization(workflowContext.workflowOptions);
auto requiredWorkflowOptions = WorkflowCustomizationHelpers::requiredWorkflowOptions();
workflowContext.workflowOptions.insert(std::end(workflowContext.workflowOptions), std::begin(requiredWorkflowOptions), std::end(requiredWorkflowOptions));

workflowContext.completionPolicies = injectCustomizations<CompletionPolicy>();
workflowContext.dispatchPolicies = injectCustomizations<DispatchPolicy>();
workflowContext.resourcePolicies = injectCustomizations<ResourcePolicy>();
workflowContext.callbacksPolicies = injectCustomizations<CallbacksPolicy>();
workflowContext.sendingPolicies = injectCustomizations<SendingPolicy>();

workflowContext.configRegistry = createRegistry();
workflowContext.configContext = createConfigContext(workflowContext.workflowOptionsRegistry, *workflowContext.configRegistry, workflowContext.workflowOptions, workflowContext.extraOptions, argc, argv);

workflowContext.specs = defineDataProcessing(*workflowContext.configContext);
overrideAll(*workflowContext.configContext, workflowContext.specs);
for (auto& spec : workflowContext.specs) {
UserCustomizationsHelper::userDefinedCustomization(spec.requiredServices);
}
UserCustomizationsHelper::userDefinedCustomization(workflowContext.channelPolicies);
auto defaultChannelPolicies = ChannelConfigurationPolicy::createDefaultPolicies(*workflowContext.configContext);
workflowContext.channelPolicies.insert(std::end(workflowContext.channelPolicies), std::begin(defaultChannelPolicies), std::end(defaultChannelPolicies));
return workflowContext;
}};
}
};

// 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)
{
previous = new DPLPluginHandle{new DPL_WORKFLOW_PLUGIN_NAME{}, strdup(STRINGIZE(DPL_WORKFLOW_PLUGIN_NAME)), o2::framework::DplPluginKind::Workflow, previous};
return previous;
}
}

#endif
Loading

0 comments on commit c381e69

Please sign in to comment.