DPL: fix a number of inconsistencies in the forwarding logic

If a message came together with a DomainInfoHeader or a SourceInfoHeader,
it would have not been forwarded.

If one (header, payload, ...) tuple in a MessageSet was to be copied,
all the subsequent ones would have been copied.

If one (header, payload, ...) tuple got redirected to more than one destination,
all the subsequent ones would have been redirected there.

Author: ktf

DataFormats/Headers/include/Headers/DataHeader.h

@@ -373,7 +373,8 @@ struct BaseHeader {
     uint32_t flags;
     struct {
       uint32_t flagsNextHeader : 1, // do we have a next header after this one?
-        flagsReserved : 15,         // reserved for future use
+        flagsReserved : 14,         // reserved for future use. MUST be filled with 0s.
+        flagsDisabled : 1,          // header should be ignored if this is 1
         flagsDerivedHeader : 16;    // reserved for usage by the derived header
     };
   };
@@ -467,15 +468,20 @@ auto get(const std::byte* buffer, size_t /*len*/ = 0)
     // otherwise, we keep the code related to the exception outside the header file.
     // Note: Can not check on size because the O2 data model requires variable size headers
     // to be supported.
-    if (current->sanityCheck(HeaderValueType::sVersion)) {
+    if (current->sanityCheck(HeaderValueType::sVersion) && current->flagsDisabled == 0) {
+      // If the first header matches and it's enabled, we return it
+      // otherwise we look for more.
       return reinterpret_cast<HeaderConstPtrType>(current);
     }
   }
   auto* prev = current;
   while ((current = current->next())) {
     prev = current;
     if (current->description == HeaderValueType::sHeaderType) {
-      if (current->sanityCheck(HeaderValueType::sVersion)) {
+      // This is needed to allow disabling some headers from being picked up
+      // even if they are matching. This is handy to have a quick
+      // way to disable a sub headers without having to drop them.
+      if (current->sanityCheck(HeaderValueType::sVersion) && current->flagsDisabled == 0) {
         return reinterpret_cast<HeaderConstPtrType>(current);
       }
     }

Framework/Core/src/DataProcessingHelpers.cxx

@@ -221,136 +221,124 @@ 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 sih = o2::header::get<SourceInfoHeader*>(header);
-  if (sih) {
-    return false;
-  }
-
-  auto dih = o2::header::get<DomainInfoHeader*>(header);
-  if (dih) {
-    return false;
-  }
-
-  auto dh = o2::header::get<header::DataHeader*>(header);
-  if (!dh) {
-    return false;
-  }
-  auto dph = o2::header::get<DataProcessingHeader*>(header);
-  if (!dph) {
-    return false;
-  }
-  return true;
-};
-
-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, TimesliceSlot slot,
+                                                   std::vector<MessageSet>& currentSetOfInputs, TimesliceIndex::OldestOutputInfo oldestTimeslice,
+                                                   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" : "");
+                    slot.index, oldestTimeslice.timeslice.value, copyByDefault ? "with copy" : "", copyByDefault && 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();
 
-    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 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
+        auto dih = o2::header::get<DomainInfoHeader*>(header->GetData());
+        auto sih = o2::header::get<SourceInfoHeader*>(header->GetData());
+        if (dih == nullptr || sih == nullptr) {
+          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) {
+        forwardingChoices.clear();
+        proxy.getMatchingForwardChannelIndexes(forwardingChoices, *dh, dph->startTime);
+      }
+
+      if (forwardingChoices.empty()) {
+        // Nothing to forward go to the next messageset
+        continue;
       }
-      auto* dh = o2::header::get<header::DataHeader*>(header->GetData());
-      auto* dph = o2::header::get<DataProcessingHeader*>(header->GetData());
 
-      if (copy) {
-        for (auto& cachedForwardingChoice : cachedForwardingChoices) {
+      // 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()) {
+        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));
+          auto dih = o2::header::get<DomainInfoHeader*>(newHeader->GetData());
+          if (dih) {
+            const_cast<DomainInfoHeader*>(dih)->flagsDisabled = 1;
+          }
+          auto sih = o2::header::get<SourceInfoHeader*>(newHeader->GetData());
+          if (sih) {
+            const_cast<SourceInfoHeader*>(sih)->flagsDisabled = 1;
+          }
+          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);
+        auto dih = o2::header::get<DomainInfoHeader*>(messageSet.header(pi)->GetData());
+        auto sih = o2::header::get<SourceInfoHeader*>(messageSet.header(pi)->GetData());
+        // We need to copy the header if it has extra timeframe accounting
+        // information attached to it, so that we can disable it without having
+        // a race condition in shared memory.
+        if (dih || sih) {
+          auto&& newHeader = header->GetTransport()->CreateMessage();
+          newHeader->Copy(*header);
+          auto dih = o2::header::get<DomainInfoHeader*>(newHeader->GetData());
+          if (dih) {
+            const_cast<DomainInfoHeader*>(dih)->flagsDisabled = 1;
+          }
+          auto sih = o2::header::get<SourceInfoHeader*>(newHeader->GetData());
+          if (sih) {
+            const_cast<SourceInfoHeader*>(sih)->flagsDisabled = 1;
+          }
+          forwardedParts[forwardingChoices.back().value].AddPart(std::move(newHeader));
+        } else {
+          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)));
         }
       }
     }

Framework/Core/test/test_ForwardInputs.cxx

@@ -162,7 +162,7 @@ TEST_CASE("ForwardInputsSingleMessageSingleRouteAtEOS")
 
   auto result = o2::framework::DataProcessingHelpers::routeForwardedMessages(proxy, slot, currentSetOfInputs, oldestTimeslice, copyByDefault, consume);
   REQUIRE(result.size() == 1); // One route
-  REQUIRE(result[0].Size() == 0); // FIXME: this is an actual error. It should be 2
+  REQUIRE(result[0].Size() == 2);
   // Correct behavior below:
   // REQUIRE(result[0].Size() == 2);
   // REQUIRE(o2::header::get<SourceInfoHeader*>(result[0].At(0)->GetData()) == nullptr);
@@ -223,10 +223,8 @@ TEST_CASE("ForwardInputsSingleMessageSingleRouteWithOldestPossible")
 
   auto result = o2::framework::DataProcessingHelpers::routeForwardedMessages(proxy, slot, currentSetOfInputs, oldestTimeslice, copyByDefault, consume);
   REQUIRE(result.size() == 1); // One route
-  REQUIRE(result[0].Size() == 0); // FIXME: this is actually wrong
-  // FIXME: actually correct behavior below
-  // REQUIRE(result[0].Size() == 2);                                                     // Two messages
-  // REQUIRE(o2::header::get<DomainInfoHeader*>(result[0].At(0)->GetData()) == nullptr); // it should not have the end of stream
+  REQUIRE(result[0].Size() == 2);
+  REQUIRE(o2::header::get<DomainInfoHeader*>(result[0].At(0)->GetData()) == nullptr); // it should not have the end of stream
 }
 
 TEST_CASE("ForwardInputsSingleMessageMultipleRoutes")
@@ -590,8 +588,7 @@ TEST_CASE("ForwardInputsSplitPayload")
   auto result = o2::framework::DataProcessingHelpers::routeForwardedMessages(proxy, slot, currentSetOfInputs, oldestTimeslice, copyByDefault, consume);
   REQUIRE(result.size() == 2);    // Two routes
   CHECK(result[0].Size() == 2); // No messages on this route
-  CHECK(result[1].Size() == 5); // FIXME: Multipart matching has side effects also for the elements
-  // CHECK(result[1].Size() == 3); // FIXME: the correct forwarding is that only the multipart goes to the same route
+  CHECK(result[1].Size() == 3);
 }
 
 TEST_CASE("ForwardInputEOSSingleRoute")