DPL: earlier forwarding

DataFormats/Headers/include/Headers/DataHeaderHelpers.h

@@ -79,7 +79,8 @@ struct fmt::formatter<o2::header::DataHeader> {
                  fmt::format("  payloadSize  : {}\n", (long long unsigned int)h.payloadSize) +
                  fmt::format("  firstTForbit : {}\n", h.firstTForbit) +
                  fmt::format("  tfCounter    : {}\n", h.tfCounter) +
-                 fmt::format("  runNumber    : {}\n", h.runNumber);
+                 fmt::format("  runNumber    : {}\n", h.runNumber) +
+                 fmt::format("  split        : {}/{}\n", h.splitPayloadIndex, h.splitPayloadParts);
       return fmt::format_to(ctx.out(), "{}", res);
     } else {
       auto res = fmt::format("{}/{}/{}",

Framework/Core/include/Framework/DataProcessingHelpers.h

@@ -16,6 +16,7 @@
 #include "Framework/TimesliceIndex.h"
 #include <fairmq/FwdDecls.h>
 #include <vector>
+#include <span>
 
 namespace o2::framework
 {
@@ -53,8 +54,13 @@ struct DataProcessingHelpers {
   /// starts the EoS timers and returns the new TransitionHandlingState in case as new state is requested
   static TransitionHandlingState updateStateTransition(ServiceRegistryRef const& ref, ProcessingPolicies const& policies);
   /// Helper to route messages for forwarding
-  static std::vector<fair::mq::Parts> routeForwardedMessages(FairMQDeviceProxy& proxy, TimesliceSlot slot, std::vector<MessageSet>& currentSetOfInputs,
-                                                             TimesliceIndex::OldestOutputInfo oldestTimeslice, bool copy, bool consume);
+  static std::vector<fair::mq::Parts> routeForwardedMessageSet(FairMQDeviceProxy& proxy, std::vector<MessageSet>& currentSetOfInputs,
+                                                               bool copy, bool consume);
+  /// Helper to route messages for forwarding
+  static void routeForwardedMessages(FairMQDeviceProxy& proxy, std::span<fair::mq::MessagePtr>& currentSetOfInputs, std::vector<fair::mq::Parts>& forwardedParts,
+                                     bool copy, bool consume);
+  /// clean the headers when finally consuming a slot
+  static void cleanForwardedMessageSet(std::vector<MessageSet>& currentSetOfInputs);
 };
 } // namespace o2::framework
 #endif // O2_FRAMEWORK_DATAPROCESSINGHELPERS_H_

Framework/Core/include/Framework/DataRelayer.h

@@ -114,6 +114,9 @@ class DataRelayer
 
   using OnDropCallback = std::function<void(TimesliceSlot, std::vector<MessageSet>&, TimesliceIndex::OldestOutputInfo info)>;
 
+  // Callback for when some messages are about to be owned by the the DataRelayer
+  using OnInsertionCallback = std::function<void(ServiceRegistryRef&, std::span<fair::mq::MessagePtr>&)>;
+
   /// Prune all the pending entries in the cache.
   void prunePending(OnDropCallback);
   /// Prune the cache for a given slot
@@ -135,6 +138,7 @@ class DataRelayer
                     InputInfo const& info,
                     size_t nMessages,
                     size_t nPayloads = 1,
+                    OnInsertionCallback onInsertion = nullptr,
                     OnDropCallback onDrop = nullptr);
 
   /// This is to set the oldest possible @a timeslice this relayer can

Framework/Core/src/DataProcessingDevice.cxx

@@ -588,10 +588,12 @@ auto decongestionCallbackLate = [](AsyncTask& task, size_t aid) -> void {
 static auto forwardInputs = [](ServiceRegistryRef registry, TimesliceSlot slot, std::vector<MessageSet>& currentSetOfInputs,
                                TimesliceIndex::OldestOutputInfo oldestTimeslice, bool copy, bool consume = true) {
   auto& proxy = registry.get<FairMQDeviceProxy>();
-  auto forwardedParts = DataProcessingHelpers::routeForwardedMessages(proxy, slot, currentSetOfInputs, oldestTimeslice, copy, consume);
 
   O2_SIGNPOST_ID_GENERATE(sid, forwarding);
-  O2_SIGNPOST_EVENT_EMIT(forwarding, sid, "forwardInputs", "Forwarding %zu messages", forwardedParts.size());
+  O2_SIGNPOST_START(forwarding, sid, "forwardInputs", "Starting forwarding for slot %zu with oldestTimeslice %zu %{public}s%{public}s%{public}s",
+                    slot.index, oldestTimeslice.timeslice.value, copy ? "with copy" : "", copy && consume ? " and " : "", consume ? "with consume" : "");
+  auto forwardedParts = DataProcessingHelpers::routeForwardedMessageSet(proxy, currentSetOfInputs, copy, consume);
+
   for (int fi = 0; fi < proxy.getNumForwardChannels(); fi++) {
     if (forwardedParts[fi].Size() == 0) {
       continue;
@@ -1852,11 +1854,56 @@ void DataProcessingDevice::handleData(ServiceRegistryRef ref, InputChannelInfo&
             VariableContextHelpers::getTimeslice(variables);
             forwardInputs(ref, slot, dropped, oldestOutputInfo, false, true);
           };
+
+          auto onInsertion = [](ServiceRegistryRef& ref, std::span<fair::mq::MessagePtr>& messages) {
+            O2_SIGNPOST_ID_GENERATE(sid, forwarding);
+
+            auto& spec = ref.get<DeviceSpec const>();
+            bool hasForwards = spec.forwards.empty() == false;
+            auto& context = ref.get<DataProcessorContext>();
+            if (context.canForwardEarly && hasForwards) {
+              O2_SIGNPOST_EVENT_EMIT(device, sid, "device", "Early forwardinding before injecting data into relayer.");
+              auto& timesliceIndex = ref.get<TimesliceIndex>();
+              auto oldestTimeslice = timesliceIndex.getOldestPossibleOutput();
+
+              auto& proxy = ref.get<FairMQDeviceProxy>();
+
+              O2_SIGNPOST_ID_GENERATE(sid, forwarding);
+              O2_SIGNPOST_START(forwarding, sid, "forwardInputs", "Starting forwarding for incoming messages with oldestTimeslice %zu with copy",
+                                oldestTimeslice.timeslice.value);
+              std::vector<fair::mq::Parts> forwardedParts;
+              forwardedParts.resize(proxy.getNumForwards());
+              DataProcessingHelpers::routeForwardedMessages(proxy, messages, forwardedParts, true, false);
+
+              for (int fi = 0; fi < proxy.getNumForwardChannels(); fi++) {
+                if (forwardedParts[fi].Size() == 0) {
+                  continue;
+                }
+                ForwardChannelInfo info = proxy.getForwardChannelInfo(ChannelIndex{fi});
+                auto& parts = forwardedParts[fi];
+                if (info.policy == nullptr) {
+                  O2_SIGNPOST_EVENT_EMIT_ERROR(forwarding, sid, "forwardInputs", "Forwarding to %{public}s %d has no policy.", info.name.c_str(), fi);
+                  continue;
+                }
+                O2_SIGNPOST_EVENT_EMIT(forwarding, sid, "forwardInputs", "Forwarding to %{public}s %d", info.name.c_str(), fi);
+                info.policy->forward(parts, ChannelIndex{fi}, ref);
+              }
+              auto& asyncQueue = ref.get<AsyncQueue>();
+              auto& decongestion = ref.get<DecongestionService>();
+              O2_SIGNPOST_ID_GENERATE(aid, async_queue);
+              O2_SIGNPOST_EVENT_EMIT(async_queue, aid, "forwardInputs", "Queuing forwarding oldestPossible %zu", oldestTimeslice.timeslice.value);
+              AsyncQueueHelpers::post(asyncQueue, AsyncTask{.timeslice = oldestTimeslice.timeslice, .id = decongestion.oldestPossibleTimesliceTask, .debounce = -1, .callback = decongestionCallbackLate}
+                                                    .user<DecongestionContext>({.ref = ref, .oldestTimeslice = oldestTimeslice}));
+              O2_SIGNPOST_END(forwarding, sid, "forwardInputs", "Forwarding done");
+            }
+          };
+
           auto relayed = relayer.relay(parts.At(headerIndex)->GetData(),
                                        &parts.At(headerIndex),
                                        input,
                                        nMessages,
                                        nPayloadsPerHeader,
+                                       onInsertion,
                                        onDrop);
           switch (relayed.type) {
             case DataRelayer::RelayChoice::Type::Backpressured:
@@ -2271,9 +2318,10 @@ bool DataProcessingDevice::tryDispatchComputation(ServiceRegistryRef ref, std::v
     bool consumeSomething = action.op == CompletionPolicy::CompletionOp::Consume || action.op == CompletionPolicy::CompletionOp::ConsumeExisting;
 
     if (context.canForwardEarly && hasForwards && consumeSomething) {
-      O2_SIGNPOST_EVENT_EMIT(device, aid, "device", "Early forwainding: %{public}s.", fmt::format("{}", action.op).c_str());
-      auto& timesliceIndex = ref.get<TimesliceIndex>();
-      forwardInputs(ref, action.slot, currentSetOfInputs, timesliceIndex.getOldestPossibleOutput(), true, action.op == CompletionPolicy::CompletionOp::Consume);
+      // We used to do fowarding here, however we now do it much earlier.
+      // We still need to clean the inputs which were already consumed
+      // via ConsumeExisting and which still have an header to hold the slot.
+      DataProcessingHelpers::cleanForwardedMessageSet(currentSetOfInputs);
     }
     markInputsAsDone(action.slot);