DPL Analysis: do not build the tasks a device does not run (#16036)

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"
@@ -3038,9 +3039,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();