run time watchdog (AliceO2Group#3927)

. stopd processing at time frame boundaries if expected run time for next time frame is expected to exceed a given run time limit
. run time limit is given in seconds by command line option time-limit

Author: pbuehler

Framework/Core/include/Framework/AODReaderHelpers.h

@@ -13,6 +13,7 @@
 
 #include "Framework/TableBuilder.h"
 #include "Framework/AlgorithmSpec.h"
+#include <uv.h>
 
 namespace o2
 {
@@ -21,6 +22,49 @@ namespace framework
 namespace readers
 {
 
+struct RuntimeWatchdog {
+  int numberTimeFrames;
+  uint64_t startTime;
+  uint64_t lastTime;
+  double runTime;
+  uint64_t runTimeLimit;
+
+  RuntimeWatchdog(Long64_t limit)
+  {
+    numberTimeFrames = -1;
+    startTime = uv_hrtime();
+    lastTime = startTime;
+    runTime = 0.;
+    runTimeLimit = limit;
+  }
+
+  bool update()
+  {
+    numberTimeFrames++;
+    if (runTimeLimit <= 0) {
+      return true;
+    }
+
+    auto nowTime = uv_hrtime();
+
+    // time spent to process the time frame
+    double time_spent = numberTimeFrames < 1 ? (double)(nowTime - lastTime) / 1.E9 : 0.;
+    runTime += time_spent;
+    lastTime = nowTime;
+
+    return ((double)(lastTime - startTime) / 1.E9 + runTime / (numberTimeFrames + 1)) < runTimeLimit;
+  }
+
+  void printOut()
+  {
+    LOGP(INFO, "RuntimeWatchdog");
+    LOGP(INFO, "  run time limit: {}", runTimeLimit);
+    LOGP(INFO, "  number of time frames: {}", numberTimeFrames);
+    LOGP(INFO, "  estimated run time per time frame: {}", (numberTimeFrames >= 0) ? runTime / (numberTimeFrames + 1) : 0.);
+    LOGP(INFO, "  estimated total run time: {}", (double)(lastTime - startTime) / 1.E9 + ((numberTimeFrames >= 0) ? runTime / (numberTimeFrames + 1) : 0.));
+  }
+};
+
 struct AODReaderHelpers {
   static AlgorithmSpec rootFileReaderCallback();
   static AlgorithmSpec aodSpawnerCallback(std::vector<InputSpec> requested);

Framework/Core/src/AODReaderHelpers.cxx

@@ -270,6 +270,9 @@ AlgorithmSpec AODReaderHelpers::rootFileReaderCallback()
       }
     }
 
+    // get the run time watchdog
+    auto* watchdog = new RuntimeWatchdog(options.get<int64_t>("time-limit"));
+
     // analyze type of requested tables
     uint64_t readMask = calculateReadMask(spec.outputs, header::DataOrigin{"AOD"});
     std::vector<OutputRoute> unknowns;
@@ -280,14 +283,24 @@ AlgorithmSpec AODReaderHelpers::rootFileReaderCallback()
     auto counter = std::make_shared<int>(0);
     return adaptStateless([readMask,
                            unknowns,
-                           counter,
+                           watchdog,
                            didir](DataAllocator& outputs, ControlService& control, DeviceSpec const& device) {
+      // check if RuntimeLimit is reached
+      if (!watchdog->update()) {
+        LOGP(INFO, "Run time exceeds run time limit of {} seconds!", watchdog->runTimeLimit);
+        LOGP(INFO, "Stopping after time frame {}.", watchdog->numberTimeFrames - 1);
+        didir->closeInputFiles();
+        control.endOfStream();
+        control.readyToQuit(QuitRequest::All);
+        return;
+      }
+
       // Each parallel reader reads the files whose index is associated to
       // their inputTimesliceId
       assert(device.inputTimesliceId < device.maxInputTimeslices);
-      size_t fi = (*counter * device.maxInputTimeslices) + device.inputTimesliceId;
-      *counter += 1;
+      size_t fi = (watchdog->numberTimeFrames * device.maxInputTimeslices) + device.inputTimesliceId;
 
+      // check if EoF is reached
       if (didir->atEnd(fi)) {
         LOGP(INFO, "All input files processed");
         didir->closeInputFiles();

Framework/Core/src/WorkflowHelpers.cxx

@@ -205,6 +205,7 @@ void WorkflowHelpers::injectServiceDevices(WorkflowSpec& workflow, ConfigContext
     readers::AODReaderHelpers::rootFileReaderCallback(),
     {ConfigParamSpec{"aod-file", VariantType::String, "aod.root", {"Input AOD file"}},
      ConfigParamSpec{"json-file", VariantType::String, {"json configuration file"}},
+     ConfigParamSpec{"time-limit", VariantType::Int64, 0ll, {"Maximum run time limit in seconds"}},
      ConfigParamSpec{"start-value-enumeration", VariantType::Int64, 0ll, {"initial value for the enumeration"}},
      ConfigParamSpec{"end-value-enumeration", VariantType::Int64, -1ll, {"final value for the enumeration"}},
      ConfigParamSpec{"step-value-enumeration", VariantType::Int64, 1ll, {"step between one value and the other"}}}};