DPL: Enforce that dpl pipeline length is at least as long as number of TFs in flight

Author: davidrohr

Framework/Core/include/Framework/DefaultsHelpers.h

@@ -12,16 +12,24 @@
 #ifndef O2_FRAMEWORK_DEFAULTHELPERS_H_
 #define O2_FRAMEWORK_DEFAULTHELPERS_H_
 
+namespace fair::mq
+{
+class ProgOptions;
+}
+
 namespace o2::framework
 {
 enum struct DeploymentMode;
+struct DeviceConfig;
 
 struct DefaultsHelpers {
   static DeploymentMode deploymentMode();
   /// @true if running online
   static bool onlineDeploymentMode();
   /// get max number of timeslices in the queue
-  static unsigned int pipelineLength();
+  static unsigned int pipelineLength(unsigned int minLength);
+  static unsigned int pipelineLength(const fair::mq::ProgOptions& options);
+  static unsigned int pipelineLength(const DeviceConfig& dc);
 };
 } // namespace o2::framework
 

Framework/Core/src/ArrowSupport.cxx

@@ -564,7 +564,7 @@ o2::framework::ServiceSpec ArrowSupport::arrowBackendSpec()
                        if (dc.options.count("timeframes-rate-limit") && dc.options["timeframes-rate-limit"].defaulted() == false) {
                          config->maxTimeframes = std::stoll(dc.options["timeframes-rate-limit"].as<std::string>());
                        } else {
-                         config->maxTimeframes = readers * DefaultsHelpers::pipelineLength();
+                         config->maxTimeframes = readers * DefaultsHelpers::pipelineLength(dc);
                        }
                        static bool once = false;
                        // Until we guarantee this is called only once...

Framework/Core/src/DataProcessingDevice.cxx

@@ -1483,7 +1483,7 @@ void DataProcessingDevice::doPrepare(ServiceRegistryRef ref)
   auto& infos = state.inputChannelInfos;
 
   if (context.balancingInputs) {
-    static int pipelineLength = DefaultsHelpers::pipelineLength();
+    static int pipelineLength = DefaultsHelpers::pipelineLength(*ref.get<RawDeviceService>().device()->fConfig);
     static uint64_t ahead = getenv("DPL_MAX_CHANNEL_AHEAD") ? std::atoll(getenv("DPL_MAX_CHANNEL_AHEAD")) : std::max(8, std::min(pipelineLength - 48, pipelineLength / 2));
     auto newEnd = std::remove_if(pollOrder.begin(), pollOrder.end(), [&infos, limitNew = currentOldest.value + ahead](int a) -> bool {
       return infos[a].oldestForChannel.value > limitNew;
@@ -2259,12 +2259,14 @@ bool DataProcessingDevice::tryDispatchComputation(ServiceRegistryRef ref, std::v
     return false;
   }
 
-  auto postUpdateStats = [ref](DataRelayer::RecordAction const& action, InputRecord const& record, uint64_t tStart, uint64_t tStartMilli) {
+  int pipelineLength = DefaultsHelpers::pipelineLength(*ref.get<RawDeviceService>().device()->fConfig);
+
+  auto postUpdateStats = [ref, pipelineLength](DataRelayer::RecordAction const& action, InputRecord const& record, uint64_t tStart, uint64_t tStartMilli) {
     auto& stats = ref.get<DataProcessingStats>();
     auto& states = ref.get<DataProcessingStates>();
     std::atomic_thread_fence(std::memory_order_release);
     char relayerSlotState[1024];
-    int written = snprintf(relayerSlotState, 1024, "%d ", DefaultsHelpers::pipelineLength());
+    int written = snprintf(relayerSlotState, 1024, "%d ", pipelineLength);
     char* buffer = relayerSlotState + written;
     for (size_t ai = 0; ai != record.size(); ai++) {
       buffer[ai] = record.isValid(ai) ? '3' : '0';
@@ -2291,11 +2293,11 @@ bool DataProcessingDevice::tryDispatchComputation(ServiceRegistryRef ref, std::v
     count++;
   };
 
-  auto preUpdateStats = [ref](DataRelayer::RecordAction const& action, InputRecord const& record, uint64_t) {
+  auto preUpdateStats = [ref, pipelineLength](DataRelayer::RecordAction const& action, InputRecord const& record, uint64_t) {
     auto& states = ref.get<DataProcessingStates>();
     std::atomic_thread_fence(std::memory_order_release);
     char relayerSlotState[1024];
-    snprintf(relayerSlotState, 1024, "%d ", DefaultsHelpers::pipelineLength());
+    snprintf(relayerSlotState, 1024, "%d ", pipelineLength);
     char* buffer = strchr(relayerSlotState, ' ') + 1;
     for (size_t ai = 0; ai != record.size(); ai++) {
       buffer[ai] = record.isValid(ai) ? '2' : '0';

Framework/Core/src/DataRelayer.cxx

@@ -37,6 +37,7 @@
 #include "Framework/DataProcessingStates.h"
 #include "Framework/DataTakingContext.h"
 #include "Framework/DefaultsHelpers.h"
+#include "Framework/RawDeviceService.h"
 
 #include "Headers/DataHeaderHelpers.h"
 #include "Framework/Formatters.h"
@@ -48,6 +49,7 @@
 #include <fairmq/Channel.h>
 #include <functional>
 #include <fairmq/shmem/Message.h>
+#include <fairmq/Device.h>
 #include <fmt/format.h>
 #include <fmt/ostream.h>
 #include <span>
@@ -81,7 +83,7 @@ DataRelayer::DataRelayer(const CompletionPolicy& policy,
   std::scoped_lock<O2_LOCKABLE(std::recursive_mutex)> lock(mMutex);
 
   if (policy.configureRelayer == nullptr) {
-    static int pipelineLength = DefaultsHelpers::pipelineLength();
+    static int pipelineLength = DefaultsHelpers::pipelineLength(*services.get<RawDeviceService>().device()->fConfig);
     setPipelineLength(pipelineLength);
   } else {
     policy.configureRelayer(*this);

Framework/Core/src/DefaultsHelpers.cxx

@@ -11,30 +11,45 @@
 
 #include "Framework/DefaultsHelpers.h"
 #include "Framework/DataTakingContext.h"
+#include "Framework/DeviceConfig.h"
+#include <fairmq/ProgOptions.h>
+
 #include <cstdlib>
 #include <cstring>
 #include <stdexcept>
 
 namespace o2::framework
 {
 
-unsigned int DefaultsHelpers::pipelineLength()
+unsigned int DefaultsHelpers::pipelineLength(unsigned int minLength)
 {
   static bool override = getenv("DPL_DEFAULT_PIPELINE_LENGTH");
   if (override) {
     static unsigned int retval = atoi(getenv("DPL_DEFAULT_PIPELINE_LENGTH"));
-    return retval;
+    return std::max(minLength, retval);
   }
   DeploymentMode deploymentMode = DefaultsHelpers::deploymentMode();
   // just some reasonable numers
   // The number should really be tuned at runtime for each processor.
   if (deploymentMode == DeploymentMode::OnlineDDS || deploymentMode == DeploymentMode::OnlineECS || deploymentMode == DeploymentMode::FST) {
-    return 512;
+    return std::max(minLength, 512u);
   } else {
-    return 64;
+    return std::max(minLength, 64u);
   }
 }
 
+unsigned int DefaultsHelpers::pipelineLength(const DeviceConfig& dc)
+{
+  static unsigned int minLength = dc.options.count("timeframes-rate-limit") ? std::max(0, atoi(dc.options["timeframes-rate-limit"].as<std::string>().c_str())) : 0;
+  return pipelineLength(minLength);
+}
+
+unsigned int DefaultsHelpers::pipelineLength(const fair::mq::ProgOptions& options)
+{
+  static unsigned int minLength = std::max(0, atoi(options.GetValue<std::string>("timeframes-rate-limit").c_str()));
+  return pipelineLength(minLength);
+}
+
 static DeploymentMode getDeploymentMode_internal()
 {
   char* explicitMode = getenv("O2_DPL_DEPLOYMENT_MODE");

Framework/Core/src/runDataProcessing.cxx

@@ -817,7 +817,8 @@ void spawnDevice(uv_loop_t* loop,
     .sendInitialValue = true,
   });
 
-  for (size_t i = 0; i < DefaultsHelpers::pipelineLength(); ++i) {
+  unsigned int pipelineLength = DefaultsHelpers::pipelineLength(DeviceConfig{varmap});
+  for (size_t i = 0; i < pipelineLength; ++i) {
     allStates.back().registerState(DataProcessingStates::StateSpec{
       .name = fmt::format("matcher_variables/{}", i),
       .stateId = static_cast<short>((short)(ProcessingStateId::CONTEXT_VARIABLES_BASE) + i),
@@ -826,7 +827,7 @@ void spawnDevice(uv_loop_t* loop,
     });
   }
 
-  for (size_t i = 0; i < DefaultsHelpers::pipelineLength(); ++i) {
+  for (size_t i = 0; i < pipelineLength; ++i) {
     allStates.back().registerState(DataProcessingStates::StateSpec{
       .name = fmt::format("data_relayer/{}", i),
       .stateId = static_cast<short>((short)(ProcessingStateId::DATA_RELAYER_BASE) + i),