DPL: move topological sort in a separate file

ktf · ktf · commit ba88af5ad7ed · 2025-07-08T17:02:16.000+02:00
Simplifies testing.
diff --git a/Framework/Core/CMakeLists.txt b/Framework/Core/CMakeLists.txt
@@ -133,6 +133,7 @@ o2_add_library(Framework
                        src/TableConsumer.cxx
                        src/TableTreeHelpers.cxx
                        src/TopologyPolicy.cxx
+                       src/TopologyPolicyHelpers.cxx
                        src/TextDriverClient.cxx
                        src/TimesliceIndex.cxx
                        src/TimingHelpers.cxx
@@ -248,6 +249,7 @@ add_executable(o2-test-framework-core
               test/test_TimeParallelPipelining.cxx
               test/test_TimesliceIndex.cxx
               test/test_TypeTraits.cxx
+              test/test_TopologyPolicies.cxx
               test/test_Variants.cxx
               test/test_WorkflowHelpers.cxx
               test/test_WorkflowSerialization.cxx
diff --git a/Framework/Core/include/Framework/TopologyPolicyHelpers.h b/Framework/Core/include/Framework/TopologyPolicyHelpers.h
@@ -0,0 +1,23 @@
+// Copyright 2019-2025 CERN and copyright holders of ALICE O2.
+// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders.
+// All rights not expressly granted are reserved.
+//
+// This software is distributed under the terms of the GNU General Public
+// License v3 (GPL Version 3), copied verbatim in the file "COPYING".
+//
+// In applying this license CERN does not waive the privileges and immunities
+// granted to it by virtue of its status as an Intergovernmental Organization
+// or submit itself to any jurisdiction.
+
+#ifndef O2_FRAMEWORK_TOPOLOGYPOLICYHELPERS_H_
+#define O2_FRAMEWORK_TOPOLOGYPOLICYHELPERS_H_
+#include "Framework/WorkflowSpec.h"
+#include <vector>
+
+namespace o2::framework
+{
+struct TopologyPolicyHelpers {
+  static auto buildEdges(WorkflowSpec& physicalWorkflow) -> std::vector<std::pair<int, int>>;
+};
+} // namespace o2::framework
+#endif // O2_FRAMEWORK_TOPOLOGYPOLICYHELPERS_H_
diff --git a/Framework/Core/src/TopologyPolicyHelpers.cxx b/Framework/Core/src/TopologyPolicyHelpers.cxx
@@ -0,0 +1,89 @@
+// Copyright 2019-2025 CERN and copyright holders of ALICE O2.
+// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders.
+// All rights not expressly granted are reserved.
+//
+// This software is distributed under the terms of the GNU General Public
+// License v3 (GPL Version 3), copied verbatim in the file "COPYING".
+//
+// In applying this license CERN does not waive the privileges and immunities
+// granted to it by virtue of its status as an Intergovernmental Organization
+// or submit itself to any jurisdiction.
+
+#include "Framework/TopologyPolicyHelpers.h"
+#include "Framework/TopologyPolicy.h"
+
+namespace o2::framework {
+namespace {
+void describeDataProcessorSpec(std::ostream& stream, DataProcessorSpec const& spec)
+{
+  stream << spec.name;
+  if (!spec.labels.empty()) {
+    stream << "(";
+    bool first = false;
+    for (auto& label : spec.labels) {
+      stream << (first ? "" : ",") << label.value;
+      first = true;
+    }
+    stream << ")";
+  }
+}
+}
+
+auto TopologyPolicyHelpers::buildEdges(WorkflowSpec& physicalWorkflow) -> std::vector < std::pair<int, int>> {
+  std::vector<TopologyPolicy> topologyPolicies = TopologyPolicy::createDefaultPolicies();
+  std::vector<TopologyPolicy::DependencyChecker> dependencyCheckers;
+  dependencyCheckers.reserve(physicalWorkflow.size());
+
+  for (auto& spec : physicalWorkflow) {
+    for (auto& policy : topologyPolicies) {
+      if (policy.matcher(spec)) {
+        dependencyCheckers.push_back(policy.checkDependency);
+        break;
+      }
+    }
+  }
+  assert(dependencyCheckers.size() == physicalWorkflow.size());
+  // check if DataProcessorSpec at i depends on j
+  auto checkDependencies = [&workflow = physicalWorkflow,
+                            &dependencyCheckers](int i, int j) {
+    TopologyPolicy::DependencyChecker& checker = dependencyCheckers[i];
+    return checker(workflow[i], workflow[j]);
+  };
+  std::vector<std::pair<int, int>> edges;
+  for (size_t i = 0; i < physicalWorkflow.size() - 1; ++i) {
+    for (size_t j = i; j < physicalWorkflow.size(); ++j) {
+      if (i == j && checkDependencies(i, j)) {
+        throw std::runtime_error(physicalWorkflow[i].name + " depends on itself");
+      }
+      bool both = false;
+      if (checkDependencies(i, j)) {
+        edges.emplace_back(j, i);
+        both = true;
+      }
+      if (checkDependencies(j, i)) {
+        edges.emplace_back(i, j);
+        if (both) {
+          std::ostringstream str;
+          describeDataProcessorSpec(str, physicalWorkflow[i]);
+          str << " has circular dependency with ";
+          describeDataProcessorSpec(str, physicalWorkflow[j]);
+          str << ":\n";
+          for (auto x : {i, j}) {
+            str << physicalWorkflow[x].name << ":\n";
+            str << "inputs:\n";
+            for (auto& input : physicalWorkflow[x].inputs) {
+              str << "- " << input << " " << (int)input.lifetime << "\n";
+            }
+            str << "outputs:\n";
+            for (auto& output : physicalWorkflow[x].outputs) {
+              str << "- " << output << " " << (int)output.lifetime << "\n";
+            }
+          }
+          throw std::runtime_error(str.str());
+        }
+      }
+    }
+  }
+  return edges;
+};
+} // namespace o2::framework
diff --git a/Framework/Core/src/runDataProcessing.cxx b/Framework/Core/src/runDataProcessing.cxx
@@ -9,6 +9,7 @@
 // granted to it by virtue of its status as an Intergovernmental Organization
 // or submit itself to any jurisdiction.
 #include <memory>
+#include "Framework/TopologyPolicyHelpers.h"
 #define BOOST_BIND_GLOBAL_PLACEHOLDERS
 #include <stdexcept>
 #include "Framework/BoostOptionsRetriever.h"
@@ -2835,19 +2836,6 @@ std::unique_ptr<o2::framework::ServiceRegistry> createRegistry()
   return std::make_unique<o2::framework::ServiceRegistry>();
 }
 
-void describeDataProcessorSpec(std::ostream& stream, DataProcessorSpec const& spec)
-{
-  stream << spec.name;
-  if (!spec.labels.empty()) {
-    stream << "(";
-    bool first = false;
-    for (auto& label : spec.labels) {
-      stream << (first ? "" : ",") << label.value;
-      first = true;
-    }
-    stream << ")";
-  }
-}
 
 // This is a toy executor for the workflow spec
 // What it needs to do is:
@@ -3034,65 +3022,12 @@ int doMain(int argc, char** argv, o2::framework::WorkflowSpec const& workflow,
                      [](OutputSpec const& a, OutputSpec const& b) { return DataSpecUtils::describe(a) < DataSpecUtils::describe(b); });
   }
 
-  std::vector<TopologyPolicy> topologyPolicies = TopologyPolicy::createDefaultPolicies();
-  std::vector<TopologyPolicy::DependencyChecker> dependencyCheckers;
-  dependencyCheckers.reserve(physicalWorkflow.size());
-
-  for (auto& spec : physicalWorkflow) {
-    for (auto& policy : topologyPolicies) {
-      if (policy.matcher(spec)) {
-        dependencyCheckers.push_back(policy.checkDependency);
-        break;
-      }
-    }
-  }
-  assert(dependencyCheckers.size() == physicalWorkflow.size());
-  // check if DataProcessorSpec at i depends on j
-  auto checkDependencies = [&workflow = physicalWorkflow,
-                            &dependencyCheckers](int i, int j) {
-    TopologyPolicy::DependencyChecker& checker = dependencyCheckers[i];
-    return checker(workflow[i], workflow[j]);
-  };
-
   // Create a list of all the edges, so that we can do a topological sort
   // before we create the graph.
   std::vector<std::pair<int, int>> edges;
 
   if (physicalWorkflow.size() > 1) {
-    for (size_t i = 0; i < physicalWorkflow.size() - 1; ++i) {
-      for (size_t j = i; j < physicalWorkflow.size(); ++j) {
-        if (i == j && checkDependencies(i, j)) {
-          throw std::runtime_error(physicalWorkflow[i].name + " depends on itself");
-        }
-        bool both = false;
-        if (checkDependencies(i, j)) {
-          edges.emplace_back(j, i);
-          both = true;
-        }
-        if (checkDependencies(j, i)) {
-          edges.emplace_back(i, j);
-          if (both) {
-            std::ostringstream str;
-            describeDataProcessorSpec(str, physicalWorkflow[i]);
-            str << " has circular dependency with ";
-            describeDataProcessorSpec(str, physicalWorkflow[j]);
-            str << ":\n";
-            for (auto x : {i, j}) {
-              str << physicalWorkflow[x].name << ":\n";
-              str << "inputs:\n";
-              for (auto& input : physicalWorkflow[x].inputs) {
-                str << "- " << input << " " << (int)input.lifetime << "\n";
-              }
-              str << "outputs:\n";
-              for (auto& output : physicalWorkflow[x].outputs) {
-                str << "- " << output << " " << (int)output.lifetime << "\n";
-              }
-            }
-            throw std::runtime_error(str.str());
-          }
-        }
-      }
-    }
+    edges = TopologyPolicyHelpers::buildEdges(physicalWorkflow);
 
     auto topoInfos = WorkflowHelpers::topologicalSort(physicalWorkflow.size(), &edges[0].first, &edges[0].second, sizeof(std::pair<int, int>), edges.size());
     if (topoInfos.size() != physicalWorkflow.size()) {
