From dcae84b622fa0baaf767e31c66f87198b2b53d6c Mon Sep 17 00:00:00 2001 From: Giulio Eulisse <10544+ktf@users.noreply.github.com> Date: Mon, 2 Nov 2026 14:25:49 +0100 Subject: [PATCH] DPL Analysis: do not build the tasks a device does not run Every device used to construct all the analysis tasks of its executable only to describe them, although the driver already passes their specs on stdin. Their constructors create histograms, which start the ROOT interpreter in the reader, the writer and the other internal devices. A device now builds only its own task and takes the specs of the others from the driver. Without a workflow on stdin, or with --clone, everything is still built. Reader of o2-testworkflows-analysis-histograms on macOS: physical footprint 120.5 MB -> 74.7 MB, libCling and the .pcm modules no longer mapped. Output identical. --- .../Core/include/Framework/AnalysisTask.h | 23 +++++++++++ .../Core/include/Framework/ConfigContext.h | 3 ++ .../Framework/DataProcessorSpecHelpers.h | 4 ++ Framework/Core/src/ConfigContext.cxx | 22 ++++++++++ .../Core/src/DataProcessorSpecHelpers.cxx | 17 ++++++++ Framework/Core/src/runDataProcessing.cxx | 10 +++++ Framework/Core/test/test_AnalysisTask.cxx | 41 +++++++++++++++++++ 7 files changed, 120 insertions(+) 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();