DPL: set TimingInfo field accordingly on new Run

ktf · ktf · commit 29dd2035db06 · 2022-11-21T19:36:52.000+01:00
diff --git a/Framework/Core/include/Framework/DataProcessingContext.h b/Framework/Core/include/Framework/DataProcessingContext.h
@@ -30,6 +30,8 @@ struct DataProcessorContext {
   // not shared by threads.
   bool* wasActive = nullptr;
   bool allDone = false;
+  /// Latest run number we processed globally for this DataProcessor.
+  int64_t lastRunNumberProcessed = 0;
 
   // These are pointers to the one owned by the DataProcessingDevice
   // but they are fully reentrant / thread safe and therefore can
diff --git a/Framework/Core/include/Framework/TimingInfo.h b/Framework/Core/include/Framework/TimingInfo.h
@@ -29,6 +29,13 @@ struct TimingInfo {
   uint32_t runNumber = -1;
   uint64_t creation = -1UL;
   uint64_t lapse = 0; // time at the start of the processing. Per thread.
+  /// Wether this TimingInfo refers to the first timeframe
+  /// from a new run.
+  bool globalRunNumberChanged = false;
+  /// Wether this TimingInfo refers to the first timeframe
+  /// from a new run, as being processed by the current stream.
+  /// FIXME: for now this is the same as the above.
+  bool streamRunNumberChanged = false;
 };
 
 } // namespace o2::framework
diff --git a/Framework/Core/src/DataProcessingDevice.cxx b/Framework/Core/src/DataProcessingDevice.cxx
@@ -1817,15 +1817,26 @@ bool DataProcessingDevice::tryDispatchComputation(ServiceRegistryRef ref, std::v
   // create messages) because the messages need to have the timeslice id into
   // it.
   auto prepareAllocatorForCurrentTimeSlice = [ref](TimesliceSlot i) -> void {
+    auto& dataProcessorContext = ref.get<DataProcessorContext>();
     auto& relayer = ref.get<DataRelayer>();
     auto& timingInfo = ref.get<TimingInfo>();
     ZoneScopedN("DataProcessingDevice::prepareForCurrentTimeslice");
     auto timeslice = relayer.getTimesliceForSlot(i);
+
     timingInfo.timeslice = timeslice.value;
     timingInfo.tfCounter = relayer.getFirstTFCounterForSlot(i);
     timingInfo.firstTForbit = relayer.getFirstTFOrbitForSlot(i);
     timingInfo.runNumber = relayer.getRunNumberForSlot(i);
     timingInfo.creation = relayer.getCreationTimeForSlot(i);
+    timingInfo.globalRunNumberChanged = dataProcessorContext.lastRunNumberProcessed <= timingInfo.runNumber;
+    // We report wether or not this timing info refers to a new Run.
+    if (timingInfo.globalRunNumberChanged) {
+      dataProcessorContext.lastRunNumberProcessed = timingInfo.runNumber;
+    }
+    // FIXME: for now there is only one stream, however we
+    //        should calculate this correctly once we finally get the
+    //        the StreamContext in.
+    timingInfo.streamRunNumberChanged = timingInfo.globalRunNumberChanged;
   };
 
   // When processing them, timers will have to be cleaned up
