DPL Analysis: add option to artificially slow down AOD reader (#13549)

Author: ktf

Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx

@@ -41,8 +41,6 @@
 #include <arrow/table.h>
 #include <arrow/util/key_value_metadata.h>
 
-#include <thread>
-
 using namespace o2;
 using namespace o2::aod;
 
@@ -140,6 +138,8 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback()
 
     auto filename = options.get<std::string>("aod-file-private");
 
+    auto maxRate = options.get<float>("aod-max-io-rate");
+
     std::string parentFileReplacement;
     if (options.isSet("aod-parent-base-path-replacement")) {
       parentFileReplacement = options.get<std::string>("aod-parent-base-path-replacement");
@@ -192,6 +192,7 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback()
                            fileCounter,
                            numTF,
                            watchdog,
+                           maxRate,
                            didir, reportTFN, reportTFFileName](Monitoring& monitoring, DataAllocator& outputs, ControlService& control, DeviceSpec const& device) {
       // Each parallel reader device.inputTimesliceId reads the files fileCounter*device.maxInputTimeslices+device.inputTimesliceId
       // the TF to read is numTF
@@ -222,6 +223,8 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback()
         return;
       }
 
+      int64_t startTime = uv_hrtime();
+      int64_t startSize = totalSizeCompressed;
       for (auto& route : requestedTables) {
         if ((device.inputTimesliceId % route.maxTimeslices) != route.timeslice) {
           continue;
@@ -278,6 +281,18 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback()
         }
         first = false;
       }
+      int64_t stopSize = totalSizeCompressed;
+      int64_t bytesDelta = stopSize - startSize;
+      int64_t stopTime = uv_hrtime();
+      float currentDelta = float(stopTime - startTime) / 1000000000; // in s
+      if (ceil(maxRate) > 0.) {
+        float extraTime = (bytesDelta / 1000000 - currentDelta * maxRate) / maxRate;
+        // We only sleep if we read faster than the max-read-rate.
+        if (extraTime > 0.) {
+          LOGP(info, "Read {} MB in {} s. Sleeping for {} seconds to stay within {} MB/s limit.", bytesDelta / 1000000, currentDelta, extraTime, maxRate);
+          uv_sleep(extraTime * 1000); // in milliseconds
+        }
+      }
       totalDFSent++;
       monitoring.send(Metric{(uint64_t)totalDFSent, "df-sent"}.addTag(Key::Subsystem, monitoring::tags::Value::DPL));
       monitoring.send(Metric{(uint64_t)totalSizeUncompressed / 1000, "aod-bytes-read-uncompressed"}.addTag(Key::Subsystem, monitoring::tags::Value::DPL));

Framework/Core/src/WorkflowHelpers.cxx

@@ -12,6 +12,7 @@
 #include "AnalysisSupportHelpers.h"
 #include "Framework/AlgorithmSpec.h"
 #include "Framework/AODReaderHelpers.h"
+#include "Framework/ConfigParamSpec.h"
 #include "Framework/ConfigParamsHelper.h"
 #include "Framework/CommonDataProcessors.h"
 #include "Framework/ConfigContext.h"
@@ -27,14 +28,14 @@
 #include "Framework/DefaultsHelpers.h"
 #include "Framework/Signpost.h"
 
+#include "Framework/Variant.h"
 #include "Headers/DataHeader.h"
 #include <algorithm>
 #include <list>
 #include <set>
 #include <utility>
 #include <vector>
 #include <climits>
-#include <thread>
 
 O2_DECLARE_DYNAMIC_LOG(workflow_helpers);
 
@@ -213,6 +214,7 @@ void WorkflowHelpers::injectServiceDevices(WorkflowSpec& workflow, ConfigContext
                          aodLifetime}},
     .algorithm = AlgorithmSpec::dummyAlgorithm(),
     .options = {ConfigParamSpec{"aod-file-private", VariantType::String, ctx.options().get<std::string>("aod-file"), {"AOD file"}},
+                ConfigParamSpec{"aod-max-io-rate", VariantType::Float, 0.f, {"Maximum I/O rate in MB/s"}},
                 ConfigParamSpec{"aod-reader-json", VariantType::String, {"json configuration file"}},
                 ConfigParamSpec{"aod-parent-access-level", VariantType::String, {"Allow parent file access up to specified level. Default: no (0)"}},
                 ConfigParamSpec{"aod-parent-base-path-replacement", VariantType::String, {R"(Replace base path of parent files. Syntax: FROM;TO. E.g. "alien:///path/in/alien;/local/path". Enclose in "" on the command line.)"}},

Framework/Core/src/runDataProcessing.cxx

@@ -2026,6 +2026,7 @@ int runStateMachine(DataProcessorSpecs const& workflow,
             "--aod-writer-resmode",
             "--aod-writer-maxfilesize",
             "--aod-writer-keep",
+            "--aod-max-io-rate",
             "--aod-parent-access-level",
             "--aod-parent-base-path-replacement",
             "--driver-client-backend",