DPL: make sure firstTFOrbit and tfCounter are set correctly (#5701)

In order to do so, we need to capture tfCounter in register 14 and firstTForbit
in register 15 when we capture the timeslice in register 0.

Author: ktf

Framework/Core/include/Framework/DataDescriptorMatcher.h

@@ -189,7 +189,7 @@ class StartTimeValueMatcher : public ValueHolder<uint64_t>
   /// This will match the timing information which is currently in
   /// the DataProcessingHeader. Notice how we apply the scale to the
   /// actual values found.
-  bool match(DataProcessingHeader const& dph, VariableContext& context) const;
+  bool match(header::DataHeader const& dh, DataProcessingHeader const& dph, VariableContext& context) const;
 
  private:
   uint64_t mScale;

Framework/Core/include/Framework/DataProcessingHeader.h

@@ -7,8 +7,8 @@
 // In applying this license CERN does not waive the privileges and immunities
 // granted to it by virtue of its status as an Intergovernmental Organization
 // or submit itself to any jurisdiction.
-#ifndef FRAMEWORK_DATAPROCESSINGHEADER_H
-#define FRAMEWORK_DATAPROCESSINGHEADER_H
+#ifndef O2_FRAMEWORK_DATAPROCESSINGHEADER_H_
+#define O2_FRAMEWORK_DATAPROCESSINGHEADER_H_
 
 #include "Headers/DataHeader.h"
 
@@ -17,9 +17,7 @@
 #include <cassert>
 #include <chrono>
 
-namespace o2
-{
-namespace framework
+namespace o2::framework
 {
 
 //__________________________________________________________________________________________________
@@ -98,7 +96,6 @@ struct DataProcessingHeader : public header::BaseHeader {
   }
 };
 
-} // namespace framework
-} // namespace o2
+} // namespace o2::framework
 
-#endif // FRAMEWORK_DATAPROCESSINGHEADER_H
+#endif // O2_FRAMEWORK_DATAPROCESSINGHEADER_H_

Framework/Core/include/Framework/DataRelayer.h

@@ -106,6 +106,11 @@ class DataRelayer
   /// Notice how this avoids exposing the timesliceIndex directly
   /// so that we can mutex on it.
   TimesliceId getTimesliceForSlot(TimesliceSlot slot);
+
+  /// Get the firstTFOrbit associate to a given slot.
+  uint32_t getFirstTFOrbitForSlot(TimesliceSlot slot);
+  /// Get the firstTFCounter associate to a given slot.
+  uint32_t getFirstTFCounterForSlot(TimesliceSlot slot);
   /// Remove all pending messages
   void clear();
 

Framework/Core/include/Framework/TimesliceIndex.h

@@ -74,6 +74,12 @@ class TimesliceIndex
   /// could correspond to different slots once we implement wildcards
   /// (e.g. if we ask for InputSpec{"*", "CLUSTERS"}).
   inline TimesliceId getTimesliceForSlot(TimesliceSlot slot) const;
+  /// Given @a slot, @return the firstTFOrbit (i.e. the variable at positiorn 15)
+  /// associated to it.
+  inline uint32_t getFirstTFOrbitForSlot(TimesliceSlot slot) const;
+  /// Given @a slot, @return the TFcounter (i.e. the variable at positiorn 14)
+  /// associated to it.
+  inline uint32_t getFirstTFCounterForSlot(TimesliceSlot slot) const;
   /// Given a slot, @return the VariableContext associated to it.
   /// This effectively means that the TimesliceIndex is now owner of the
   /// VariableContext.

Framework/Core/include/Framework/TimesliceIndex.inc

@@ -110,13 +110,36 @@ inline TimesliceSlot TimesliceIndex::findOldestSlot() const
 inline TimesliceId TimesliceIndex::getTimesliceForSlot(TimesliceSlot slot) const
 {
   assert(mVariables.size() > slot.index);
+  // timeslice is always at register 0
   auto pval = std::get_if<uint64_t>(&mVariables[slot.index].get(0));
   if (pval == nullptr) {
     return TimesliceId{TimesliceId::INVALID};
   }
   return TimesliceId{*pval};
 }
 
+inline uint32_t TimesliceIndex::getFirstTFOrbitForSlot(TimesliceSlot slot) const
+{
+  assert(mVariables.size() > slot.index);
+  // firstTForbit is always at register 15
+  auto pval = std::get_if<uint32_t>(&mVariables[slot.index].get(15));
+  if (pval == nullptr) {
+    return -1;
+  }
+  return *pval;
+}
+
+inline uint32_t TimesliceIndex::getFirstTFCounterForSlot(TimesliceSlot slot) const
+{
+  assert(mVariables.size() > slot.index);
+  // firstTForbit is always at register 15
+  auto pval = std::get_if<uint32_t>(&mVariables[slot.index].get(14));
+  if (pval == nullptr) {
+    return -1;
+  }
+  return *pval;
+}
+
 inline data_matcher::VariableContext& TimesliceIndex::getVariablesForSlot(TimesliceSlot slot)
 {
   assert(mVariables.size() > slot.index);

Framework/Core/include/Framework/TimingInfo.h

@@ -8,15 +8,18 @@
 // granted to it by virtue of its status as an Intergovernmental Organization
 // or submit itself to any jurisdiction.
 
-#ifndef FRAMEWORK_TIMINGINFO_H
-#define FRAMEWORK_TIMINGINFO_H
+#ifndef O2_FRAMEWORK_TIMINGINFO_H_
+#define O2_FRAMEWORK_TIMINGINFO_H_
 
 #include <cstddef>
+#include <cstdint>
 
 /// This class holds the information about timing
 /// of the messages being processed.
 struct TimingInfo {
   size_t timeslice; /// the timeslice associated to current processing
+  uint32_t firstTFOrbit = -1; /// the orbit the TF begins
+  uint32_t tfCounter = -1;    // the counter associated to a TF
 };
 
-#endif // Timing information for the current computation
+#endif // O2_FRAMEWORK_TIMINGINFO_H_

Framework/Core/src/DataAllocator.cxx

@@ -98,6 +98,8 @@ FairMQMessagePtr DataAllocator::headerMessageFromOutput(Output const& spec,
   dh.subSpecification = spec.subSpec;
   dh.payloadSize = payloadSize;
   dh.payloadSerializationMethod = method;
+  dh.tfCounter = mTimingInfo->tfCounter;
+  dh.firstTForbit = mTimingInfo->firstTFOrbit;
 
   DataProcessingHeader dph{mTimingInfo->timeslice, 1};
   auto& context = mRegistry->get<MessageContext>();

Framework/Core/src/DataDescriptorMatcher.cxx

@@ -16,11 +16,7 @@
 #include "Framework/RuntimeError.h"
 #include <iostream>
 
-namespace o2
-{
-namespace framework
-{
-namespace data_matcher
+namespace o2::framework::data_matcher
 {
 
 ContextElement::Value const& VariableContext::get(size_t pos) const
@@ -101,14 +97,18 @@ bool SubSpecificationTypeValueMatcher::match(header::DataHeader const& header, V
 /// This will match the timing information which is currently in
 /// the DataProcessingHeader. Notice how we apply the scale to the
 /// actual values found.
-bool StartTimeValueMatcher::match(DataProcessingHeader const& dph, VariableContext& context) const
+bool StartTimeValueMatcher::match(header::DataHeader const& dh, DataProcessingHeader const& dph, VariableContext& context) const
 {
   if (auto ref = std::get_if<ContextRef>(&mValue)) {
     auto& variable = context.get(ref->index);
     if (auto value = std::get_if<uint64_t>(&variable)) {
       return (dph.startTime / mScale) == *value;
     }
     context.put({ref->index, dph.startTime / mScale});
+    // We always put in 14 the tfCounter
+    context.put({14, dh.tfCounter});
+    // We always put in 15 the firstTForbit
+    context.put({15, dh.firstTForbit});
     return true;
   } else if (auto v = std::get_if<uint64_t>(&mValue)) {
     return (dph.startTime / mScale) == *v;
@@ -262,11 +262,12 @@ bool DataDescriptorMatcher::match(char const* d, VariableContext& context) const
   } else if (auto pval4 = std::get_if<ConstantValueMatcher>(&mLeft)) {
     leftValue = pval4->match();
   } else if (auto pval5 = std::get_if<StartTimeValueMatcher>(&mLeft)) {
+    auto dh = o2::header::get<header::DataHeader*>(d);
     auto dph = o2::header::get<DataProcessingHeader*>(d);
     if (dph == nullptr) {
       throw runtime_error("Cannot find DataProcessingHeader");
     }
-    leftValue = pval5->match(*dph, context);
+    leftValue = pval5->match(*dh, *dph, context);
   } else {
     throw runtime_error("Bad parsing tree");
   }
@@ -295,8 +296,9 @@ bool DataDescriptorMatcher::match(char const* d, VariableContext& context) const
   } else if (auto pval4 = std::get_if<ConstantValueMatcher>(&mRight)) {
     rightValue = pval4->match();
   } else if (auto pval5 = std::get_if<StartTimeValueMatcher>(&mRight)) {
+    auto dh = o2::header::get<header::DataHeader*>(d);
     auto dph = o2::header::get<DataProcessingHeader*>(d);
-    rightValue = pval5->match(*dph, context);
+    rightValue = pval5->match(*dh, *dph, context);
   }
   // There are cases in which not having a rightValue might be legitimate,
   // so we do not throw an exception.
@@ -478,6 +480,4 @@ std::ostream& operator<<(std::ostream& os, DataDescriptorMatcher::Op const& op)
   return os;
 }
 
-} // namespace data_matcher
-} // namespace framework
-} // namespace o2
+} // namespace o2::framework::data_matcher

Framework/Core/src/DataProcessingDevice.cxx

@@ -845,6 +845,8 @@ bool DataProcessingDevice::tryDispatchComputation(DataProcessorContext& context,
     ZoneScopedN("DataProcessingDevice::prepareForCurrentTimeslice");
     auto timeslice = relayer->getTimesliceForSlot(i);
     timingInfo->timeslice = timeslice.value;
+    timingInfo->tfCounter = relayer->getFirstTFCounterForSlot(i);
+    timingInfo->firstTFOrbit = relayer->getFirstTFOrbitForSlot(i);
   };
 
   // When processing them, timers will have to be cleaned up

Framework/Core/src/DataRelayer.cxx

@@ -602,6 +602,18 @@ DataRelayerStats const& DataRelayer::getStats() const
   return mStats;
 }
 
+uint32_t DataRelayer::getFirstTFOrbitForSlot(TimesliceSlot slot)
+{
+  std::scoped_lock<LockableBase(std::recursive_mutex)> lock(mMutex);
+  return mTimesliceIndex.getFirstTFOrbitForSlot(slot);
+}
+
+uint32_t DataRelayer::getFirstTFCounterForSlot(TimesliceSlot slot)
+{
+  std::scoped_lock<LockableBase(std::recursive_mutex)> lock(mMutex);
+  return mTimesliceIndex.getFirstTFCounterForSlot(slot);
+}
+
 void DataRelayer::sendContextState()
 {
   std::scoped_lock<LockableBase(std::recursive_mutex)> lock(mMutex);