DPL: combine resource offers to reduce processing latency

ktf · ktf · commit 29823ca8b40b · 2026-05-24T09:31:26.000+02:00
diff --git a/Framework/Core/src/ArrowSupport.cxx b/Framework/Core/src/ArrowSupport.cxx
@@ -120,31 +120,37 @@ struct ResourceStats {
 struct ResourceSpec {
   char const* name;
   char const* unit;
-  char const* api;                /// The callback to give resources to a device
   int64_t maxAvailable;           /// Maximum available quantity for a resource
   int64_t maxQuantum;             /// Largest offer which can be given
   int64_t minQuantum;             /// Smallest offer which can be given
   int64_t metricOfferScaleFactor; /// The scale factor between the metric accounting and offers accounting
 };
+struct PendingResourceOffer {
+  int64_t amount = 0;
+  int64_t candidate = -1;
+};
 
-auto offerResources(ResourceState& resourceState,
-                    ResourceSpec const& resourceSpec,
-                    ResourceStats& resourceStats,
-                    std::vector<DeviceSpec> const& specs,
-                    std::vector<DeviceInfo> const& infos,
-                    DevicesManager& manager,
-                    int64_t offerConsumedCurrentValue,
-                    int64_t offerExpiredCurrentValue,
-                    int64_t acquiredResourceCurrentValue,
-                    int64_t disposedResourceCurrentValue,
-                    size_t timestamp,
-                    DeviceMetricsInfo& driverMetrics,
-                    std::function<void(DeviceMetricsInfo&, int value, size_t timestamp)>& availableResourceMetric,
-                    std::function<void(DeviceMetricsInfo&, int value, size_t timestamp)>& unusedOfferedResourceMetric,
-                    std::function<void(DeviceMetricsInfo&, int value, size_t timestamp)>& offeredResourceMetric,
-                    void* signpostId) -> void
+/// Compute how much of a resource can be offered and to which device,
+/// update accounting, but do NOT send the message yet.
+/// Returns the candidate device index and offer amount.
+auto computeResourceOffer(ResourceState& resourceState,
+                          ResourceSpec const& resourceSpec,
+                          ResourceStats& resourceStats,
+                          std::vector<DeviceSpec> const& specs,
+                          std::vector<DeviceInfo> const& infos,
+                          int64_t offerConsumedCurrentValue,
+                          int64_t offerExpiredCurrentValue,
+                          int64_t acquiredResourceCurrentValue,
+                          int64_t disposedResourceCurrentValue,
+                          size_t timestamp,
+                          DeviceMetricsInfo& driverMetrics,
+                          std::function<void(DeviceMetricsInfo&, int value, size_t timestamp)>& availableResourceMetric,
+                          std::function<void(DeviceMetricsInfo&, int value, size_t timestamp)>& unusedOfferedResourceMetric,
+                          std::function<void(DeviceMetricsInfo&, int value, size_t timestamp)>& offeredResourceMetric,
+                          void* signpostId) -> PendingResourceOffer
 {
   O2_SIGNPOST_ID_FROM_POINTER(sid, rate_limiting, signpostId);
+  PendingResourceOffer result;
   /// We loop over the devices, starting from where we stopped last time
   /// offering the minimum offer to each one
   int64_t lastCandidate = -1;
@@ -196,10 +202,11 @@ auto offerResources(ResourceState& resourceState,
     O2_SIGNPOST_EVENT_EMIT(rate_limiting, sid, "offer",
                            "Offering %llu %{public}s out of %llu to %{public}s",
                            possibleOffer, resourceSpec.unit, resourceState.available, specs[candidate].id.c_str());
-    manager.queueMessage(specs[candidate].id.c_str(), fmt::format(fmt::runtime(resourceSpec.api), possibleOffer).data());
     resourceState.available -= possibleOffer;
     resourceState.offered += possibleOffer;
     lastCandidate = candidate;
+    result.amount = possibleOffer;
+    result.candidate = candidate;
   }
   // We had at least a valid candidate, so
   // next time we offer to the next device.
@@ -236,6 +243,7 @@ auto offerResources(ResourceState& resourceState,
   unusedOfferedResourceMetric(driverMetrics, unusedOfferedResource, timestamp);
 
   offeredResourceMetric(driverMetrics, resourceState.offered, timestamp);
+  return result;
 };
 
 auto processTimeslices = [](size_t index, DeviceMetricsInfo& deviceMetrics, bool& changed,
@@ -472,7 +480,6 @@ o2::framework::ServiceSpec ArrowSupport::arrowBackendSpec()
                        static const ResourceSpec shmResourceSpec{
                          .name = "shared memory",
                          .unit = "MB",
-                         .api = "/shm-offer {}",
                          .maxAvailable = (int64_t)registry.get<RateLimitConfig>().maxMemory,
                          .maxQuantum = 100,
                          .minQuantum = 50,
@@ -481,7 +488,6 @@ o2::framework::ServiceSpec ArrowSupport::arrowBackendSpec()
                        static const ResourceSpec timesliceResourceSpec{
                          .name = "timeslice",
                          .unit = "timeslices",
-                         .api = "/timeslice-offer {}",
                          .maxAvailable = (int64_t)registry.get<RateLimitConfig>().maxTimeframes,
                          .maxQuantum = 1,
                          .minQuantum = 1,
@@ -502,17 +508,33 @@ o2::framework::ServiceSpec ArrowSupport::arrowBackendSpec()
                          .lowCount = shmResourceState.available - shmResourceSpec.minQuantum > 0 ? 0 : 1
                        };
 
-                       offerResources(timesliceResourceState, timesliceResourceSpec, timesliceResourceStats,
-                                      specs, infos, manager, totalTimeframesConsumed, totalTimeslicesExpired,
+                       auto timesliceOffer = computeResourceOffer(timesliceResourceState, timesliceResourceSpec, timesliceResourceStats,
+                                      specs, infos, totalTimeframesConsumed, totalTimeslicesExpired,
                                       totalTimeslicesStarted, totalTimeslicesDone, timestamp, driverMetrics,
                                       availableTimeslicesMetric, unusedOfferedTimeslicesMetric, offeredTimeslicesMetric,
                                       (void*)&sm);
 
-                       offerResources(shmResourceState, shmResourceSpec, shmResourceStats,
-                                      specs, infos, manager, shmOfferBytesConsumed, totalBytesExpired,
+                       auto shmOffer = computeResourceOffer(shmResourceState, shmResourceSpec, shmResourceStats,
+                                      specs, infos, shmOfferBytesConsumed, totalBytesExpired,
                                       totalBytesCreated, totalBytesDestroyed, timestamp, driverMetrics,
                                       availableSharedMemoryMetric, unusedOfferedSharedMemoryMetric, offeredSharedMemoryMetric,
-                                      (void*)&sm); },
+                                      (void*)&sm);
+
+                       // Send a single combined offer so the reader gets both resources atomically
+                       if (timesliceOffer.candidate >= 0 && shmOffer.candidate >= 0) {
+                         O2_SIGNPOST_ID_FROM_POINTER(sid, rate_limiting, (void*)&sm);
+                         O2_SIGNPOST_EVENT_EMIT(rate_limiting, sid, "offer",
+                                                "Sending combined offer: %lld MB shm + %lld timeslices to %{public}s",
+                                                shmOffer.amount, timesliceOffer.amount, specs[timesliceOffer.candidate].id.c_str());
+                         manager.queueMessage(specs[timesliceOffer.candidate].id.c_str(),
+                                              fmt::format("/combined-offer {} {}", shmOffer.amount, timesliceOffer.amount).data());
+                       } else if (timesliceOffer.candidate >= 0) {
+                         manager.queueMessage(specs[timesliceOffer.candidate].id.c_str(),
+                                              fmt::format("/timeslice-offer {}", timesliceOffer.amount).data());
+                       } else if (shmOffer.candidate >= 0) {
+                         manager.queueMessage(specs[shmOffer.candidate].id.c_str(),
+                                              fmt::format("/shm-offer {}", shmOffer.amount).data());
+                       } },
     .postDispatching = [](ProcessingContext& ctx, void* service) {
                        using DataHeader = o2::header::DataHeader;
                        auto* arrow = reinterpret_cast<ArrowContext*>(service);
diff --git a/Framework/Core/src/ComputingQuotaEvaluator.cxx b/Framework/Core/src/ComputingQuotaEvaluator.cxx
@@ -240,9 +240,18 @@ void ComputingQuotaEvaluator::dispose(int taskId)
     if (offer.valid == false) {
       continue;
     }
-    if (offer.sharedMemory <= 0) {
+    // Decrement timeslices after each use and track if a decrement happened.
+    // - SHM-only offers (ts=0 from start): no decrement, kept alive while shm > 0
+    // - Timeslice-only offers (shm=0, ts>0): ts decremented, invalidated when shm <= 0
+    // - Combined offers (shm>0, ts>0): ts decremented, invalidated when ts reaches 0
+    bool timesliceConsumed = false;
+    if (offer.timeslices > 0) {
+      offer.timeslices--;
+      timesliceConsumed = true;
+    }
+    if (offer.sharedMemory <= 0 || (timesliceConsumed && offer.timeslices <= 0)) {
       O2_SIGNPOST_ID_FROM_POINTER(oid, quota, (void*)(int64_t)(oi * 8));
-      O2_SIGNPOST_END(quota, oid, "offers", "Offer %d back to not needed.", oi);
+      O2_SIGNPOST_END(quota, oid, "offers", "Offer %d back to not needed (shm=%lli, ts=%lli).", oi, offer.sharedMemory, offer.timeslices);
       offer.valid = false;
       offer.score = OfferScore::Unneeded;
     }
diff --git a/Framework/Core/src/WSDriverClient.cxx b/Framework/Core/src/WSDriverClient.cxx
@@ -120,6 +120,48 @@ void on_connect(uv_connect_t* connection, int status)
 
     state.pendingOffers.push_back(offer);
   });
+  client->observe("/combined-offer", [ref = context->ref](std::string_view cmd) {
+    O2_SIGNPOST_ID_GENERATE(wid, ws_client);
+    O2_SIGNPOST_START(ws_client, wid, "combined-offer", "Received combined offer.");
+    auto& state = ref.get<DeviceState>();
+    static constexpr int prefixSize = std::string_view{"/combined-offer "}.size();
+    if (prefixSize > cmd.size()) {
+      O2_SIGNPOST_END_WITH_ERROR(ws_client, wid, "combined-offer", "Malformed combined offer");
+      return;
+    }
+    cmd.remove_prefix(prefixSize);
+    // Parse "<shm_mb> <timeslices>"
+    size_t shmMB;
+    auto shmError = std::from_chars(cmd.data(), cmd.data() + cmd.size(), shmMB);
+    if (shmError.ec != std::errc() || shmError.ptr >= cmd.data() + cmd.size()) {
+      O2_SIGNPOST_END_WITH_ERROR(ws_client, wid, "combined-offer", "Malformed combined offer (shm)");
+      return;
+    }
+    // Skip space
+    auto remaining = std::string_view(shmError.ptr, cmd.data() + cmd.size() - shmError.ptr);
+    if (remaining.empty() || remaining[0] != ' ') {
+      O2_SIGNPOST_END_WITH_ERROR(ws_client, wid, "combined-offer", "Malformed combined offer (separator)");
+      return;
+    }
+    remaining.remove_prefix(1);
+    int64_t timeslices;
+    auto tsError = std::from_chars(remaining.data(), remaining.data() + remaining.size(), timeslices);
+    if (tsError.ec != std::errc()) {
+      O2_SIGNPOST_END_WITH_ERROR(ws_client, wid, "combined-offer", "Malformed combined offer (timeslices)");
+      return;
+    }
+    ComputingQuotaOffer offer{
+      .cpu = 0,
+      .memory = 0,
+      .sharedMemory = (int64_t)(shmMB * 1000000),
+      .timeslices = timeslices,
+      .runtime = 10000,
+      .user = -1,
+      .valid = true};
+    state.pendingOffers.push_back(offer);
+    O2_SIGNPOST_END(ws_client, wid, "combined-offer", "Received combined offer: %zu MB shm + %lli timeslices. Total pending %zu.",
+                    shmMB, timeslices, state.pendingOffers.size());
+  });
   client->observe("/timeslice-offer", [ref = context->ref](std::string_view cmd) {
     O2_SIGNPOST_ID_GENERATE(wid, ws_client);
     O2_SIGNPOST_START(ws_client, wid, "timeslice-offer", "Received timeslice offer.");
