Syncing to new DPL CompletionPolicy API after O2 PR 2886 (#304)

Pull request AliceO2Group/AliceO2#2886 changes the argument of the policy
callback from span<PartRef> to span<DataRef> which is a span of non-message-owning
reference objects.

In order to make depending packages easier to implement, the callback signature is
now defined in CompletionPolicy.h header in AliceO2Group/AliceO2#2918, thee
definitions are now used in the TaskRunner.

Author: matthiasrichter

Framework/include/QualityControl/TaskRunner.h

@@ -82,7 +82,7 @@ class TaskRunner : public framework::Task
   void run(framework::ProcessingContext& pCtx) override;
 
   /// \brief TaskRunner's completion policy callback
-  static framework::CompletionPolicy::CompletionOp completionPolicyCallback(gsl::span<framework::PartRef const> const& inputs);
+  static framework::CompletionPolicy::CompletionOp completionPolicyCallback(o2::framework::CompletionPolicy::InputSet inputs);
 
   std::string getDeviceName() { return mDeviceName; };
   const framework::Inputs& getInputsSpecs() { return mInputSpecs; };

Framework/src/TaskRunner.cxx

@@ -24,6 +24,7 @@
 #include <Monitoring/MonitoringFactory.h>
 #include <Framework/DataSampling.h>
 #include <Framework/CallbackService.h>
+#include <Framework/CompletionPolicyHelpers.h>
 #include <Framework/TimesliceIndex.h>
 #include <Framework/DataSpecUtils.h>
 #include <Framework/DataDescriptorQueryBuilder.h>
@@ -110,7 +111,7 @@ void TaskRunner::run(ProcessingContext& pCtx)
   }
 }
 
-CompletionPolicy::CompletionOp TaskRunner::completionPolicyCallback(gsl::span<PartRef const> const& inputs)
+CompletionPolicy::CompletionOp TaskRunner::completionPolicyCallback(o2::framework::CompletionPolicy::InputSet inputs)
 {
   // fixme: we assume that there is one timer input and the rest are data inputs. If some other implicit inputs are
   //  added, this will break.
@@ -125,7 +126,7 @@ CompletionPolicy::CompletionOp TaskRunner::completionPolicyCallback(gsl::span<Pa
       continue;
     }
 
-    const auto* dataHeader = get<DataHeader*>(input.header.get()->GetData());
+    const auto* dataHeader = CompletionPolicyHelpers::getHeader<DataHeader>(input);
     assert(dataHeader);
 
     if (!strncmp(dataHeader->dataDescription.str, "TIMER", 5)) {

Framework/src/runAdvanced.cxx

@@ -34,6 +34,7 @@
 #include <Framework/CompletionPolicyHelpers.h>
 #include <Framework/DataSampling.h>
 #include <Framework/DataSpecUtils.h>
+#include <Framework/CompletionPolicyHelpers.h>
 #include "QualityControl/InfrastructureGenerator.h"
 
 using namespace o2;
@@ -48,16 +49,7 @@ void customize(std::vector<CompletionPolicy>& policies)
 {
   DataSampling::CustomizeInfrastructure(policies);
   quality_control::customizeInfrastructure(policies);
-  CompletionPolicy mergerConsumesASAP{
-    "mergers-always-consume",
-    [](DeviceSpec const& device) {
-      return device.name.find("merger") != std::string::npos;
-    },
-    [](gsl::span<PartRef const> const& /*inputs*/) {
-      return CompletionPolicy::CompletionOp::Consume;
-    }
-  };
-  policies.push_back(mergerConsumesASAP);
+  policies.push_back(CompletionPolicyHelpers::defineByName(".*merger.*", CompletionPolicy::CompletionOp::Consume));
 }
 
 void customize(std::vector<ChannelConfigurationPolicy>& policies)
@@ -169,4 +161,4 @@ WorkflowSpec defineDataProcessing(ConfigContext const&)
   quality_control::generateRemoteInfrastructure(specs, qcConfigurationSource);
 
   return specs;
-}
+}