Merge pull request #29061 from Lazin/ct/compute-data-threshold-across-shards

[CORE-15334] ct: Introduce new scheduling algorithm for lower latency

Author: Lazin

src/v/cloud_topics/level_zero/batcher/BUILD

@@ -65,6 +65,7 @@ redpanda_cc_library(
         "//src/v/cloud_topics/level_zero/pipeline:write_request",
         "//src/v/config",
         "//src/v/model",
+        "//src/v/ssx:semaphore",
         "//src/v/utils:retry_chain_node",
         "//src/v/utils:uuid",
         "@abseil-cpp//absl/container:btree",

src/v/cloud_topics/level_zero/batcher/aggregator.cc

@@ -90,7 +90,7 @@ iobuf aggregator<Clock>::get_stream() {
     iobuf concat;
     for (auto& p : _aggregated) {
         if (p->ref != nullptr) {
-            concat.append(std::move(p->ref->data_chunk.payload));
+            concat.append_fragments(std::move(p->ref->data_chunk.payload));
         }
     }
     return concat;

src/v/cloud_topics/level_zero/batcher/batcher.cc

@@ -51,7 +51,10 @@ batcher<Clock>::batcher(
   , _rtc(_as)
   , _logger(cd_log, _rtc)
   , _stage(std::move(stage))
-  , _probe(config::shard_local_cfg().disable_metrics()) {}
+  , _probe(config::shard_local_cfg().disable_metrics())
+  , _upload_sem(
+      config::shard_local_cfg().cloud_storage_max_connections(), "l0/batcher") {
+}
 
 template<class Clock>
 ss::future<> batcher<Clock>::start() {
@@ -254,29 +257,56 @@ ss::future<> batcher<Clock>::bg_controller_loop() {
             co_return;
         }
 
+        // Acquire semaphore units to limit concurrent background fibers.
+        // This blocks until a slot is available.
+        auto units_fut = co_await ss::coroutine::as_future(
+          ss::get_units(_upload_sem, 1, _as));
+
         auto list = _stage.pull_write_requests(
-          10_MiB); // TODO: use configuration parameter
+          config::shard_local_cfg()
+            .cloud_topics_produce_batching_size_threshold(),
+          config::shard_local_cfg()
+            .cloud_topics_produce_cardinality_threshold());
+
+        bool complete = list.complete;
+
+        if (units_fut.failed()) {
+            vlog(
+              _logger.info,
+              "Batcher upload loop is shutting down: {}",
+              units_fut.get_exception());
+            co_return;
+        }
+        auto units = std::move(units_fut.get());
 
         // We can spawn the work in the background without worrying about memory
-        // usage because the pipeline tracks the memory usage for us and will
-        // stop accepting new write requests if we go over the limit.
-        ssx::spawn_with_gate(_gate, [this, list = std::move(list)]() mutable {
-            return run_once(std::move(list))
-              .then([this](std::expected<std::monostate, errc> res) {
-                  if (!res.has_value()) {
-                      if (res.error() == errc::shutting_down) {
-                          vlog(
-                            _logger.info,
-                            "Batcher upload loop is shutting down");
-                      } else {
-                          vlog(
-                            _logger.info,
-                            "Batcher upload loop error: {}",
-                            res.error());
-                      }
-                  }
-              });
-        });
+        // usage because the background fibers is holding units acquired above.
+        ssx::spawn_with_gate(
+          _gate,
+          [this, list = std::move(list), units = std::move(units)]() mutable {
+              return run_once(std::move(list))
+                .then([this](std::expected<std::monostate, errc> res) {
+                    if (!res.has_value()) {
+                        if (res.error() == errc::shutting_down) {
+                            vlog(
+                              _logger.info,
+                              "Batcher upload loop is shutting down");
+                        } else {
+                            vlog(
+                              _logger.info,
+                              "Batcher upload loop error: {}",
+                              res.error());
+                        }
+                    }
+                })
+                .finally([u = std::move(units)] {});
+          });
+
+        // The work is spawned in the background so we can grab data for the
+        // next L0 object. If complete==true, all pending requests were pulled,
+        // so wait for more. If complete==false, there are more pending
+        // requests.
+        more_work = !complete;
     }
 }
 

src/v/cloud_topics/level_zero/batcher/batcher.h

@@ -24,6 +24,7 @@
 #include "config/property.h"
 #include "model/fundamental.h"
 #include "model/record_batch_reader.h"
+#include "ssx/semaphore.h"
 #include "utils/retry_chain_node.h"
 #include "utils/uuid.h"
 
@@ -121,5 +122,8 @@ class batcher {
     write_pipeline<Clock>::stage _stage;
 
     batcher_probe _probe;
+
+    // Limit the number of concurrent background fibers running run_once
+    ssx::named_semaphore<Clock> _upload_sem;
 };
 } // namespace cloud_topics::l0

src/v/cloud_topics/level_zero/common/level_zero_probe.cc

@@ -211,31 +211,16 @@ void write_request_scheduler_probe::setup_internal_metrics(bool disable) {
     _metrics.add_group(
       prometheus_sanitize::metrics_name("cloud_topics_write_request_scheduler"),
       {sm::make_counter(
-         "data_threshold_requests",
-         [this] { return _data_threshold_requests; },
+         "scheduler_requests",
+         [this] { return _scheduler_requests; },
          sm::description(
-           "Number of write requests scheduled by data threshold policy."),
+           "Number of write requests scheduled by the scheduler."),
          labels),
 
        sm::make_counter(
-         "data_threshold_bytes",
-         [this] { return _data_threshold_bytes; },
-         sm::description(
-           "Total number of bytes scheduled by data threshold policy."),
-         labels),
-
-       sm::make_counter(
-         "time_fallback_requests",
-         [this] { return _time_fallback_requests; },
-         sm::description(
-           "Number of write requests scheduled by time based fallback policy."),
-         labels),
-
-       sm::make_counter(
-         "time_fallback_bytes",
-         [this] { return _time_fallback_bytes; },
-         sm::description(
-           "Total number of bytes scheduled by time based fallback policy."),
+         "scheduler_bytes",
+         [this] { return _scheduler_bytes; },
+         sm::description("Total number of bytes scheduled by scheduler."),
          labels),
 
        sm::make_counter(
@@ -261,6 +246,12 @@ void write_request_scheduler_probe::setup_internal_metrics(bool disable) {
          "rx_bytes_xshard",
          [this] { return _rx_bytes_xshard; },
          sm::description("Total number of bytes received from another shard."),
+         labels),
+
+       sm::make_gauge(
+         "active_groups",
+         [this] { return _active_groups; },
+         sm::description("Number of active upload groups in the scheduler."),
          labels)});
 }
 batcher_probe::batcher_probe(bool disable) { setup_internal_metrics(disable); }

src/v/cloud_topics/level_zero/common/level_zero_probe.h

@@ -97,14 +97,9 @@ class write_request_scheduler_probe {
 public:
     explicit write_request_scheduler_probe(bool disable);
 
-    void register_data_threshold(size_t bytes) {
-        _data_threshold_requests += 1;
-        _data_threshold_bytes += bytes;
-    }
-
-    void register_time_fallback(size_t bytes) {
-        _time_fallback_requests += 1;
-        _time_fallback_bytes += bytes;
+    void register_request(size_t bytes) {
+        _scheduler_requests += 1;
+        _scheduler_bytes += bytes;
     }
 
     void register_send_xshard(size_t bytes) {
@@ -117,23 +112,23 @@ class write_request_scheduler_probe {
         _rx_bytes_xshard += bytes;
     }
 
+    void set_active_groups(uint64_t count) { _active_groups = count; }
+
 private:
     void setup_internal_metrics(bool disable);
 
-    /// Number of write requests and total bytes scheduled by data threshold
-    /// policy.
-    uint64_t _data_threshold_requests{0};
-    uint64_t _data_threshold_bytes{0};
-    /// Number of write requests and total bytes scheduled by time based
-    /// fallback policy.
-    uint64_t _time_fallback_requests{0};
-    uint64_t _time_fallback_bytes{0};
+    /// Number of write requests and total bytes scheduled
+    uint64_t _scheduler_requests{0};
+    uint64_t _scheduler_bytes{0};
     /// Number of requests and total bytes proxied to another shard
     uint64_t _tx_requests_xshard{0};
     uint64_t _tx_bytes_xshard{0};
     /// Number of requests and total bytes received from another shard
     uint64_t _rx_requests_xshard{0};
     uint64_t _rx_bytes_xshard{0};
+    /// Number of active upload groups
+    uint64_t _active_groups{0};
+
     metrics::internal_metric_groups _metrics;
 };
 

src/v/cloud_topics/level_zero/pipeline/base_pipeline.h

@@ -191,6 +191,12 @@ class base_pipeline {
         return _stages.next_stage(s);
     }
 
+    /// Return next stage index without checking if stage is registered.
+    /// This is useful for accessing pre-allocated resources.
+    int next_stage_index(pipeline_stage s) const {
+        return _stages.next_stage_index(s);
+    }
+
     /// Resolve every pending write that matches the predicate with an error.
     template<typename Pred>
     void remove_requests(Pred pred, errc error, std::string_view reason) {

src/v/cloud_topics/level_zero/pipeline/pipeline_stage.cc

@@ -50,6 +50,20 @@ pipeline_stage pipeline_stage_container::register_pipeline_stage() noexcept {
     return pipeline_stage(&_stages.at(_registered++));
 }
 
+int pipeline_stage_container::next_stage_index(pipeline_stage old) const {
+    if (old == unassigned_pipeline_stage) {
+        // First stage is index 0
+        return 0;
+    }
+    auto old_ix = old()->get_numeric_id();
+    auto next_ix = old_ix + 1;
+    // Return -1 if we would exceed the allocated stages
+    if (static_cast<size_t>(next_ix) >= _stages.size()) {
+        return -1;
+    }
+    return next_ix;
+}
+
 } // namespace cloud_topics::l0
 
 auto fmt::formatter<cloud_topics::l0::pipeline_stage>::format(

src/v/cloud_topics/level_zero/pipeline/pipeline_stage.h

@@ -48,6 +48,15 @@ class pipeline_stage_container {
     pipeline_stage first_stage() const;
     pipeline_stage register_pipeline_stage() noexcept;
 
+    /// Get the numeric index of the next stage after the given stage.
+    /// Unlike next_stage(), this method does not check if the next stage
+    /// is registered. It returns the index even if the stage hasn't been
+    /// registered yet. This is useful for accessing pre-allocated resources
+    /// (like counters) that are indexed by stage number.
+    /// \param old The current pipeline stage
+    /// \return The index of the next stage, or -1 if old is unassigned or last
+    int next_stage_index(pipeline_stage old) const;
+
 private:
     std::vector<pipeline_stage_id> _stages;
     size_t _registered{0};

src/v/cloud_topics/level_zero/pipeline/write_pipeline.cc

@@ -269,6 +269,20 @@ void write_pipeline<Clock>::stage::signal_next_stage() {
     _parent->signal(_parent->next_stage(_ps));
 }
 
+template<class Clock>
+void write_pipeline<Clock>::stage::enqueue_foreign_request(
+  write_request<Clock>& req, bool signal) {
+    // Foreign requests are proxied from another shard where their bytes
+    // were already accounted for. We place them directly at the next stage
+    // without any byte accounting.
+    auto next = _parent->next_stage(_ps);
+    req.stage = next;
+    _parent->get_pending().push_back(req);
+    if (signal) {
+        _parent->signal(next);
+    }
+}
+
 template<class Clock>
 write_pipeline<Clock>::write_requests_list
 write_pipeline<Clock>::stage::pull_write_requests(
@@ -374,14 +388,32 @@ size_t write_pipeline<Clock>::stage_bytes(pipeline_stage s) const {
     if (s == unassigned_pipeline_stage) {
         return 0;
     }
-    return _stage_bytes[static_cast<size_t>(s()->get_numeric_id())];
+    return _stage_bytes[static_cast<size_t>(s()->get_numeric_id())].count;
+}
+
+template<class Clock>
+const std::atomic<size_t>*
+write_pipeline<Clock>::stage_bytes_ref(pipeline_stage s) const {
+    if (s == unassigned_pipeline_stage) {
+        return nullptr;
+    }
+    return &_stage_bytes[static_cast<size_t>(s()->get_numeric_id())].count;
+}
+
+template<class Clock>
+const std::atomic<size_t>*
+write_pipeline<Clock>::stage_bytes_ref_by_index(int index) const {
+    if (index < 0 || static_cast<size_t>(index) >= _stage_bytes.size()) {
+        return nullptr;
+    }
+    return &_stage_bytes[static_cast<size_t>(index)].count;
 }
 
 template<class Clock>
 size_t write_pipeline<Clock>::current_size() const {
     size_t total = 0;
-    for (auto bytes : _stage_bytes) {
-        total += bytes;
+    for (const auto& bytes : _stage_bytes) {
+        total += bytes.count.load(std::memory_order_relaxed);
     }
     return total;
 }