Merge pull request #30857 from redpanda-data/stephan/inline-storage-batches

various: more alloc opts

Author: StephanDollberg

src/v/raft/group_configuration.h

@@ -291,6 +291,7 @@ class group_configuration
     void for_each_learner(Func&& f) const;
 
     const std::vector<vnode>& all_nodes() const;
+    const std::vector<vnode>& replicas() const { return _all_replicas; }
 
     std::optional<vnode> find_by_node_id(model::node_id) const;
 
@@ -493,7 +494,7 @@ void group_configuration::for_each_broker(Func&& f) const {
 
 template<typename Func>
 void group_configuration::for_each_replica(Func&& f) const {
-    std::ranges::for_each(_all_replicas, std::forward<Func>(f));
+    std::ranges::for_each(replicas(), std::forward<Func>(f));
 }
 
 template<typename Func, typename Ret>

src/v/raft/replicate_batcher.cc

@@ -247,7 +247,7 @@ ss::future<> replicate_batcher::flush(
         auto meta = _ptr->meta();
         const auto term = model::term_id(meta.term);
         chunked_vector<model::record_batch> data;
-        std::vector<item_ptr> notifications;
+        notifications_t notifications;
         ssx::semaphore_units item_memory_units(_max_batch_size_sem, 0);
         auto force_flush_requested = false;
         auto has_quorum_ack_requests = false;
@@ -266,9 +266,16 @@ ss::future<> replicate_batcher::flush(
                 has_quorum_ack_requests
                   = has_quorum_ack_requests
                     || (n->get_consistency_level() == consistency_level::quorum_ack);
-                for (auto& b : batches) {
-                    b.set_term(term);
-                    data.push_back(std::move(b));
+                if (data.empty()) {
+                    data = std::move(batches);
+                    for (auto& b : data) {
+                        b.set_term(term);
+                    }
+                } else {
+                    for (auto& b : batches) {
+                        b.set_term(term);
+                        data.push_back(std::move(b));
+                    }
                 }
                 notifications.push_back(std::move(n));
             } else {
@@ -323,7 +330,7 @@ ss::future<> replicate_batcher::flush(
 template<typename Predicate>
 static void propagate_result(
   result<replicate_result> r,
-  std::vector<replicate_batcher::item_ptr>& notifications,
+  replicate_batcher::notifications_t& notifications,
   const Predicate& pred) {
     if (r.has_error()) {
         // propagate an error
@@ -344,8 +351,8 @@ static void propagate_result(
     }
 }
 
-static void propagate_current_exception(
-  std::vector<replicate_batcher::item_ptr>& notifications) {
+static void
+propagate_current_exception(replicate_batcher::notifications_t& notifications) {
     // iterate backward to calculate last offsets
     auto e = std::current_exception();
     for (auto& n : notifications) {
@@ -354,7 +361,7 @@ static void propagate_current_exception(
 }
 
 ss::future<> replicate_batcher::do_flush(
-  std::vector<replicate_batcher::item_ptr> notifications,
+  replicate_batcher::notifications_t notifications,
   append_entries_request req,
   std::vector<ssx::semaphore_units> u,
   absl::flat_hash_map<vnode, follower_req_seq> seqs) {

src/v/raft/replicate_batcher.h

@@ -12,6 +12,7 @@
 #pragma once
 
 #include "absl/container/flat_hash_map.h"
+#include "absl/container/inlined_vector.h"
 #include "base/outcome.h"
 #include "container/chunked_vector.h"
 #include "raft/types.h"
@@ -83,6 +84,7 @@ class replicate_batcher {
         ss::optimized_optional<ss::abort_source::subscription> _abort_sub;
     };
     using item_ptr = ss::lw_shared_ptr<item>;
+    using notifications_t = absl::InlinedVector<item_ptr, 5>;
     explicit replicate_batcher(consensus* ptr, size_t cache_size);
 
     replicate_batcher(replicate_batcher&&) noexcept = default;
@@ -100,7 +102,7 @@ class replicate_batcher {
 
 private:
     ss::future<> do_flush(
-      std::vector<item_ptr>,
+      notifications_t,
       append_entries_request,
       std::vector<ssx::semaphore_units>,
       absl::flat_hash_map<vnode, follower_req_seq>);

src/v/raft/replicate_entries_stm.cc

@@ -9,6 +9,7 @@
 
 #include "raft/replicate_entries_stm.h"
 
+#include "absl/container/inlined_vector.h"
 #include "base/outcome.h"
 #include "base/outcome_future_utils.h"
 #include "model/fundamental.h"
@@ -238,13 +239,15 @@ inline bool replicate_entries_stm::should_skip_follower_request(vnode id) {
 
 ss::future<result<replicate_result>> replicate_entries_stm::apply(units_t u) {
     // first append lo leader log, no flushing
-    auto cfg = _ptr->config();
-    cfg.for_each_replica([this](const vnode& rni) {
+    const auto& config_replicas = _ptr->config().replicas();
+    absl::InlinedVector<vnode, 5> replicas(
+      config_replicas.begin(), config_replicas.end());
+    for (const auto& rni : replicas) {
         // suppress follower heartbeat, before appending to self log
         if (rni != _ptr->_self) {
             _inflight_appends.emplace(rni, _ptr->track_append_inflight(rni));
         }
-    });
+    }
     _units = ss::make_lw_shared<units_t>(std::move(u));
     _append_result = co_await append_to_self();
 
@@ -255,12 +258,12 @@ ss::future<result<replicate_result>> replicate_entries_stm::apply(units_t u) {
     // store committed offset to check if it advanced
     _initial_committed_offset = _ptr->committed_offset();
     // dispatch requests to followers & leader flush
-    cfg.for_each_replica([this](const vnode& rni) {
+    for (const auto& rni : replicas) {
         // We are not dispatching request to followers that are
         // recovering
         if (should_skip_follower_request(rni)) {
             _inflight_appends[rni].mark_finished();
-            return;
+            continue;
         }
         if (rni != _ptr->self()) {
             auto it = _ptr->_fstates.find(rni);
@@ -271,7 +274,7 @@ ss::future<result<replicate_result>> replicate_entries_stm::apply(units_t u) {
         }
         ++_requests_count;
         (void)dispatch_one(rni); // background
-    });
+    }
 
     // wait for the requests to be dispatched in background and then release
     // units