DPL: move completed to be a stream variable

Author: ktf

Framework/Core/include/Framework/DataProcessingContext.h

@@ -44,7 +44,6 @@ struct DataProcessorContext {
 
   // FIXME: move stuff here from the list below... ;-)
   ServiceRegistry* registry = nullptr;
-  std::vector<DataRelayer::RecordAction> completed;
   std::vector<ExpirationHandler> expirationHandlers;
   AlgorithmSpec::InitCallback init;
   AlgorithmSpec::ProcessCallback statefulProcess;

Framework/Core/include/Framework/StreamContext.h

@@ -11,6 +11,7 @@
 #ifndef O2_FRAMEWORK_STREAMCONTEXT_H_
 #define O2_FRAMEWORK_STREAMCONTEXT_H_
 
+#include "Framework/DataRelayer.h"
 #include "Framework/ServiceHandle.h"
 #include "ProcessingContext.h"
 #include "ServiceSpec.h"
@@ -64,6 +65,10 @@ struct StreamContext {
   // the callback will be called for all of them.
   std::vector<ServiceStartStreamHandle> preStartStreamHandles;
 
+  /// Per-stream list of actions ready to be dispatched. Populated by
+  /// getReadyToProcess() and consumed by tryDispatchComputation().
+  std::vector<DataRelayer::RecordAction> completed;
+
   // Information on wether or not all the required routes have been created.
   // This is used to check if the LifetimeTimeframe routes were all created
   // for a given iteration.

Framework/Core/src/DataProcessingDevice.cxx

@@ -1644,6 +1644,7 @@ void DataProcessingDevice::doPrepare(ServiceRegistryRef ref)
 void DataProcessingDevice::doRun(ServiceRegistryRef ref)
 {
   auto& context = ref.get<DataProcessorContext>();
+  auto& streamContext = ref.get<StreamContext>();
   O2_SIGNPOST_ID_FROM_POINTER(dpid, device, &context);
   auto& state = ref.get<DeviceState>();
   auto& spec = ref.get<DeviceSpec const>();
@@ -1652,9 +1653,9 @@ void DataProcessingDevice::doRun(ServiceRegistryRef ref)
     return;
   }
 
-  context.completed.clear();
-  context.completed.reserve(16);
-  if (DataProcessingDevice::tryDispatchComputation(ref, context.completed)) {
+  streamContext.completed.clear();
+  streamContext.completed.reserve(16);
+  if (DataProcessingDevice::tryDispatchComputation(ref, streamContext.completed)) {
     state.lastActiveDataProcessor.store(&context);
   }
   DanglingContext danglingContext{*context.registry};
@@ -1668,8 +1669,8 @@ void DataProcessingDevice::doRun(ServiceRegistryRef ref)
     state.lastActiveDataProcessor = &context;
   }
 
-  context.completed.clear();
-  if (DataProcessingDevice::tryDispatchComputation(ref, context.completed)) {
+  streamContext.completed.clear();
+  if (DataProcessingDevice::tryDispatchComputation(ref, streamContext.completed)) {
     state.lastActiveDataProcessor = &context;
   }
 
@@ -1695,7 +1696,7 @@ void DataProcessingDevice::doRun(ServiceRegistryRef ref)
 
     bool shouldProcess = DataProcessingHelpers::hasOnlyGenerated(spec) == false;
 
-    while (DataProcessingDevice::tryDispatchComputation(ref, context.completed) && shouldProcess) {
+    while (DataProcessingDevice::tryDispatchComputation(ref, streamContext.completed) && shouldProcess) {
       relayer.processDanglingInputs(context.expirationHandlers, *context.registry, false);
     }