DPL: more refactoring of the forwarding code

Use a single helper function to improve readability.

Author: ktf

Framework/Core/include/Framework/DataProcessingHelpers.h

@@ -53,8 +53,8 @@ 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> routeForwardedMessages(FairMQDeviceProxy& proxy, std::vector<MessageSet>& currentSetOfInputs,
+                                                             bool copy, bool consume);
 };
 } // namespace o2::framework
 #endif // O2_FRAMEWORK_DATAPROCESSINGHELPERS_H_

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::routeForwardedMessages(proxy, currentSetOfInputs, copy, consume);
+
   for (int fi = 0; fi < proxy.getNumForwardChannels(); fi++) {
     if (forwardedParts[fi].Size() == 0) {
       continue;

Framework/Core/src/DataProcessingHelpers.cxx

@@ -221,129 +221,99 @@ TransitionHandlingState DataProcessingHelpers::updateStateTransition(ServiceRegi
   }
 }
 
-static auto toBeForwardedHeader = [](void* header) -> bool {
-  // If is now possible that the record is not complete when
-  // we forward it, because of a custom completion policy.
-  // this means that we need to skip the empty entries in the
-  // record for being forwarded.
-  if (header == nullptr) {
-    return false;
-  }
-  auto dh = o2::header::get<header::DataHeader*>(header);
-  if (!dh) {
-    return false;
-  }
-  bool retval = !o2::header::get<SourceInfoHeader*>(header) &&
-                !o2::header::get<DomainInfoHeader*>(header) &&
-                o2::header::get<DataProcessingHeader*>(header);
-  // DataHeader is there. Complain if we have unexpected headers present / missing
-  if (!retval) {
-    LOGP(error, "Dropping data because of malformed header structure");
-  }
-  return retval;
-};
-
-static auto toBeforwardedMessageSet = [](std::vector<ChannelIndex>& cachedForwardingChoices,
-                                         FairMQDeviceProxy& proxy,
-                                         std::unique_ptr<fair::mq::Message>& header,
-                                         std::unique_ptr<fair::mq::Message>& payload,
-                                         size_t total,
-                                         bool consume) {
-  if (header.get() == nullptr) {
-    // Missing an header is not an error anymore.
-    // it simply means that we did not receive the
-    // given input, but we were asked to
-    // consume existing, so we skip it.
-    return false;
-  }
-  if (payload.get() == nullptr && consume == true) {
-    // If the payload is not there, it means we already
-    // processed it with ConsumeExisiting. Therefore we
-    // need to do something only if this is the last consume.
-    header.reset(nullptr);
-    return false;
-  }
-
-  auto fdph = o2::header::get<DataProcessingHeader*>(header->GetData());
-  if (fdph == nullptr) {
-    LOG(error) << "Data is missing DataProcessingHeader";
-    return false;
-  }
-  auto fdh = o2::header::get<header::DataHeader*>(header->GetData());
-  if (fdh == nullptr) {
-    LOG(error) << "Data is missing DataHeader";
-    return false;
-  }
-
-  // We need to find the forward route only for the first
-  // part of a split payload. All the others will use the same.
-  // but always check if we have a sequence of multiple payloads
-  if (fdh->splitPayloadIndex == 0 || fdh->splitPayloadParts <= 1 || total > 1) {
-    proxy.getMatchingForwardChannelIndexes(cachedForwardingChoices, *fdh, fdph->startTime);
-  }
-  return cachedForwardingChoices.empty() == false;
-};
-
-std::vector<fair::mq::Parts> DataProcessingHelpers::routeForwardedMessages(FairMQDeviceProxy& proxy, TimesliceSlot slot, std::vector<MessageSet>& currentSetOfInputs,
-                                                                           TimesliceIndex::OldestOutputInfo oldestTimeslice, bool copy, bool consume)
+auto DataProcessingHelpers::routeForwardedMessages(FairMQDeviceProxy& proxy,
+                                                   std::vector<MessageSet>& currentSetOfInputs,
+                                                   const bool copyByDefault, bool consume) -> std::vector<fair::mq::Parts>
 {
   // we collect all messages per forward in a map and send them together
   std::vector<fair::mq::Parts> forwardedParts;
   forwardedParts.resize(proxy.getNumForwards());
-  std::vector<ChannelIndex> cachedForwardingChoices{};
+  std::vector<ChannelIndex> forwardingChoices{};
   O2_SIGNPOST_ID_GENERATE(sid, forwarding);
-  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" : "");
 
   for (size_t ii = 0, ie = currentSetOfInputs.size(); ii < ie; ++ii) {
     auto& messageSet = currentSetOfInputs[ii];
-    // In case the messageSet is empty, there is nothing to be done.
-    if (messageSet.size() == 0) {
-      continue;
-    }
-    if (!toBeForwardedHeader(messageSet.header(0)->GetData())) {
-      continue;
-    }
-    cachedForwardingChoices.clear();
+    forwardingChoices.clear();
 
-    for (size_t pi = 0; pi < currentSetOfInputs[ii].size(); ++pi) {
-      auto& messageSet = currentSetOfInputs[ii];
+    for (size_t pi = 0; pi < messageSet.size(); ++pi) {
       auto& header = messageSet.header(pi);
+
+      // If is now possible that the record is not complete when
+      // we forward it, because of a custom completion policy.
+      // this means that we need to skip the empty entries in the
+      // record for being forwarded.
+      if (header->GetData() == nullptr) {
+        continue;
+      }
+      auto dih = o2::header::get<DomainInfoHeader*>(header->GetData());
+      if (dih) {
+        continue;
+      }
+      auto sih = o2::header::get<SourceInfoHeader*>(header->GetData());
+      if (sih) {
+        continue;
+      }
+
+      auto dph = o2::header::get<DataProcessingHeader*>(header->GetData());
+      auto dh = o2::header::get<o2::header::DataHeader*>(header->GetData());
+
+      if (dph == nullptr || dh == nullptr) {
+        // Complain only if this is not an out-of-band message
+        LOGP(error, "Data is missing {}{}{}",
+             dph ? "DataProcessingHeader" : "", dph || dh ? "and" : "", dh ? "DataHeader" : "");
+        continue;
+      }
+
       auto& payload = messageSet.payload(pi);
-      auto total = messageSet.getNumberOfPayloads(pi);
 
-      if (!toBeforwardedMessageSet(cachedForwardingChoices, proxy, header, payload, total, consume)) {
+      if (payload.get() == nullptr && consume == true) {
+        // If the payload is not there, it means we already
+        // processed it with ConsumeExisiting. Therefore we
+        // need to do something only if this is the last consume.
+        header.reset(nullptr);
         continue;
       }
 
-      // In case of more than one forward route, we need to copy the message.
-      // This will eventually use the same mamory if running with the same backend.
-      if (cachedForwardingChoices.size() > 1) {
-        copy = true;
+      // We need to find the forward route only for the first
+      // part of a split payload. All the others will use the same.
+      // Therefore, we reset and recompute the forwarding choice:
+      //
+      // - If this is the first payload of a [header0][payload0][header0][payload1] sequence,
+      //   which is actually always created and handled together
+      // - If the message is not a multipart (splitPayloadParts 0) or has only one part
+      // - If it's a message of the kind [header0][payload1][payload2][payload3]... and therefore
+      //   we will already use the same choice in the for loop below.
+      if (dh->splitPayloadIndex == 0 || dh->splitPayloadParts <= 1 || messageSet.getNumberOfPayloads(pi) > 0) {
+        proxy.getMatchingForwardChannelIndexes(forwardingChoices, *dh, dph->startTime);
       }
-      auto* dh = o2::header::get<header::DataHeader*>(header->GetData());
-      auto* dph = o2::header::get<DataProcessingHeader*>(header->GetData());
 
-      if (copy) {
-        for (auto& cachedForwardingChoice : cachedForwardingChoices) {
+      if (forwardingChoices.empty()) {
+        // Nothing to forward go to the next messageset
+        continue;
+      }
+
+      // In case of more than one forward route, we need to copy the message.
+      // This will eventually use the same memory if running with the same backend.
+      if (copyByDefault || forwardingChoices.size() > 1) {
+        for (auto& choice : forwardingChoices) {
           auto&& newHeader = header->GetTransport()->CreateMessage();
           O2_SIGNPOST_EVENT_EMIT(forwarding, sid, "forwardInputs", "Forwarding a copy of %{public}s to route %d.",
-                                 fmt::format("{}/{}/{}@timeslice:{} tfCounter:{}", dh->dataOrigin, dh->dataDescription, dh->subSpecification, dph->startTime, dh->tfCounter).c_str(), cachedForwardingChoice.value);
+                                 fmt::format("{}/{}/{}@timeslice:{} tfCounter:{}", dh->dataOrigin, dh->dataDescription, dh->subSpecification, dph->startTime, dh->tfCounter).c_str(), choice.value);
           newHeader->Copy(*header);
-          forwardedParts[cachedForwardingChoice.value].AddPart(std::move(newHeader));
+          forwardedParts[choice.value].AddPart(std::move(newHeader));
 
           for (size_t payloadIndex = 0; payloadIndex < messageSet.getNumberOfPayloads(pi); ++payloadIndex) {
             auto&& newPayload = header->GetTransport()->CreateMessage();
             newPayload->Copy(*messageSet.payload(pi, payloadIndex));
-            forwardedParts[cachedForwardingChoice.value].AddPart(std::move(newPayload));
+            forwardedParts[choice.value].AddPart(std::move(newPayload));
           }
         }
       } else {
         O2_SIGNPOST_EVENT_EMIT(forwarding, sid, "forwardInputs", "Forwarding %{public}s to route %d.",
-                               fmt::format("{}/{}/{}@timeslice:{} tfCounter:{}", dh->dataOrigin, dh->dataDescription, dh->subSpecification, dph->startTime, dh->tfCounter).c_str(), cachedForwardingChoices.back().value);
-        forwardedParts[cachedForwardingChoices.back().value].AddPart(std::move(messageSet.header(pi)));
+                               fmt::format("{}/{}/{}@timeslice:{} tfCounter:{}", dh->dataOrigin, dh->dataDescription, dh->subSpecification, dph->startTime, dh->tfCounter).c_str(), forwardingChoices.back().value);
+        forwardedParts[forwardingChoices.back().value].AddPart(std::move(messageSet.header(pi)));
         for (size_t payloadIndex = 0; payloadIndex < messageSet.getNumberOfPayloads(pi); ++payloadIndex) {
-          forwardedParts[cachedForwardingChoices.back().value].AddPart(std::move(messageSet.payload(pi, payloadIndex)));
+          forwardedParts[forwardingChoices.back().value].AddPart(std::move(messageSet.payload(pi, payloadIndex)));
         }
       }
     }