DPL will propagate DPH.creation time of 1st input slot

shahor02 · shahor02 · commit 28e32fd1f3ac · 2021-11-11T11:32:57.000+01:00
diff --git a/Framework/Core/include/Framework/DataDescriptorMatcher.h b/Framework/Core/include/Framework/DataDescriptorMatcher.h
@@ -47,7 +47,8 @@ enum ContextPos {
   STARTTIME_POS = 0,     /// The DataProcessingHeader::startTime associated to the timeslice
   TFCOUNTER_POS = 14,    /// The DataHeader::tfCounter associated to the timeslice
   FIRSTTFORBIT_POS = 15, /// The DataHeader::firstTFOrbit associated to the timeslice
-  RUNNUMBER_POS = 13     /// The DataHeader::runNumber associated to the timeslice
+  RUNNUMBER_POS = 13,    /// The DataHeader::runNumber associated to the timeslice
+  CREATIONTIME_POS = 12  /// The DataProcessingHeader::creation associated to the timeslice
 };
 
 /// An element of the matching context. Context itself is really a vector of
diff --git a/Framework/Core/include/Framework/DataRelayer.h b/Framework/Core/include/Framework/DataRelayer.h
@@ -137,6 +137,8 @@ class DataRelayer
   uint32_t getFirstTFCounterForSlot(TimesliceSlot slot);
   /// Get the runNumber associated to a given slot
   uint32_t getRunNumberForSlot(TimesliceSlot slot);
+  /// Get the creation time associated to a given slot
+  uint64_t getCreationTimeForSlot(TimesliceSlot slot);
   /// Remove all pending messages
   void clear();
 
diff --git a/Framework/Core/include/Framework/TimingInfo.h b/Framework/Core/include/Framework/TimingInfo.h
@@ -22,6 +22,7 @@ struct TimingInfo {
   uint32_t firstTFOrbit = -1; /// the orbit the TF begins
   uint32_t tfCounter = -1;    // the counter associated to a TF
   uint32_t runNumber = -1;
+  uint64_t creation = -1UL;
 };
 
 #endif // O2_FRAMEWORK_TIMINGINFO_H_
diff --git a/Framework/Core/include/Framework/VariableContextHelpers.h b/Framework/Core/include/Framework/VariableContextHelpers.h
@@ -58,6 +58,16 @@ struct VariableContextHelpers {
     }
     return *pval;
   }
+
+  static inline uint64_t getCreationTime(data_matcher::VariableContext const& variables)
+  {
+    // creation time is always at register 14
+    auto pval = std::get_if<uint64_t>(&variables.get(data_matcher::CREATIONTIME_POS));
+    if (pval == nullptr) {
+      return -1UL;
+    }
+    return *pval;
+  }
 };
 } // namespace o2::framework
 
diff --git a/Framework/Core/src/DataAllocator.cxx b/Framework/Core/src/DataAllocator.cxx
@@ -104,7 +104,7 @@ FairMQMessagePtr DataAllocator::headerMessageFromOutput(Output const& spec,
   dh.firstTForbit = timingInfo.firstTFOrbit;
   dh.runNumber = timingInfo.runNumber;
 
-  DataProcessingHeader dph{timingInfo.timeslice, 1};
+  DataProcessingHeader dph{timingInfo.timeslice, 1, timingInfo.creation};
   auto& context = mRegistry->get<MessageContext>();
 
   auto channelAlloc = o2::pmr::getTransportAllocator(context.proxy().getTransport(channel, 0));
diff --git a/Framework/Core/src/DataDescriptorMatcher.cxx b/Framework/Core/src/DataDescriptorMatcher.cxx
@@ -106,6 +106,8 @@ bool StartTimeValueMatcher::match(header::DataHeader const& dh, DataProcessingHe
       return (dph.startTime / mScale) == *value;
     }
     context.put({ref->index, dph.startTime / mScale});
+    // We always put in 12 the creation time
+    context.put({CREATIONTIME_POS, dph.creation});
     // We always put in 13 the runNumber
     context.put({RUNNUMBER_POS, dh.runNumber});
     // We always put in 14 the tfCounter
diff --git a/Framework/Core/src/DataProcessingDevice.cxx b/Framework/Core/src/DataProcessingDevice.cxx
@@ -1194,6 +1194,7 @@ bool DataProcessingDevice::tryDispatchComputation(DataProcessorContext& context,
     timingInfo->tfCounter = relayer->getFirstTFCounterForSlot(i);
     timingInfo->firstTFOrbit = relayer->getFirstTFOrbitForSlot(i);
     timingInfo->runNumber = relayer->getRunNumberForSlot(i);
+    timingInfo->creation = relayer->getCreationTimeForSlot(i);
   };
 
   // When processing them, timers will have to be cleaned up
diff --git a/Framework/Core/src/DataRelayer.cxx b/Framework/Core/src/DataRelayer.cxx
@@ -746,6 +746,12 @@ uint32_t DataRelayer::getRunNumberForSlot(TimesliceSlot slot)
   return VariableContextHelpers::getRunNumber(mTimesliceIndex.getVariablesForSlot(slot));
 }
 
+uint64_t DataRelayer::getCreationTimeForSlot(TimesliceSlot slot)
+{
+  std::scoped_lock<LockableBase(std::recursive_mutex)> lock(mMutex);
+  return VariableContextHelpers::getCreationTime(mTimesliceIndex.getVariablesForSlot(slot));
+}
+
 void DataRelayer::sendContextState()
 {
   std::scoped_lock<LockableBase(std::recursive_mutex)> lock(mMutex);

Original file line number	Diff line number	Diff line change
`@@ -106,6 +106,8 @@ bool StartTimeValueMatcher::match(header::DataHeader const& dh, DataProcessingHe`
`106`	`106`	`return (dph.startTime / mScale) == *value;`
`107`	`107`	`}`
`108`	`108`	`context.put({ref->index, dph.startTime / mScale});`
	`109`	`+ // We always put in 12 the creation time`
	`110`	`+ context.put({CREATIONTIME_POS, dph.creation});`
`109`	`111`	`// We always put in 13 the runNumber`
`110`	`112`	`context.put({RUNNUMBER_POS, dh.runNumber});`
`111`	`113`	`// We always put in 14 the tfCounter`
Original file line number	Diff line number	Diff line change
`@@ -746,6 +746,12 @@ uint32_t DataRelayer::getRunNumberForSlot(TimesliceSlot slot)`
`746`	`746`	`return VariableContextHelpers::getRunNumber(mTimesliceIndex.getVariablesForSlot(slot));`
`747`	`747`	`}`
`748`	`748`
	`749`	`+uint64_t DataRelayer::getCreationTimeForSlot(TimesliceSlot slot)`
	`750`	`+{`
	`751`	`+ std::scoped_lock<LockableBase(std::recursive_mutex)> lock(mMutex);`
	`752`	`+ return VariableContextHelpers::getCreationTime(mTimesliceIndex.getVariablesForSlot(slot));`
	`753`	`+}`
	`754`	`+`
`749`	`755`	`void DataRelayer::sendContextState()`
`750`	`756`	`{`
`751`	`757`	`std::scoped_lock<LockableBase(std::recursive_mutex)> lock(mMutex);`