Adding helper function to create parallel pipelines from a pre-defined workflow

This DPL helper function creates a workflow with parallel pipelines, each serving
multiples of data channels of the same origin and description but different sub
specification. If total number of parallel channels is larger than the number of
pipelines, the channels are distributed among the pipelines.

Author: matthiasrichter

Framework/Core/CMakeLists.txt

@@ -276,6 +276,7 @@ set(TEST_SRCS
       test/test_Variants.cxx
       test/test_WorkflowHelpers.cxx
       test/test_DeviceSpecHelpers.cxx
+      test/test_ParallelPipeline.cxx
    )
 
 set(BENCH_SRCS 

Framework/Core/include/Framework/WorkflowSpec.h

@@ -37,6 +37,44 @@ WorkflowSpec parallel(WorkflowSpec specs,
                       size_t maxIndex,
                       std::function<void(DataProcessorSpec&, size_t id)> amendCallback);
 
+/// create parallel pipelines of processors from a template sequence for a number of
+/// parallel sub specification IDs. The sub specifications are distributed among the
+/// pipelines.
+/// serves the case where each input id (subspec) corresponds to outputs amended with the
+/// same subspec. Two callback functions allow two configure the list of subspecs for the
+/// call.
+///
+/// Schematic workflow illustration:
+///            template pipeline                           parallel pipelines
+///                                                    ----       -----       ----
+///                                                   |  A0|---->|A0 B0|---->|B0   |
+///                                                   |    |     |   C0|---->|C0   |
+///                                                   |  A1|---->|A1 B1|---->|B1   |
+///                                                   |    |     |   C1|---->|C1   |
+///                                                    ----       -----       ----
+///
+///                                                    ----       -----       ----
+///       ----       ----       ----                  |  A2|---->|A2 B2|---->|B2   |
+///      |   A|---->|A  B|---->|B   |    becomes      |    |     |   C2|---->|C2   |
+///      |    |     |   C|---->|C   |    ======>      |  A3|---->|A3 B3|---->|B3   |
+///       ----       ----       ----                  |    |     |   C3|---->|C3   |
+///                                                    ----       -----       ----
+///                                                                 .
+///                                                                 .
+///                                                    ----       -----       ----
+///                                                   |  An|---->|An Bn|---->|Bn   |
+///                                                   |    |     |   Cn|---->|Cn   |
+///                                                    ----       -----       ----
+///
+/// @param specs               the template to be multiplied
+/// @param nPipelines          number of pipelines
+/// @param getNumberOfSubspecs callback function to return the number of subspecs
+/// @param getSubSpec          callback function to return the subspecs at index
+WorkflowSpec parallelPipeline(const WorkflowSpec& specs,
+                              size_t nPipelines,
+                              std::function<size_t()> getNumberOfSubspecs,
+                              std::function<size_t(size_t)> getSubSpec);
+
 /// The purpose of this helper is to duplicate an InputSpec @a original
 /// as many times as specified in maxIndex and to amend each instance
 /// by invoking amendCallback on them with their own @a id. This can be

Framework/Core/src/WorkflowSpec.cxx

@@ -10,6 +10,7 @@
 #include "Framework/WorkflowSpec.h"
 #include "Framework/DataProcessorSpec.h"
 #include "Framework/DataDescriptorQueryBuilder.h"
+#include "Framework/DataSpecUtils.h"
 
 #include <cstddef>
 #include <functional>
@@ -49,6 +50,63 @@ WorkflowSpec parallel(WorkflowSpec specs,
   return results;
 }
 
+WorkflowSpec parallelPipeline(const WorkflowSpec& specs,
+                              size_t nPipelines,
+                              std::function<size_t()> getNumberOfSubspecs,
+                              std::function<size_t(size_t)> getSubSpec)
+{
+  WorkflowSpec result;
+  size_t numberOfSubspecs = getNumberOfSubspecs();
+  if (numberOfSubspecs < nPipelines) {
+    // no need to create more pipelines than the number of parallel Ids, in that case
+    // each pipeline serves one id
+    nPipelines = numberOfSubspecs;
+  }
+  for (auto process : specs) {
+    size_t index = 0;
+    size_t inputMultiplicity = numberOfSubspecs / nPipelines;
+    if (numberOfSubspecs % nPipelines) {
+      inputMultiplicity += 1;
+    }
+    auto amendProcess = [numberOfSubspecs, nPipelines, &index, &inputMultiplicity, getSubSpec](DataProcessorSpec& spec, size_t pipeline) {
+      auto inputs = std::move(spec.inputs);
+      auto outputs = std::move(spec.outputs);
+      spec.inputs.reserve(inputMultiplicity);
+      spec.outputs.reserve(inputMultiplicity);
+      for (size_t inputNo = 0; inputNo < inputMultiplicity; ++inputNo) {
+        for (auto& input : inputs) {
+          spec.inputs.push_back(input);
+          spec.inputs.back().binding += std::to_string(inputNo);
+          DataSpecUtils::updateMatchingSubspec(spec.inputs.back(), getSubSpec(index + inputNo));
+        }
+        for (auto& output : outputs) {
+          spec.outputs.push_back(output);
+          spec.outputs.back().binding.value += std::to_string(inputNo);
+          spec.outputs.back().subSpec = getSubSpec(index + inputNo);
+        }
+      }
+      index += inputMultiplicity;
+      if (inputMultiplicity > numberOfSubspecs / nPipelines &&
+          ((numberOfSubspecs - index) % (nPipelines - (pipeline + 1))) == 0) {
+        // if the remaining ids can be distributed equally among the remaining pipelines
+        // we can decrease multiplicity
+        inputMultiplicity = numberOfSubspecs / nPipelines;
+      }
+    };
+
+    if (nPipelines > 1) {
+      // add multiple processes and distribute inputs among them
+      auto amendedProcessors = parallel(process, nPipelines, amendProcess);
+      result.insert(result.end(), amendedProcessors.begin(), amendedProcessors.end());
+    } else if (nPipelines == 1) {
+      // add one single process with all the inputs
+      amendProcess(process, 0);
+      result.push_back(process);
+    }
+  }
+  return result;
+}
+
 Inputs mergeInputs(InputSpec original,
                    size_t maxIndex,
                    std::function<void(InputSpec &, size_t)> amendCallback) {

Framework/Core/test/test_ParallelPipeline.cxx

@@ -0,0 +1,153 @@
+// Copyright CERN and copyright holders of ALICE O2. This software is
+// distributed under the terms of the GNU General Public License v3 (GPL
+// Version 3), copied verbatim in the file "COPYING".
+//
+// See http://alice-o2.web.cern.ch/license for full licensing information.
+//
+// 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/InputSpec.h"
+#include "Framework/DataProcessorSpec.h"
+#include "Framework/DataSpecUtils.h"
+#include "Framework/ParallelContext.h"
+#include "Framework/runDataProcessing.h"
+#include "Framework/ControlService.h"
+#include "Framework/ParallelContext.h"
+#include <iostream>
+#include <algorithm>
+#include <memory>
+#include <unordered_map>
+
+#define ASSERT_ERROR(condition)                                   \
+  if ((condition) == false) {                                     \
+    LOG(ERROR) << R"(Test condition ")" #condition R"(" failed)"; \
+  }
+
+using DataHeader = o2::header::DataHeader;
+using namespace o2::framework;
+
+size_t nPipelines = 4;
+size_t nParallelChannels = 6;
+size_t nRolls = 1;
+
+std::vector<DataProcessorSpec> defineDataProcessing(ConfigContext const&)
+{
+  // define a template workflow with processors to be executed in a pipeline
+  std::vector<DataProcessorSpec> workflowSpecs{
+    { "processor1",
+      Inputs{
+        { "input", "TST", "TRIGGER", 0, Lifetime::Timeframe } },
+      Outputs{
+        { { "output" }, "TST", "PREPROC", 0, Lifetime::Timeframe } },
+      AlgorithmSpec{ [](ProcessingContext& ctx) {
+        for (auto const& input : ctx.inputs()) {
+          auto const& parallelContext = ctx.services().get<ParallelContext>();
+          std::cout << "instance " << parallelContext.index1D() << " of " << parallelContext.index1DSize() << ": "
+                    << *input.spec << ": " << *((int*)input.payload) << std::endl;
+          auto const* dataheader = DataRefUtils::getHeader<o2::header::DataHeader*>(input);
+          //auto data& = ctx.outputs().make<int>(OutputRef{"output", dataheader->subSpecification});
+          auto& data = ctx.outputs().make<int>(Output{ "TST", "PREPROC", dataheader->subSpecification, Lifetime::Timeframe });
+          ASSERT_ERROR(ctx.inputs().get<int>(input.spec->binding.c_str()) == parallelContext.index1D());
+          data = parallelContext.index1D();
+        }
+      } } },
+    { "processor2",
+      Inputs{
+        { "input", "TST", "PREPROC", 0, Lifetime::Timeframe } },
+      Outputs{
+        { { "output" }, "TST", "DATA", 0, Lifetime::Timeframe },
+        { { "metadt" }, "TST", "META", 0, Lifetime::Timeframe } },
+      AlgorithmSpec{ [](ProcessingContext& ctx) {
+        for (auto const& input : ctx.inputs()) {
+          auto const& parallelContext = ctx.services().get<ParallelContext>();
+          std::cout << "instance " << parallelContext.index1D() << " of " << parallelContext.index1DSize() << ": "
+                    << *input.spec << ": " << *((int*)input.payload) << std::endl;
+          ASSERT_ERROR(ctx.inputs().get<int>(input.spec->binding.c_str()) == parallelContext.index1D());
+          auto const* dataheader = DataRefUtils::getHeader<o2::header::DataHeader*>(input);
+          // TODO: there is a bug in the API for using OutputRef, returns an rvalue which can not be bound to
+          // lvalue reference
+          //auto& data = ctx.outputs().make<int>(OutputRef{"output", dataheader->subSpecification});
+          auto& data = ctx.outputs().make<int>(Output{ "TST", "DATA", dataheader->subSpecification, Lifetime::Timeframe });
+          data = ctx.inputs().get<int>(input.spec->binding.c_str());
+          //auto meta& = ctx.outputs().make<int>(OutputRef{"metadt", dataheader->subSpecification});
+          auto& meta = ctx.outputs().make<int>(Output{ "TST", "META", dataheader->subSpecification, Lifetime::Timeframe });
+          meta = dataheader->subSpecification;
+        }
+      } } },
+  };
+
+  // create parallel pipelines from the template workflow, the number of parallel channel is defined by
+  // nParallelChannels and is distributed among the pipelines
+  std::vector<o2::header::DataHeader::SubSpecificationType> subspecs(nParallelChannels);
+  std::generate(subspecs.begin(), subspecs.end(), [counter = std::make_shared<int>(0)]() { return 0x1 << (*counter)++; });
+  workflowSpecs = parallelPipeline(workflowSpecs, nPipelines,
+                                   [&subspecs]() { return subspecs.size(); },
+                                   [&subspecs](size_t index) { return subspecs[index]; });
+
+  // define a producer process with outputs for all subspecs
+  auto producerOutputs = [&subspecs]() {
+    Outputs outputs;
+    for (auto const& subspec : subspecs) {
+      outputs.emplace_back("TST", "TRIGGER", subspec, Lifetime::Timeframe);
+    }
+    return outputs;
+  };
+
+  // we keep the correspondence between the subspec and the instance which serves this particular subspec
+  // this is checked in the final consumer
+  auto checkMap = std::make_shared<std::unordered_map<o2::header::DataHeader::SubSpecificationType, int>>();
+  workflowSpecs.emplace_back(DataProcessorSpec{
+    "trigger",
+    Inputs{},
+    producerOutputs(),
+    AlgorithmSpec{ [subspecs, checkMap, counter = std::make_shared<int>(0)](ProcessingContext& ctx) {
+      if (*counter < nRolls) {
+        size_t multiplicity = subspecs.size() / nPipelines;
+        if (subspecs.size() % nPipelines) {
+          multiplicity++;
+        }
+        size_t instance = 0;
+        size_t inputRank = 0;
+        for (size_t index = 0, end = subspecs.size(); index < end; index++) {
+          ctx.outputs().make<int>(Output{ "TST", "TRIGGER", subspecs[index], Lifetime::Timeframe }) = instance;
+          (*checkMap)[subspecs[index]] = instance;
+          if (++inputRank == multiplicity) {
+            inputRank = 0;
+            instance++;
+            if (instance < nPipelines && ((subspecs.size() - index - 1) % (nPipelines - instance)) == 0) {
+              multiplicity = subspecs.size() / nPipelines;
+            }
+          }
+        }
+        (*counter)++;
+      }
+      if (*counter == nRolls) {
+        ctx.services().get<ControlService>().readyToQuit(false);
+      }
+    } } });
+
+  // the final consumer
+  workflowSpecs.emplace_back(DataProcessorSpec{
+    "consumer",
+    mergeInputs({ { "datain", "TST", "DATA", 0, Lifetime::Timeframe },
+                  { "metain", "TST", "META", 0, Lifetime::Timeframe } },
+                subspecs.size(),
+                [&subspecs](InputSpec& input, size_t index) {
+                  DataSpecUtils::updateMatchingSubspec(input, subspecs[index]);
+                }),
+    Outputs(),
+    AlgorithmSpec{ [checkMap](ProcessingContext& ctx) {
+      for (auto const& input : ctx.inputs()) {
+        std::cout << "consuming : " << *input.spec << ": " << *((int*)input.payload) << std::endl;
+        auto const* dataheader = DataRefUtils::getHeader<o2::header::DataHeader*>(input);
+        if (input.spec->binding.compare(0, 6, "datain") == 0) {
+          ASSERT_ERROR((*checkMap)[dataheader->subSpecification] == ctx.inputs().get<int>(input.spec->binding.c_str()));
+        }
+      }
+      ctx.services().get<ControlService>().readyToQuit(true);
+    } } });
+
+  return workflowSpecs;
+}