diff --git a/Framework/Core/include/Framework/AnalysisTask.h b/Framework/Core/include/Framework/AnalysisTask.h index 48a189f241cf0..bfd67a03ecbc1 100644 --- a/Framework/Core/include/Framework/AnalysisTask.h +++ b/Framework/Core/include/Framework/AnalysisTask.h @@ -19,6 +19,7 @@ #include "Framework/ConfigParamsHelper.h" #include "Framework/ControlService.h" #include "Framework/DataProcessorSpec.h" +#include "Framework/DataProcessorSpecHelpers.h" #include "Framework/Expressions.h" #include "Framework/EndOfStreamContext.h" #include "Framework/GroupSlicer.h" @@ -450,6 +451,24 @@ struct TaskName { namespace { +template +std::string getTaskName(TaskName const& first, A const&...) +{ + return first.value; +} + +template +std::string getTaskName(SetDefaultProcesses const&, TaskName const& second, A const&...) +{ + return second.value; +} + +template +std::string getTaskName(A const&...) +{ + return type_to_task_name(type_name()); +} + template auto getTaskNameSetProcesses(std::string& outputName, TaskName first, SetDefaultProcesses second, A... args) { @@ -519,6 +538,10 @@ auto getTaskNameSetProcesses(std::string& outputName, A... args) template DataProcessorSpec adaptAnalysisTask(ConfigContext const& ctx, Args&&... args) { + if (auto name = getTaskName(args...) + ctx.options().get("workflow-suffix"); !ctx.mayRun(name)) { + return DataProcessorSpecHelpers::placeholder(name); + } + TH1::AddDirectory(false); std::string name_str; diff --git a/Framework/Core/include/Framework/ConfigContext.h b/Framework/Core/include/Framework/ConfigContext.h index 87259f0519915..138e9e577e774 100644 --- a/Framework/Core/include/Framework/ConfigContext.h +++ b/Framework/Core/include/Framework/ConfigContext.h @@ -29,6 +29,9 @@ class ConfigContext [[nodiscard]] ServiceRegistryRef services() const { return mServices; } [[nodiscard]] bool helpOnCommandLine() const; + /// Whether this process may run @a processor. A device only runs its own; + /// it gets the specs of the others from the driver, so they need not be built. + [[nodiscard]] bool mayRun(std::string_view processor) const; [[nodiscard]] int argc() const { return mArgc; } [[nodiscard]] char* const* argv() const { return mArgv; } diff --git a/Framework/Core/include/Framework/DataProcessorSpecHelpers.h b/Framework/Core/include/Framework/DataProcessorSpecHelpers.h index 361242f44b570..ec89b3ac350e6 100644 --- a/Framework/Core/include/Framework/DataProcessorSpecHelpers.h +++ b/Framework/Core/include/Framework/DataProcessorSpecHelpers.h @@ -18,6 +18,10 @@ namespace o2::framework { struct DataProcessorSpecHelpers { static bool hasLabel(DataProcessorSpec const& spec, char const* name); + /// Stands for a processor this process does not run: everything but the name + /// is filled from the workflow the driver passes to the device. + static DataProcessorSpec placeholder(std::string name); + static bool isPlaceholder(DataProcessorSpec const& spec); }; } // namespace o2::framework #endif diff --git a/Framework/Core/src/ConfigContext.cxx b/Framework/Core/src/ConfigContext.cxx index 9b121b1884998..184dd3f004b10 100644 --- a/Framework/Core/src/ConfigContext.cxx +++ b/Framework/Core/src/ConfigContext.cxx @@ -10,6 +10,9 @@ // or submit itself to any jurisdiction. #include "Framework/ConfigContext.h" +#include +#include +#include namespace o2::framework { @@ -29,4 +32,23 @@ bool ConfigContext::helpOnCommandLine() const return helpasked; } +bool ConfigContext::mayRun(std::string_view processor) const +{ + std::string_view id; + for (int argi = 0; argi + 1 < mArgc; ++argi) { + if (strcmp(mArgv[argi], "--id") == 0) { + id = mArgv[argi + 1]; + } + } + // The driver, a device without a workflow on stdin and clones, which keep the + // name of their template, build everything. + struct stat s; + if (id.empty() || fstat(STDIN_FILENO, &s) != 0 || !(S_ISFIFO(s.st_mode) || S_ISREG(s.st_mode)) || !mOptions.get("clone").empty()) { + return true; + } + // Pipelined devices get a _t suffix, DDS ones a _dds<...> one. + auto name = std::regex_replace(std::regex_replace(std::string{id}, std::regex{"_dds.*"}, ""), std::regex{"_t[0-9]+$"}, ""); + return id == processor || name == processor; +} + } // namespace o2::framework diff --git a/Framework/Core/src/DataProcessorSpecHelpers.cxx b/Framework/Core/src/DataProcessorSpecHelpers.cxx index 17fc58a69afcf..fa23bcea4ca82 100644 --- a/Framework/Core/src/DataProcessorSpecHelpers.cxx +++ b/Framework/Core/src/DataProcessorSpecHelpers.cxx @@ -10,6 +10,7 @@ // or submit itself to any jurisdiction. #include "Framework/DataProcessorSpecHelpers.h" +#include "Framework/Logger.h" #include #include @@ -20,4 +21,20 @@ bool DataProcessorSpecHelpers::hasLabel(DataProcessorSpec const& spec, char cons auto sameLabel = [other = DataProcessorLabel{{label}}](DataProcessorLabel const& label) { return label == other; }; return std::find_if(spec.labels.begin(), spec.labels.end(), sameLabel) != spec.labels.end(); } + +DataProcessorSpec DataProcessorSpecHelpers::placeholder(std::string name) +{ + return DataProcessorSpec{ + .name = name, + .algorithm = AlgorithmSpec::InitCallback{[name](InitContext&) -> AlgorithmSpec::ProcessCallback { + LOGP(fatal, "{} was not built in this process, which therefore cannot run it.", name); + return {}; + }}, + .metadata = {{"dpl-placeholder", "true"}}}; +} + +bool DataProcessorSpecHelpers::isPlaceholder(DataProcessorSpec const& spec) +{ + return std::ranges::any_of(spec.metadata, [](auto const& m) { return m.key == "dpl-placeholder"; }); +} } // namespace o2::framework diff --git a/Framework/Core/src/runDataProcessing.cxx b/Framework/Core/src/runDataProcessing.cxx index c4b2cb892f79d..0f75b9ee82cb9 100644 --- a/Framework/Core/src/runDataProcessing.cxx +++ b/Framework/Core/src/runDataProcessing.cxx @@ -26,6 +26,7 @@ #include "Framework/DataProcessingDevice.h" #include "Framework/DataProcessingContext.h" #include "Framework/DataProcessorSpec.h" +#include "Framework/DataProcessorSpecHelpers.h" #include "Framework/PluginManager.h" #include "Framework/DeviceControl.h" #include "Framework/DeviceExecution.h" @@ -3037,9 +3038,18 @@ int doMain(int argc, char** argv, o2::framework::WorkflowSpec const& workflow, if (found == physicalWorkflow.end()) { physicalWorkflow.push_back(dp); rankIndex.insert(std::make_pair(dp.name, workflowHashB)); + } else if (DataProcessorSpecHelpers::isPlaceholder(*found)) { + auto algorithm = std::move(found->algorithm); + *found = dp; + found->algorithm = std::move(algorithm); } } } + for (auto& dp : physicalWorkflow) { + if (DataProcessorSpecHelpers::isPlaceholder(dp)) { + LOGP(fatal, "{} was not built, but the driver did not pass its spec.", dp.name); + } + } /// Iterate over the physicalWorkflow, any DataProcessorSpec that has a /// expendable label should have all the timeframe lifetime outputs changed diff --git a/Framework/Core/test/test_AnalysisTask.cxx b/Framework/Core/test/test_AnalysisTask.cxx index cb710b9a3871c..aeb79e82de383 100644 --- a/Framework/Core/test/test_AnalysisTask.cxx +++ b/Framework/Core/test/test_AnalysisTask.cxx @@ -14,6 +14,7 @@ #include "Framework/AnalysisTask.h" #include "Framework/AnalysisDataModel.h" #include +#include #include @@ -197,6 +198,46 @@ struct MTask { void process(aod::McCollision const&, soa::SmallGroups> const&) {} }; +struct BuildCounter { + static inline int built = 0; + BuildCounter() { ++built; } +}; + +struct CountedTask { + BuildCounter counter; + void process(aod::Tracks const&) {} +}; + +TEST_CASE("AdaptorBuildsOnlyTheTaskOfTheDevice") +{ + auto cfgc = makeEmptyConfigContext(); + // A device gets the workflow from the driver on stdin. + int fds[2]; + REQUIRE(pipe(fds) == 0); + int stdinCopy = dup(STDIN_FILENO); + dup2(fds[0], STDIN_FILENO); + char const* otherArgs[] = {"workflow", "--id", "another-task"}; + char const* ownArgs[] = {"workflow", "--id", "counted-task_t1"}; + ConfigContext otherDevice{cfgc->options(), cfgc->services(), 3, const_cast(otherArgs)}; + ConfigContext ownDevice{cfgc->options(), cfgc->services(), 3, const_cast(ownArgs)}; + auto skipped = adaptAnalysisTask(otherDevice); + auto own = adaptAnalysisTask(ownDevice); + auto renamed = adaptAnalysisTask(otherDevice, SetDefaultProcesses{}, TaskName{"another-task"}); + dup2(stdinCopy, STDIN_FILENO); + close(stdinCopy); + close(fds[0]); + close(fds[1]); + auto driver = adaptAnalysisTask(*cfgc); + + REQUIRE(BuildCounter::built == 3); + REQUIRE(DataProcessorSpecHelpers::isPlaceholder(skipped)); + REQUIRE(skipped.name == own.name); + REQUIRE(skipped.inputs.empty()); + REQUIRE(!DataProcessorSpecHelpers::isPlaceholder(own)); + REQUIRE(own.inputs.size() == driver.inputs.size()); + REQUIRE(!DataProcessorSpecHelpers::isPlaceholder(renamed)); +} + TEST_CASE("AdaptorCompilation") { auto cfgc = makeEmptyConfigContext();