Merge 634c458 into sapling-pr-archive-ktf

Author: ktf

Detectors/EMCAL/base/src/ClusterFactory.cxx

@@ -528,8 +528,9 @@ void ClusterFactory<InputType>::evalNExMax(gsl::span<const int> inputsIndices, A
 
     // loop over all other cells in cluster
     for (size_t j = 0; j < n; j++) {
-      if (i == j)
+      if (i == j) {
         continue;
+      }
 
       // adjacent cell is any cell with adjacent phi or eta index
       if (std::abs(rows[i] - rows[j]) <= 1 &&

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"
 

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,36 +34,41 @@ 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>
 
 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) {
       };
     }}});