Merge 5b5620d into sapling-pr-archive-ktf

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")