DPL Data Model: introduce bit to keep track of data created after EoS (#13315)

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?
-        flagsUnused : 31;           // currently unused
+        flagsReserved : 15,         // reserved for future use
+        flagsDerivedHeader : 16;    // reserved for usage by the derived header
     };
   };
 

Framework/Core/include/Framework/DataProcessingHeader.h

@@ -42,6 +42,9 @@ namespace o2::framework
 /// @ingroup aliceo2_dataformats_dataheader
 struct DataProcessingHeader : public header::BaseHeader {
   static constexpr uint64_t DUMMY_CREATION_TIME_OFFSET = 0x8000000000000000;
+  // The following flags are used to indicate the behavior of the data processing
+  static constexpr int32_t KEEP_AT_EOS_FLAG = 1;
+
   /// We return some number of milliseconds, offsetting int by 0x8000000000000000
   /// to make sure we can understand when the dummy constructor of DataProcessingHeader was
   /// used without overriding it with an actual real time from epoch.

Framework/Core/include/Framework/TimingInfo.h

@@ -23,7 +23,7 @@ namespace o2::framework
 
 struct TimingInfo {
   constexpr static ServiceKind service_kind = ServiceKind::Stream;
-  size_t timeslice; /// the timeslice associated to current processing
+  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
   uint32_t runNumber = -1;
@@ -36,9 +36,11 @@ struct TimingInfo {
   /// from a new run, as being processed by the current stream.
   /// FIXME: for now this is the same as the above.
   bool streamRunNumberChanged = false;
+  /// Wether this kind of data should be flushed during end of stream.
+  bool keepAtEndOfStream = false;
 
   static bool timesliceIsTimer(size_t timeslice) { return timeslice > 1652945069870351; }
-  bool isTimer() const { return timesliceIsTimer(timeslice); };
+  [[nodiscard]] bool isTimer() const { return timesliceIsTimer(timeslice); };
 };
 
 } // namespace o2::framework

Framework/Core/src/DataAllocator.cxx

@@ -121,6 +121,7 @@ fair::mq::MessagePtr DataAllocator::headerMessageFromOutput(Output const& spec,
   dh.runNumber = timingInfo.runNumber;
 
   DataProcessingHeader dph{timingInfo.timeslice, 1, timingInfo.creation};
+  static_cast<o2::header::BaseHeader&>(dph).flagsDerivedHeader |= timingInfo.keepAtEndOfStream ? DataProcessingHeader::KEEP_AT_EOS_FLAG : 0;
   auto& proxy = mRegistry.get<FairMQDeviceProxy>();
   auto* transport = proxy.getOutputTransport(routeIndex);
 

Framework/Core/src/DataProcessingDevice.cxx

@@ -1710,6 +1710,12 @@ void DataProcessingDevice::doRun(ServiceRegistryRef ref)
     while (DataProcessingDevice::tryDispatchComputation(ref, context.completed) && shouldProcess) {
       relayer.processDanglingInputs(context.expirationHandlers, *context.registry, false);
     }
+
+    auto& timingInfo = ref.get<TimingInfo>();
+    // We should keep the data generated at end of stream only for those
+    // which are not sources.
+    timingInfo.keepAtEndOfStream = shouldProcess;
+
     EndOfStreamContext eosContext{*context.registry, ref.get<DataAllocator>()};
 
     context.preEOSCallbacks(eosContext);