DPL: oldest possible timeframe triggered CompletionPolicy (#15046)

This will trigger the processing whenever a given slot will not receive data
anymore in virtue of its timeslice being past the oldest possible timeframe.

Author: ktf

Framework/Core/include/Framework/CompletionPolicyHelpers.h

@@ -1,4 +1,4 @@
-// Copyright 2019-2020 CERN and copyright holders of ALICE O2.
+// Copyright 2019-2026 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.
 //
@@ -54,6 +54,12 @@ struct CompletionPolicyHelpers {
   }
   static CompletionPolicy consumeWhenAny(std::string matchName);
 
+  // Consume all the data captured until the oldest possible timeframe
+  // in input indicates that nothing else can be added to this timeslice.
+  // Useful in case of wildcards which multiplex multiple subspecs on the
+  // same input.
+  static CompletionPolicy consumeWhenPastOldestPossibleTimeframe(const char* name, CompletionPolicy::Matcher matcher);
+
   /// When any of the parts of the record have been received, consume them.
   static CompletionPolicy consumeWhenAnyWithAllConditions(const char* name, CompletionPolicy::Matcher matcher);
   /// Default matcher applies for all devices

Framework/Core/src/CompletionPolicyHelpers.cxx

@@ -1,4 +1,4 @@
-// Copyright 2019-2020 CERN and copyright holders of ALICE O2.
+// Copyright 2019-2026 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.
 //
@@ -11,6 +11,7 @@
 
 #include "Framework/CompletionPolicyHelpers.h"
 #include "Framework/CompletionPolicy.h"
+#include "Framework/DataProcessingHeader.h"
 #include "Framework/InputSpan.h"
 #include "Framework/DeviceSpec.h"
 #include "Framework/CompilerBuiltins.h"
@@ -263,6 +264,32 @@ CompletionPolicy CompletionPolicyHelpers::consumeWhenAnyZeroCount(const char* na
   return CompletionPolicy{name, matcher, callback, false};
 }
 
+CompletionPolicy CompletionPolicyHelpers::consumeWhenPastOldestPossibleTimeframe(const char* name, CompletionPolicy::Matcher matcher)
+{
+  auto callback = [](InputSpan const& inputs, std::vector<InputSpec> const&, ServiceRegistryRef& ref) -> CompletionPolicy::CompletionOp {
+    size_t currentTimeslice = -1;
+    for (auto& input : inputs) {
+      if (input.header == nullptr) {
+        continue;
+      }
+      o2::framework::DataProcessingHeader const* dph = o2::header::get<o2::framework::DataProcessingHeader*>(input.header);
+      if (dph && !TimingInfo::timesliceIsTimer(dph->startTime)) {
+        currentTimeslice = dph->startTime;
+        break;
+      }
+    }
+
+    auto& timesliceIndex = ref.get<TimesliceIndex>();
+    auto oldestPossibleTimeslice = timesliceIndex.getOldestPossibleInput().timeslice.value;
+
+    if (currentTimeslice >= oldestPossibleTimeslice) {
+      return CompletionPolicy::CompletionOp::Retry;
+    }
+    return CompletionPolicy::CompletionOp::Consume;
+  };
+  return CompletionPolicy{name, matcher, callback, false};
+}
+
 CompletionPolicy CompletionPolicyHelpers::consumeWhenAny(const char* name, CompletionPolicy::Matcher matcher)
 {
   auto callback = [](InputSpan const& inputs, std::vector<InputSpec> const&, ServiceRegistryRef& ref) -> CompletionPolicy::CompletionOp {