DPL Analysis: add option to artificially slow down AOD reader

ktf · Sep 30, 2024 · 56d0945 · 56d0945
1 parent 1de0de3
commit 56d0945
Show file tree

Hide file tree

Showing 3 changed files with 12 additions and 3 deletions.
diff --git a/Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx b/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,7 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback()
         return;
       }
 
+      int64_t startTime = uv_hrtime();
       for (auto& route : requestedTables) {
         if ((device.inputTimesliceId % route.maxTimeslices) != route.timeslice) {
           continue;
@@ -278,6 +280,13 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback()
         }
         first = false;
       }
+      int64_t stopTime = uv_hrtime();
+      int64_t currentDelta = (stopTime - startTime) / 1000000000; // in s
+      float extraTime = (totalSizeUncompressed / 1000000 - currentDelta * maxRate) / maxRate;
+      // We only sleep if we read faster than the max-read-rate.
+      if (extraTime > 0.) {
+        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));

diff --git a/Framework/Core/src/WorkflowCustomizationHelpers.cxx b/Framework/Core/src/WorkflowCustomizationHelpers.cxx
@@ -68,6 +68,7 @@ std::vector<ConfigParamSpec> WorkflowCustomizationHelpers::requiredWorkflowOptio
            {"aod-writer-keep", VariantType::String, "", {"Comma separated list of ORIGIN/DESCRIPTION/SUBSPECIFICATION:treename:col1/col2/..:filename"}},
 
            {"fairmq-rate-logging", VariantType::Int, 0, {"Rate logging for FairMQ channels"}},
+           {"aod-max-io-rate", VariantType::Float, 0.0f, {"Maximum I/O throughput in MB/s"}},
            {"fairmq-recv-buffer-size", VariantType::Int, 4, {"recvBufferSize option for FairMQ channels"}},
            {"fairmq-send-buffer-size", VariantType::Int, 4, {"sendBufferSize option for FairMQ channels"}},
            /// Find out a place where we can write the sockets

diff --git a/Framework/Core/src/WorkflowHelpers.cxx b/Framework/Core/src/WorkflowHelpers.cxx
@@ -34,7 +34,6 @@
 #include <utility>
 #include <vector>
 #include <climits>
-#include <thread>
 
 O2_DECLARE_DYNAMIC_LOG(workflow_helpers);