From 634c45837afb12c9741aa31de7dd84225ca9a916 Mon Sep 17 00:00:00 2001
From: Giulio Eulisse <10544+ktf@users.noreply.github.com>
Date: Wed, 11 Feb 2026 11:15:24 +0100
Subject: [PATCH] DPL Examples: use the new completion policy for parallel
 processing

This demonstrates how the new policy can be used in conjunction
with wildcards in order to simplify parallelism based on the
subSpecification.
---
 .../Framework/CompletionPolicyHelpers.h       |  1 -
 .../TestWorkflows/src/o2ParallelWorkflow.cxx  | 77 ++++++++++++-------
 2 files changed, 48 insertions(+), 30 deletions(-)

diff --git a/Framework/Core/include/Framework/CompletionPolicyHelpers.h b/Framework/Core/include/Framework/CompletionPolicyHelpers.h
index 9fce626854e5b..09ea8b7ea6b61 100644
--- a/Framework/Core/include/Framework/CompletionPolicyHelpers.h
+++ b/Framework/Core/include/Framework/CompletionPolicyHelpers.h
@@ -11,7 +11,6 @@
 #ifndef O2_FRAMEWORK_COMPLETIONPOLICYHELPERS_H_
 #define O2_FRAMEWORK_COMPLETIONPOLICYHELPERS_H_
 
-#include "Framework/ChannelSpec.h"
 #include "Framework/CompletionPolicy.h"
 #include "Headers/DataHeader.h"
 
diff --git a/Framework/TestWorkflows/src/o2ParallelWorkflow.cxx b/Framework/TestWorkflows/src/o2ParallelWorkflow.cxx
index 841f4a8f2b9bd..bdc08ad45ea24 100644
--- a/Framework/TestWorkflows/src/o2ParallelWorkflow.cxx
+++ b/Framework/TestWorkflows/src/o2ParallelWorkflow.cxx
@@ -9,7 +9,12 @@
 // granted to it by virtue of its status as an Intergovernmental Organization
 // or submit itself to any jurisdiction.
 
+#include "Framework/ConcreteDataMatcher.h"
 #include "Framework/ConfigParamSpec.h"
+#include "Framework/CompletionPolicy.h"
+#include "Framework/CompletionPolicyHelpers.h"
+#include "Framework/InputRecordWalker.h"
+#include "Framework/Logger.h"
 
 #include <chrono>
 #include <thread>
@@ -29,13 +34,16 @@ void customize(std::vector<ConfigParamSpec>& workflowOptions)
     ConfigParamSpec{"3-layer-pipelining", VariantType::Int, 1, {timeHelp}});
 }
 
+void customize(std::vector<CompletionPolicy>& policies)
+{
+  policies = {
+    CompletionPolicyHelpers::consumeWhenPastOldestPossibleTimeframe("merger-policy", [](auto const&) -> bool { return true; })};
+}
+
 #include "Framework/runDataProcessing.h"
 #include "Framework/DataProcessorSpec.h"
 #include "Framework/DataSpecUtils.h"
 #include "Framework/ParallelContext.h"
-#include "Framework/ControlService.h"
-
-#include "Framework/Logger.h"
 
 #include <vector>
 
@@ -43,22 +51,24 @@ using DataHeader = o2::header::DataHeader;
 
 DataProcessorSpec templateProcessor()
 {
-  return DataProcessorSpec{"some-processor", {
-                                               InputSpec{"x", "TST", "A", 0, Lifetime::Timeframe},
-                                             },
-                           {
+  return DataProcessorSpec{.name = "some-processor",
+                           .inputs = {
+                             InputSpec{"x", "TST", "A", 0, Lifetime::Timeframe},
+                           },
+                           .outputs = {
                              OutputSpec{"TST", "P", 0, Lifetime::Timeframe},
                            },
                            // The producer is stateful, we use a static for the state in this
                            // particular case, but a Singleton or a captured new object would
                            // work as well.
-                           AlgorithmSpec{[](InitContext& setup) {
+                           .algorithm = AlgorithmSpec{[](InitContext& setup) {
                              srand(setup.services().get<ParallelContext>().index1D());
                              return [](ProcessingContext& ctx) {
                                // Create a single output.
                                size_t index = ctx.services().get<ParallelContext>().index1D();
-                               auto& aData = ctx.outputs().make<int>(
+                               auto& i = ctx.outputs().make<int>(
                                  Output{"TST", "P", static_cast<o2::header::DataHeader::SubSpecificationType>(index)}, 1);
+                               i[0] = index;
                                std::this_thread::sleep_for(std::chrono::seconds(rand() % 5));
                              };
                            }}};
@@ -86,34 +96,43 @@ WorkflowSpec defineDataProcessing(ConfigContext const& config)
     outputSpecs.emplace_back("TST", "A", ssi);
   }
 
-  workflow.push_back(DataProcessorSpec{"reader", {}, outputSpecs, AlgorithmSpec{[jobs](InitContext& initCtx) {
-                                         return [jobs](ProcessingContext& ctx) {
-                                           for (size_t ji = 0; ji < jobs; ++ji) {
-                                             ctx.outputs().make<int>(Output{"TST", "A", static_cast<o2::header::DataHeader::SubSpecificationType>(ji)},
-                                                                     1);
-                                           }
-                                         };
-                                       }}});
+  workflow.push_back(DataProcessorSpec{
+    .name = "reader",
+    .outputs = outputSpecs,
+    .algorithm = AlgorithmSpec{[jobs](InitContext& initCtx) {
+      return [jobs](ProcessingContext& ctx) {
+        static int count = 0;
+        for (size_t ji = 0; ji < jobs; ++ji) {
+          int& i = ctx.outputs().make<int>(Output{"TST", "A", static_cast<o2::header::DataHeader::SubSpecificationType>(ji)});
+          i = count * 100 + ji;
+        }
+        count++;
+      };
+    }}});
   workflow.push_back(timePipeline(DataProcessorSpec{
-                                    "merger",
-                                    mergeInputs(InputSpec{"x", "TST", "P"},
-                                                jobs,
-                                                [](InputSpec& input, size_t index) {
-                                                  DataSpecUtils::updateMatchingSubspec(input, index);
-                                                }),
-                                    {OutputSpec{{"out"}, "TST", "M"}},
-                                    AlgorithmSpec{[](InitContext& setup) {
+                                    .name = "merger",
+                                    .inputs = {InputSpec{"all", ConcreteDataTypeMatcher{"TST", "P"}}},
+                                    .outputs = {OutputSpec{{"out"}, "TST", "M"}},
+                                    .algorithm = AlgorithmSpec{[](InitContext& setup) {
                                       return [](ProcessingContext& ctx) {
+                                        LOGP(info, "Run");
+                                        for (const auto& input : o2::framework::InputRecordWalker(ctx.inputs())) {
+                                          if (input.header == nullptr) {
+                                            LOGP(error, "Missing header");
+                                            continue;
+                                          }
+                                          int record = *(int*)input.payload;
+                                          LOGP(info, "Record {}", record);
+                                        }
                                         ctx.outputs().make<int>(OutputRef("out", 0), 1);
                                       };
                                     }}},
                                   stages));
 
   workflow.push_back(DataProcessorSpec{
-    "writer",
-    {InputSpec{"x", "TST", "M"}},
-    {},
-    AlgorithmSpec{[](InitContext& setup) {
+    .name = "writer",
+    .inputs = {InputSpec{"x", "TST", "M"}},
+    .algorithm = AlgorithmSpec{[](InitContext& setup) {
       return [](ProcessingContext& ctx) {
       };
     }}});