Merge b33b8bc into sapling-pr-archive-ktf

Author: ktf

DataFormats/simulation/src/DigitizationContext.cxx

@@ -577,7 +577,7 @@ void DigitizationContext::applyMaxCollisionFilter(std::vector<std::tuple<int, in
     if (indices_old_to_new.find(lastindex) != indices_old_to_new.end()) {
       std::get<1>(tf_indices) = indices_old_to_new[lastindex]; // end;
     } else {
-      std::get<1>(tf_indices) = newrecords.size(); // end;
+      std::get<1>(tf_indices) = newrecords.size() - 1; // end; -1 since index inclusif
     }
     if (indices_old_to_new.find(previndex) != indices_old_to_new.end()) {
       std::get<2>(tf_indices) = indices_old_to_new[previndex]; // previous or "early" index
@@ -591,11 +591,6 @@ void DigitizationContext::applyMaxCollisionFilter(std::vector<std::tuple<int, in
 std::vector<std::tuple<int, int, int>> DigitizationContext::calcTimeframeIndices(long startOrbit, long orbitsPerTF, double orbitsEarly) const
 {
   auto timeframeindices = getTimeFrameBoundaries(mEventRecords, startOrbit, orbitsPerTF, orbitsEarly);
-  LOG(info) << "Fixed " << timeframeindices.size() << " timeframes ";
-  for (auto p : timeframeindices) {
-    LOG(info) << std::get<0>(p) << " " << std::get<1>(p) << " " << std::get<2>(p);
-  }
-
   return timeframeindices;
 }
 
@@ -708,7 +703,7 @@ DigitizationContext DigitizationContext::extractSingleTimeframe(int timeframeid,
     auto tf_ranges = timeframeindices.at(timeframeid);
 
     auto startindex = std::get<0>(tf_ranges);
-    auto endindex = std::get<1>(tf_ranges) + 1; // +1 due to endindex being "including"
+    auto endindex = std::get<1>(tf_ranges) + 1;
     auto earlyindex = std::get<2>(tf_ranges);
 
     if (earlyindex >= 0) {

Framework/Core/CMakeLists.txt

@@ -240,6 +240,7 @@ add_executable(o2-test-framework-core
               test/test_OverrideLabels.cxx
               test/test_O2DataModelHelpers.cxx
               test/test_RootConfigParamHelpers.cxx
+              test/test_ResourcesMonitoringHelpers.cxx
               test/test_Services.cxx
               test/test_StringHelpers.cxx
               test/test_StaticFor.cxx

Framework/Core/include/Framework/DataProcessingHelpers.h

@@ -21,6 +21,10 @@ struct ForwardChannelState;
 struct OutputChannelInfo;
 struct OutputChannelSpec;
 struct OutputChannelState;
+struct ProcessingPolicies;
+struct DeviceSpec;
+enum struct StreamingState;
+enum struct TransitionHandlingState;
 
 /// Generic helpers for DataProcessing releated functions.
 struct DataProcessingHelpers {
@@ -35,6 +39,12 @@ struct DataProcessingHelpers {
   static bool sendOldestPossibleTimeframe(ServiceRegistryRef const& ref, OutputChannelInfo const& info, OutputChannelState& state, size_t timeslice);
   /// Broadcast the oldest possible timeslice to all channels in output
   static void broadcastOldestPossibleTimeslice(ServiceRegistryRef const& ref, size_t timeslice);
+  /// change the device StreamingState to newState
+  static void switchState(ServiceRegistryRef const& ref, StreamingState newState);
+  /// check if spec is a source devide
+  static bool hasOnlyGenerated(DeviceSpec const& spec);
+  /// starts the EoS timers and returns the new TransitionHandlingState in case as new state is requested
+  static TransitionHandlingState updateStateTransition(ServiceRegistryRef const& ref, ProcessingPolicies const& policies);
 };
 
 } // namespace o2::framework

Framework/Core/include/Framework/DriverInfo.h

@@ -149,6 +149,8 @@ struct DriverInfo {
   std::string uniqueWorkflowId = "";
   /// Metrics gathering interval
   unsigned short resourcesMonitoringInterval = 0;
+  /// Where to dump the metrics
+  std::string resourcesMonitoringFilename = "performanceMetrics.json";
   /// Metrics gathering dump to disk interval
   unsigned short resourcesMonitoringDumpInterval = 0;
   /// Port used by the websocket control. 0 means not initialised.

Framework/Core/src/DataProcessingDevice.cxx

@@ -125,63 +125,6 @@ void on_idle_timer(uv_timer_t* handle)
   state->loopReason |= DeviceState::TIMER_EXPIRED;
 }
 
-bool hasOnlyTimers(DeviceSpec const& spec)
-{
-  return std::all_of(spec.inputs.cbegin(), spec.inputs.cend(), [](InputRoute const& route) -> bool { return route.matcher.lifetime == Lifetime::Timer; });
-}
-
-bool hasOnlyGenerated(DeviceSpec const& spec)
-{
-  return (spec.inputChannels.size() == 1) && (spec.inputs[0].matcher.lifetime == Lifetime::Timer || spec.inputs[0].matcher.lifetime == Lifetime::Enumeration);
-}
-
-void on_transition_requested_expired(uv_timer_t* handle)
-{
-  auto* ref = (ServiceRegistryRef*)handle->data;
-  auto& state = ref->get<DeviceState>();
-  state.loopReason |= DeviceState::TIMER_EXPIRED;
-  // Check if this is a source device
-  O2_SIGNPOST_ID_FROM_POINTER(cid, device, handle);
-  auto& spec = ref->get<DeviceSpec const>();
-  std::string messageOnExpire = hasOnlyGenerated(spec) ? "DPL exit transition grace period for source expired. Exiting." : fmt::format("DPL exit transition grace period for {} expired. Exiting.", state.allowedProcessing == DeviceState::CalibrationOnly ? "calibration" : "data & calibration").c_str();
-  if (!ref->get<RawDeviceService>().device()->GetConfig()->GetValue<bool>("error-on-exit-transition-timeout")) {
-    O2_SIGNPOST_EVENT_EMIT_WARN(calibration, cid, "callback", "%{public}s", messageOnExpire.c_str());
-  } else {
-    O2_SIGNPOST_EVENT_EMIT_ERROR(calibration, cid, "callback", "%{public}s", messageOnExpire.c_str());
-  }
-  state.transitionHandling = TransitionHandlingState::Expired;
-}
-
-auto switchState(ServiceRegistryRef& ref, StreamingState newState) -> void
-{
-  auto& state = ref.get<DeviceState>();
-  auto& context = ref.get<DataProcessorContext>();
-  O2_SIGNPOST_ID_FROM_POINTER(dpid, device, &context);
-  O2_SIGNPOST_END(device, dpid, "state", "End of processing state %d", (int)state.streaming);
-  O2_SIGNPOST_START(device, dpid, "state", "Starting processing state %d", (int)newState);
-  state.streaming = newState;
-  ref.get<ControlService>().notifyStreamingState(state.streaming);
-};
-
-void on_data_processing_expired(uv_timer_t* handle)
-{
-  auto* ref = (ServiceRegistryRef*)handle->data;
-  auto& state = ref->get<DeviceState>();
-  auto& spec = ref->get<DeviceSpec const>();
-  state.loopReason |= DeviceState::TIMER_EXPIRED;
-
-  // Check if this is a source device
-  O2_SIGNPOST_ID_FROM_POINTER(cid, device, handle);
-
-  if (hasOnlyGenerated(spec)) {
-    O2_SIGNPOST_EVENT_EMIT_INFO(calibration, cid, "callback", "Grace period for data processing expired. Switching to EndOfStreaming.");
-    switchState(*ref, StreamingState::EndOfStreaming);
-  } else {
-    O2_SIGNPOST_EVENT_EMIT_INFO(calibration, cid, "callback", "Grace period for data processing expired. Only calibrations from this point onwards.");
-    state.allowedProcessing = DeviceState::CalibrationOnly;
-  }
-}
-
 void on_communication_requested(uv_async_t* s)
 {
   auto* state = (DeviceState*)s->data;
@@ -1267,7 +1210,7 @@ void DataProcessingDevice::PreRun()
   O2_SIGNPOST_ID_FROM_POINTER(cid, device, state.loop);
   O2_SIGNPOST_START(device, cid, "PreRun", "Entering PreRun callback.");
   state.quitRequested = false;
-  switchState(ref, StreamingState::Streaming);
+  DataProcessingHelpers::switchState(ref, StreamingState::Streaming);
   state.allowedProcessing = DeviceState::Any;
   for (auto& info : state.inputChannelInfos) {
     if (info.state != InputChannelState::Pull) {
@@ -1338,58 +1281,6 @@ void DataProcessingDevice::Reset()
   ref.get<CallbackService>().call<CallbackService::Id::Reset>();
 }
 
-TransitionHandlingState updateStateTransition(ServiceRegistryRef& ref, ProcessingPolicies const& policies)
-{
-  auto& state = ref.get<DeviceState>();
-  auto& deviceProxy = ref.get<FairMQDeviceProxy>();
-  if (state.transitionHandling != TransitionHandlingState::NoTransition || deviceProxy.newStateRequested() == false) {
-    return state.transitionHandling;
-  }
-  O2_SIGNPOST_ID_FROM_POINTER(lid, device, state.loop);
-  auto& deviceContext = ref.get<DeviceContext>();
-  // Check if we only have timers
-  auto& spec = ref.get<DeviceSpec const>();
-  if (hasOnlyTimers(spec)) {
-    switchState(ref, StreamingState::EndOfStreaming);
-  }
-
-  // We do not do anything in particular if the data processing timeout would go past the exitTransitionTimeout
-  if (deviceContext.dataProcessingTimeout > 0 && deviceContext.dataProcessingTimeout < deviceContext.exitTransitionTimeout) {
-    uv_update_time(state.loop);
-    O2_SIGNPOST_EVENT_EMIT(calibration, lid, "timer_setup", "Starting %d s timer for dataProcessingTimeout.", deviceContext.dataProcessingTimeout);
-    uv_timer_start(deviceContext.dataProcessingGracePeriodTimer, on_data_processing_expired, deviceContext.dataProcessingTimeout * 1000, 0);
-  }
-  if (deviceContext.exitTransitionTimeout != 0 && state.streaming != StreamingState::Idle) {
-    ref.get<CallbackService>().call<CallbackService::Id::ExitRequested>(ServiceRegistryRef{ref});
-    uv_update_time(state.loop);
-    O2_SIGNPOST_EVENT_EMIT(calibration, lid, "timer_setup", "Starting %d s timer for exitTransitionTimeout.",
-                           deviceContext.exitTransitionTimeout);
-    uv_timer_start(deviceContext.gracePeriodTimer, on_transition_requested_expired, deviceContext.exitTransitionTimeout * 1000, 0);
-    bool onlyGenerated = hasOnlyGenerated(spec);
-    int timeout = onlyGenerated ? deviceContext.dataProcessingTimeout : deviceContext.exitTransitionTimeout;
-    if (policies.termination == TerminationPolicy::QUIT && DefaultsHelpers::onlineDeploymentMode() == false) {
-      O2_SIGNPOST_EVENT_EMIT_INFO(device, lid, "run_loop", "New state requested. Waiting for %d seconds before quitting.", timeout);
-    } else {
-      O2_SIGNPOST_EVENT_EMIT_INFO(device, lid, "run_loop",
-                                  "New state requested. Waiting for %d seconds before %{public}s",
-                                  timeout,
-                                  onlyGenerated ? "dropping remaining input and switching to READY state." : "switching to READY state.");
-    }
-    return TransitionHandlingState::Requested;
-  } else {
-    if (deviceContext.exitTransitionTimeout == 0 && policies.termination == TerminationPolicy::QUIT) {
-      O2_SIGNPOST_EVENT_EMIT_INFO(device, lid, "run_loop", "New state requested. No timeout set, quitting immediately as per --completion-policy");
-    } else if (deviceContext.exitTransitionTimeout == 0 && policies.termination != TerminationPolicy::QUIT) {
-      O2_SIGNPOST_EVENT_EMIT_INFO(device, lid, "run_loop", "New state requested. No timeout set, switching to READY state immediately");
-    } else if (policies.termination == TerminationPolicy::QUIT) {
-      O2_SIGNPOST_EVENT_EMIT_INFO(device, lid, "run_loop", "New state pending and we are already idle, quitting immediately as per --completion-policy");
-    } else {
-      O2_SIGNPOST_EVENT_EMIT_INFO(device, lid, "run_loop", "New state pending and we are already idle, switching to READY immediately.");
-    }
-    return TransitionHandlingState::Expired;
-  }
-}
-
 void DataProcessingDevice::Run()
 {
   ServiceRegistryRef ref{mServiceRegistry};
@@ -1442,7 +1333,7 @@ void DataProcessingDevice::Run()
         shouldNotWait = true;
         state.loopReason |= DeviceState::LoopReason::NEW_STATE_PENDING;
       }
-      state.transitionHandling = updateStateTransition(ref, ref.get<DeviceContext>().processingPolicies);
+      state.transitionHandling = DataProcessingHelpers::updateStateTransition(ref, ref.get<DeviceContext>().processingPolicies);
       // If we are Idle, we can then consider the transition to be expired.
       if (state.transitionHandling == TransitionHandlingState::Requested && state.streaming == StreamingState::Idle) {
         O2_SIGNPOST_EVENT_EMIT(device, lid, "run_loop", "State transition requested and we are now in Idle. We can consider it to be completed.");
@@ -1828,7 +1719,7 @@ void DataProcessingDevice::doRun(ServiceRegistryRef ref)
   // dependent on the callback, not something which is controlled by the
   // framework itself.
   if (context.allDone == true && state.streaming == StreamingState::Streaming) {
-    switchState(ref, StreamingState::EndOfStreaming);
+    DataProcessingHelpers::switchState(ref, StreamingState::EndOfStreaming);
     state.lastActiveDataProcessor = &context;
   }
 
@@ -1841,7 +1732,7 @@ void DataProcessingDevice::doRun(ServiceRegistryRef ref)
     /// timers as they do not need to be further processed.
     auto& relayer = ref.get<DataRelayer>();
 
-    bool shouldProcess = hasOnlyGenerated(spec) == false;
+    bool shouldProcess = DataProcessingHelpers::hasOnlyGenerated(spec) == false;
 
     while (DataProcessingDevice::tryDispatchComputation(ref, context.completed) && shouldProcess) {
       relayer.processDanglingInputs(context.expirationHandlers, *context.registry, false);
@@ -1874,7 +1765,7 @@ void DataProcessingDevice::doRun(ServiceRegistryRef ref)
     }
     // This is needed because the transport is deleted before the device.
     relayer.clear();
-    switchState(ref, StreamingState::Idle);
+    DataProcessingHelpers::switchState(ref, StreamingState::Idle);
     // In case  we should process, note the data processor responsible for it
     if (shouldProcess) {
       state.lastActiveDataProcessor = &context;
@@ -2567,7 +2458,7 @@ bool DataProcessingDevice::tryDispatchComputation(ServiceRegistryRef ref, std::v
           O2_SIGNPOST_EVENT_EMIT(device, pcid, "device", "Skipping processing because we are discarding.");
         } else {
           O2_SIGNPOST_EVENT_EMIT(device, pcid, "device", "No processing callback provided. Switching to %{public}s.", "Idle");
-          switchState(ref, StreamingState::Idle);
+          DataProcessingHelpers::switchState(ref, StreamingState::Idle);
         }
         if (shouldProcess(action)) {
           auto& timingInfo = ref.get<TimingInfo>();
@@ -2655,7 +2546,7 @@ bool DataProcessingDevice::tryDispatchComputation(ServiceRegistryRef ref, std::v
     for (auto& channel : spec.outputChannels) {
       DataProcessingHelpers::sendEndOfStream(ref, channel);
     }
-    switchState(ref, StreamingState::Idle);
+    DataProcessingHelpers::switchState(ref, StreamingState::Idle);
   }
 
   return true;