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.

Author: ktf

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 <typename T, typename... A>
+std::string getTaskName(TaskName const& first, A const&...)
+{
+  return first.value;
+}
+
+template <typename T, typename... A>
+std::string getTaskName(SetDefaultProcesses const&, TaskName const& second, A const&...)
+{
+  return second.value;
+}
+
+template <typename T, typename... A>
+std::string getTaskName(A const&...)
+{
+  return type_to_task_name(type_name<T>());
+}
+
 template <typename T, typename... A>
 auto getTaskNameSetProcesses(std::string& outputName, TaskName first, SetDefaultProcesses second, A... args)
 {
@@ -519,6 +538,10 @@ auto getTaskNameSetProcesses(std::string& outputName, A... args)
 template <typename T, typename... Args>
 DataProcessorSpec adaptAnalysisTask(ConfigContext const& ctx, Args&&... args)
 {
+  if (auto name = getTaskName<T>(args...) + ctx.options().get<std::string>("workflow-suffix"); !ctx.mayRun(name)) {
+    return DataProcessorSpecHelpers::placeholder(name);
+  }
+
   TH1::AddDirectory(false);
 
   std::string name_str;

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; }

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

Framework/Core/src/ConfigContext.cxx

@@ -10,6 +10,9 @@
 // or submit itself to any jurisdiction.
 
 #include "Framework/ConfigContext.h"
+#include <regex>
+#include <sys/stat.h>
+#include <unistd.h>
 
 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<std::string>("clone").empty()) {
+    return true;
+  }
+  // Pipelined devices get a _t<N> 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

Framework/Core/src/DataProcessorSpecHelpers.cxx

@@ -10,6 +10,7 @@
 // or submit itself to any jurisdiction.
 
 #include "Framework/DataProcessorSpecHelpers.h"
+#include "Framework/Logger.h"
 #include <string>
 #include <algorithm>
 
@@ -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

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

Framework/Core/test/test_AnalysisTask.cxx

@@ -14,6 +14,7 @@
 #include "Framework/AnalysisTask.h"
 #include "Framework/AnalysisDataModel.h"
 #include <iostream>
+#include <unistd.h>
 
 #include <catch_amalgamated.hpp>
 
@@ -197,6 +198,46 @@ struct MTask {
   void process(aod::McCollision const&, soa::SmallGroups<soa::Join<aod::Collisions, aod::McCollisionLabels>> 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<char**>(otherArgs)};
+  ConfigContext ownDevice{cfgc->options(), cfgc->services(), 3, const_cast<char**>(ownArgs)};
+  auto skipped = adaptAnalysisTask<CountedTask>(otherDevice);
+  auto own = adaptAnalysisTask<CountedTask>(ownDevice);
+  auto renamed = adaptAnalysisTask<CountedTask>(otherDevice, SetDefaultProcesses{}, TaskName{"another-task"});
+  dup2(stdinCopy, STDIN_FILENO);
+  close(stdinCopy);
+  close(fds[0]);
+  close(fds[1]);
+  auto driver = adaptAnalysisTask<CountedTask>(*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();