DPL: provide ability to customise consumption order

AliceO2Group · Jul 24, 2024 · 642289b · 642289b
1 parent c57a63a
commit 642289b
Show file tree

Hide file tree

Showing 2 changed files with 20 additions and 0 deletions.
diff --git a/Framework/Core/include/Framework/CompletionPolicy.h b/Framework/Core/include/Framework/CompletionPolicy.h
@@ -62,6 +62,13 @@ struct CompletionPolicy {
     Retry,
   };
 
+  /// Order in which the completed slots must be consumed
+  enum struct CompletionOrder {
+    Any,
+    Timeslice,
+    Slot
+  };
+
   using Matcher = std::function<bool(DeviceSpec const& device)>;
   using InputSetElement = DataRef;
   using CallbackFull = std::function<CompletionOp(InputSpan const&, std::vector<InputSpec> const&, ServiceRegistryRef&)>;
@@ -91,6 +98,8 @@ struct CompletionPolicy {
   /// data.
   bool balanceChannels = true;
 
+  CompletionOrder order = CompletionOrder::Any;
+
   /// Helper to create the default configuration.
   static std::vector<CompletionPolicy> createDefaultPolicies();
 };

diff --git a/Framework/Core/src/DataProcessingDevice.cxx b/Framework/Core/src/DataProcessingDevice.cxx
@@ -2307,6 +2307,17 @@ bool DataProcessingDevice::tryDispatchComputation(ServiceRegistryRef ref, std::v
   using namespace o2::framework;
   stats.updateStats({(int)ProcessingStatsId::PENDING_INPUTS, DataProcessingStats::Op::Set, static_cast<int64_t>(relayer.getParallelTimeslices() - completed.size())});
   stats.updateStats({(int)ProcessingStatsId::INCOMPLETE_INPUTS, DataProcessingStats::Op::Set, completed.empty() ? 1 : 0});
+  switch (spec.completionPolicy.order) {
+    case CompletionPolicy::CompletionOrder::Timeslice:
+      std::sort(completed.begin(), completed.end(), [](auto const& a, auto const& b) { return a.timeslice.value < b.timeslice.value; });
+      break;
+    case CompletionPolicy::CompletionOrder::Slot:
+      std::sort(completed.begin(), completed.end(), [](auto const& a, auto const& b) { return a.slot.index < b.slot.index; });
+      break;
+    case CompletionPolicy::CompletionOrder::Any:
+    default:
+      break;
+  }
 
   for (auto action : completed) {
     O2_SIGNPOST_ID_GENERATE(aid, device);