perf(redis): reduce large String receive and segment copies

CMakeLists.txt

@@ -2190,6 +2190,14 @@ endforeach()
     )
     set_tests_properties(lavik_reply_fragments_e2e PROPERTIES TIMEOUT 120)
 
+    add_test(
+      NAME lavik_large_bulk_e2e
+      COMMAND ${Python3_EXECUTABLE}
+              ${CMAKE_CURRENT_SOURCE_DIR}/tests/large_bulk_e2e.py
+              $<TARGET_FILE:lavik>
+    )
+    set_tests_properties(lavik_large_bulk_e2e PROPERTIES TIMEOUT 180)
+
     add_executable(lavik_grouped_recovery_e2e_test
       tests/grouped/recovery_e2e_test.cpp
     )

docs/architecture/02-request-serving.md

@@ -116,16 +116,23 @@ racing closure is either rejected or remains visible to the drain.
 
 ## Request pipeline
 
-1. `ReadCommandBatch` reads into a connection-local buffer and feeds
-   `RespCommandParser`. Up to 128 parsed commands are retained in wire order.
+1. `ReadCommandBatch` feeds `RespCommandParser` through a small connection-local
+   buffer. Large bulk bodies can instead receive directly into parser-owned
+   argument storage after their first payload bytes arrive. The opt-in Bycorf
+   read drains already-received buffers without waiting to fill the destination;
+   TLS continues through its decrypted read path. Up to 128 parsed commands are
+   retained in wire order.
    Ordinary connections collectively hold at most `maxmemory-clients`, which
    accepts an absolute byte size or a percentage of effective maxmemory and
    defaults to 5%. The effective limit is divided into fixed worker shares.
    This abuse-control quota is independent of retained-memory admission; the
    server does not reserve temporary request bytes against `maxmemory` a second
-   time. Admission runs once after each socket read;
-   parsed wire bytes follow their command until execution finishes, while blank
-   input is retired as soon as the parser returns to its idle state. The
+   time. Ordinary reads admit received bytes; direct bulk reads admit bounded
+   destination windows before allocation. Prepaid bytes become received-byte
+   charges without a second admission, and unread reservations are released
+   on connection teardown. Parsed wire bytes follow their command until
+   execution finishes, while blank input is retired as soon as the parser
+   returns to its idle state. The
    offending connection receives an error and closes when its worker's share is
    exhausted. Zero disables this client-specific limit. As in Valkey, every
    nonzero process-wide client allowance has a 128 KiB
@@ -136,14 +143,13 @@ racing closure is either rejected or remains visible to the drain.
    Independently, `client-query-buffer-limit` bounds the wire bytes retained
    while one connection incrementally assembles a command. It defaults to 1
    GiB and, like Redis, accepts an absolute value from 1 MiB through
-   `LONG_MAX`. Lavik consumes the small socket input window directly into
-   argument strings, so this is a parser-retained-byte limit rather than a
-   requirement for a second contiguous query buffer. Completed pipelined and
+   `LONG_MAX`. Both receive paths assemble argument strings without requiring
+   a second contiguous command buffer. Completed pipelined and
    `MULTI` commands remain governed by `maxmemory-clients` after parser
    ownership ends. `CONFIG SET client-query-buffer-limit` updates a process-wide
-   atomic. Existing connections refresh it once per socket-read parsing round,
-   avoiding an atomic load for each RESP token while ensuring a partially read
-   command is checked against the new limit when more bytes arrive.
+   atomic. Existing connections refresh it once per socket-read parsing round
+   and before committing a direct read, so a partially read command is checked
+   against the new limit when more bytes arrive.
 2. `BuildCommandRequest` resolves case-insensitive static metadata from the
    command table and copies the connection's current database ID into the
    request.

docs/operations/metrics.md

@@ -62,7 +62,11 @@ It accepts either a percentage such as `5%` or an absolute size such as
 `512mb`; the default is `5%`, and `0` disables this client-specific limit.
 Every nonzero value has a 128 KiB effective minimum so an accidentally tiny
 setting does not make administrative commands unusable. The process-wide value
-is divided into fixed worker shares and checked once after each socket read.
+is divided into fixed worker shares. Ordinary reads charge received bytes;
+large bulk reads charge bounded destination windows before allocation, then
+consume that reservation as data arrives. Unread reservations are released
+when the connection closes. The quota tracks received and reserved request
+bytes, not allocator capacity or exact heap usage.
 
 `--client-query-buffer-limit` is the independent per-connection hard limit for
 an incomplete command. It accepts an absolute Redis-style memory size, defaults
@@ -72,7 +76,8 @@ configuration files. The connection closes if incremental parsing crosses the
 limit; completed commands continue to count against `maxmemory-clients` until
 execution or transaction teardown releases them. `CONFIG GET/SET
 client-query-buffer-limit` reads or changes the live process-wide value;
-existing connections apply a change before their next parsing round.
+existing connections apply a change before their next parsing round and before
+committing bytes received directly into a large argument.
 
 The process budget is split evenly across workers and unused capacity is not
 borrowed across workers. SET/MSET/INCR performs one worker-local retained-growth
@@ -218,7 +223,8 @@ sum by (result) (rate(lavik_storage_defrag_runs_total[5m]))
 - `lavik_client_request_buffer_limit_bytes`: effective client request-buffer
   limit after resolving percentages and the nonzero 128 KiB floor.
 - `lavik_client_buffered_request_bytes`: wire bytes currently retained by
-  ordinary parsers, ready command batches, and queued transactions.
+  ordinary parsers, ready command batches, and queued transactions, plus
+  prepaid destination bytes for incomplete direct bulk reads.
 - `lavik_fullsync_reserved_memory_bytes`: reusable retained-memory headroom
   promised to active full-sync coverage maps.
 - `lavik_memory_admission_pending_bytes`: short-lived worker-local permits

include/lavik/memory.h

@@ -133,20 +133,21 @@ class RetainedMemoryCharge {
   std::size_t bytes_ = 0;
 };
 
-// Accounts request bytes read from ordinary client connections until their
-// command finishes. One instance belongs to one connection and must remain on
-// the worker where it was created; this keeps acquire/release operations local
-// to that worker's cache line.
+// Accounts received request bytes and prepaid direct-bulk destinations from
+// ordinary client connections until their command finishes or the connection
+// closes. One instance belongs to one connection and must remain on the worker
+// where it was created; acquire/release touches only that worker's cache line.
 class ClientBufferReservation {
  public:
   ClientBufferReservation() noexcept;
   ClientBufferReservation(const ClientBufferReservation&) = delete;
   ClientBufferReservation& operator=(const ClientBufferReservation&) = delete;
   ~ClientBufferReservation();
 
-  // Acquires bytes from the calling worker's share of the 5% client-buffer
-  // budget. The check happens once after a socket read, not for every parser
-  // append. Failure leaves the reservation unchanged.
+  // Acquires bytes from the calling worker's configured client-buffer share.
+  // Ordinary reads acquire received bytes; direct bulk reads acquire a bounded
+  // destination before allocation and do not recharge it as data arrives.
+  // Failure leaves the reservation unchanged.
   [[nodiscard]] bool TryAcquire(std::size_t bytes) noexcept;
   // Releases bytes after parser-discarded input or a completed command. The
   // caller must not release more than it has acquired.

include/lavik/resp.h

@@ -18,6 +18,7 @@
 
 #include <cstddef>
 #include <cstdint>
+#include <span>
 #include <string>
 #include <string_view>
 #include <utility>
@@ -64,6 +65,27 @@ class RespCommandParser {
 
   RespParseResult Parse(std::string_view input);
 
+  // Suggests bounded additional body storage for a partially received large
+  // bulk argument. Returns zero until payload arrives, while prepared storage
+  // remains, or when the current query limit leaves no allowance. The caller
+  // must admit this many request-buffer bytes before allocating them.
+  [[nodiscard]] std::size_t BulkReadReservationSize() const noexcept;
+  // Preallocates 1..BulkReadReservationSize() bytes in the final argument.
+  // This does not consume input or account wire bytes; unused space stays
+  // owned by the parser until filled or Reset. Invalid sizes leave it intact.
+  absl::Status PrepareBulkRead(std::size_t bytes);
+  // Returns the next bounded writable body window, excluding the terminator.
+  // Commit bytes written here instead of passing them to Parse. The span is
+  // invalidated by Parse, Reset, or another preparation; none may run while a
+  // transport read owns it. The caller retains its admission across the read.
+  // A lower query limit does not revoke this admitted span; Commit rechecks it.
+  [[nodiscard]] std::span<char> BulkReadBuffer() noexcept;
+  // Commits at most the exposed window's bytes and rechecks the latest query
+  // limit. Partial/zero commits are allowed while a prepared window remains;
+  // failure leaves logical parser state intact. Parse still consumes CRLF and
+  // returns completed commands. It can also fill a partially used window.
+  absl::Status CommitBulkRead(std::size_t bytes);
+
   void Reset();
   // Changes the guard without disturbing partially parsed state. If the new
   // limit is already below the retained command size, the next input byte is
@@ -97,6 +119,9 @@ class RespCommandParser {
   std::string current_argument_;
   std::size_t arguments_remaining_ = 0;
   std::size_t bulk_remaining_ = 0;
+  // String size includes this prepared but unwritten tail. Received payload
+  // ends before it; bulk_remaining_ includes it. Parse-only callers keep zero.
+  std::size_t bulk_prepared_ = 0;
   std::size_t terminator_bytes_ = 0;
   std::size_t command_bytes_ = 0;
   std::size_t query_buffer_limit_ = kDefaultClientQueryBufferLimit;

src/metrics.cpp

@@ -667,7 +667,7 @@ bycorf::Task<absl::Status> RenderPrometheusMetrics(
       "lavik_client_request_buffer_limit_bytes ",
       memory_metrics.client_buffer_limit_bytes_, "\n",
       "# HELP lavik_client_buffered_request_bytes Ordinary client request "
-      "bytes read but not yet retired.\n"
+      "bytes received or reserved for direct reads but not yet retired.\n"
       "# TYPE lavik_client_buffered_request_bytes gauge\n"
       "lavik_client_buffered_request_bytes ",
       memory_metrics.client_buffered_bytes_, "\n",

src/redis/resp.cpp

@@ -19,6 +19,7 @@
 #include <algorithm>
 #include <cassert>
 #include <charconv>
+#include <cstring>
 #include <limits>
 #include <new>
 #include <stdexcept>
@@ -36,10 +37,9 @@ namespace {
 constexpr long long kMaxArrayLen = std::numeric_limits<int>::max();
 constexpr std::size_t kInitialArgCapacity = 1024;
 constexpr std::size_t kMaxBulkLen = 512ULL * 1024 * 1024;
-// TODO(memory-control): Once retained-memory admission is settled, let bulk
-// arguments >= 32 KiB expose their final string storage as the ReadSome()
-// destination. This removes the RequestInputBuffer-to-string copy without
-// pinning a shared provided buffer across an await or a stalled client.
+constexpr std::size_t kDirectBulkThreshold = 32 * 1024;
+constexpr std::size_t kDirectBulkReservation = 1024 * 1024;
+constexpr std::size_t kDirectBulkReadBytes = 256 * 1024;
 constexpr std::size_t kMaxLengthTextBytes = 32;
 absl::Status AppendArgument(std::string* output, std::string_view value) {
   if (value.size() > std::numeric_limits<std::size_t>::max() - output->size()) {
@@ -168,12 +168,59 @@ void RespCommandParser::ResetCommand() {
   command_ = RespCommand{};
   arguments_remaining_ = 0;
   bulk_remaining_ = 0;
+  bulk_prepared_ = 0;
   terminator_bytes_ = 0;
   command_bytes_ = 0;
 }
 
 void RespCommandParser::Reset() { ResetCommand(); }
 
+std::size_t RespCommandParser::BulkReadReservationSize() const noexcept {
+  if (state_ != State::kBulkData || current_argument_.empty() ||
+      bulk_prepared_ != 0 ||
+      current_argument_.size() + bulk_remaining_ < kDirectBulkThreshold ||
+      command_bytes_ >= query_buffer_limit_)
+    return 0;
+  return std::min({bulk_remaining_, kDirectBulkReservation,
+                   query_buffer_limit_ - command_bytes_});
+}
+
+absl::Status RespCommandParser::PrepareBulkRead(std::size_t bytes) {
+  if (bytes == 0 || bytes > BulkReadReservationSize())
+    return absl::InvalidArgumentError("invalid RESP bulk read reservation");
+  // Resize makes the whole writable range valid string storage. Tracking its
+  // unread tail prevents initialized but unreceived bytes becoming payload.
+  // Allocation failure follows Parse's process-fatal allocation policy.
+  current_argument_.resize(current_argument_.size() + bytes);
+  bulk_prepared_ = bytes;
+  return absl::OkStatus();
+}
+
+std::span<char> RespCommandParser::BulkReadBuffer() noexcept {
+  if (state_ != State::kBulkData) return {};
+  const auto bytes = std::min(bulk_prepared_, kDirectBulkReadBytes);
+  return {current_argument_.data() + current_argument_.size() - bulk_prepared_,
+          bytes};
+}
+
+absl::Status RespCommandParser::CommitBulkRead(std::size_t bytes) {
+  if (state_ != State::kBulkData || bulk_prepared_ == 0 ||
+      bytes > std::min(bulk_prepared_, kDirectBulkReadBytes))
+    return absl::InvalidArgumentError("invalid RESP bulk read length");
+  // The query limit can change while a socket read is suspended. Recheck
+  // accounting before advancing any logical state, even for an earlier span.
+  if (!Account(bytes))
+    return absl::ResourceExhaustedError(
+        "client request exceeds the query buffer limit");
+  bulk_prepared_ -= bytes;
+  bulk_remaining_ -= bytes;
+  if (bulk_remaining_ == 0) {
+    terminator_bytes_ = 0;
+    state_ = State::kBulkTerminator;
+  }
+  return absl::OkStatus();
+}
+
 bool RespCommandParser::Account(std::size_t bytes) {
   if (bytes > query_buffer_limit_ ||
       command_bytes_ > query_buffer_limit_ - bytes) {
@@ -312,6 +359,7 @@ RespParseResult RespCommandParser::Parse(std::string_view input) {
                     absl::OutOfRangeError("RESP bulk string too large"), pos);
               }
               bulk_remaining_ = static_cast<std::size_t>(parsed);
+              assert(bulk_prepared_ == 0);
               current_argument_.clear();
               state_ = State::kBulkData;
             }
@@ -409,7 +457,18 @@ RespParseResult RespCommandParser::Parse(std::string_view input) {
           // length_error handler on every ordinary RESP argument is redundant.
           // Allocation failure remains process-fatal through the coroutine's
           // unhandled-exception policy.
-          current_argument_.append(input.substr(pos, take));
+          if (bulk_prepared_ == 0) [[likely]] {
+            current_argument_.append(input.substr(pos, take));
+          } else {
+            const auto prepared = std::min(take, bulk_prepared_);
+            std::memcpy(current_argument_.data() + current_argument_.size() -
+                            bulk_prepared_,
+                        input.data() + pos, prepared);
+            bulk_prepared_ -= prepared;
+            if (take > prepared)
+              current_argument_.append(
+                  input.substr(pos + prepared, take - prepared));
+          }
           bulk_remaining_ -= take;
           if (!consume(take)) {
             return Error(absl::ResourceExhaustedError(

src/redis/server.cpp

@@ -1044,6 +1044,8 @@ Task<absl::Status> ReadCommandBatch(
     const std::atomic<std::size_t>* query_buffer_limit,
     std::size_t* unassigned_input_bytes,
     std::optional<absl::Status>* deferred_error) {
+  constexpr std::size_t kBulkBytesPerYield = 256 * 1024;
+  std::size_t bulk_bytes_since_yield = 0;
   while (ready->empty()) {
     // CONFIG SET may run on another worker while this coroutine is suspended
     // in ReadSome(). Refresh once per parsing round; token-level loads would
@@ -1082,6 +1084,55 @@ Task<absl::Status> ReadCommandBatch(
     }
     if (!ready->empty()) co_return absl::OkStatus();
 
+    const std::size_t bulk_reservation = parser->BulkReadReservationSize();
+    if (bulk_reservation != 0) {
+      // A direct destination includes bytes the peer has not sent yet. Charge
+      // that bounded window before allocation, then let received bytes inherit
+      // its allowance instead of charging them again. The connection owns any
+      // unread remainder until teardown; completed commands retain the usual
+      // wire-byte charge through execution (or MULTI).
+      if (!client_buffers->TryAcquire(bulk_reservation)) [[unlikely]] {
+        RecordMemoryRejection();
+        co_return absl::ResourceExhaustedError(
+            "client request buffers exceed the memory limit");
+      }
+      const auto prepared = parser->PrepareBulkRead(bulk_reservation);
+      if (!prepared.ok()) [[unlikely]] {
+        client_buffers->Release(bulk_reservation);
+        co_return prepared;
+      }
+    }
+    if (auto body = parser->BulkReadBuffer(); !body.empty()) {
+      body = body.first(
+          std::min(body.size(), kBulkBytesPerYield - bulk_bytes_since_yield));
+      auto read_result =
+          co_await stream.ReadSomeBuffered(std::as_writable_bytes(body));
+      if (!read_result.ok()) [[unlikely]]
+        co_return read_result.status();
+      if (*read_result == 0) [[unlikely]] {
+        co_return absl::UnavailableError("peer closed connection");
+      }
+      // The limit may have shrunk while the direct read was suspended. Never
+      // dispatch a command using a stale guard merely because its destination
+      // was admitted before CONFIG SET ran.
+      parser->SetQueryBufferLimit(
+          query_buffer_limit->load(std::memory_order_acquire));
+      const auto committed = parser->CommitBulkRead(*read_result);
+      if (!committed.ok()) [[unlikely]]
+        co_return committed;
+      assert(*read_result <=
+             std::numeric_limits<std::size_t>::max() - *unassigned_input_bytes);
+      *unassigned_input_bytes += *read_result;
+      // Buffered reads can finish without suspending. Bound uninterrupted
+      // receive work so a large body cannot monopolize the connection worker.
+      bulk_bytes_since_yield += *read_result;
+      if (bulk_bytes_since_yield == kBulkBytesPerYield) {
+        bulk_bytes_since_yield = 0;
+        co_await Yield(*ThisWorker().self_);
+      }
+      continue;
+    }
+
     std::span<std::byte> buffer = input->AppendBuffer();
     auto read_result = co_await stream.ReadSome(buffer);
     if (!read_result.ok()) [[unlikely]] {

src/storage/engine/grouped/string.cpp

@@ -32,12 +32,16 @@ OrderedCollectionRoot StringRoot(std::size_t bytes, std::uint64_t incarnation) {
 
 OrderedGroupSnapshot StringPage(const OrderedCollectionRoot& root,
                                 std::uint64_t id, std::string bytes) {
-  return {.kind_ = OrderedCollectionKind::kString,
-          .incarnation_ = root.incarnation_,
-          .id_ = id,
-          .previous_ = id - 1,
-          .next_ = id == root.last_group_ ? 0 : id + 1,
-          .entries_ = {{.value_ = std::move(bytes)}}};
+  OrderedGroupSnapshot page{.kind_ = OrderedCollectionKind::kString,
+                            .incarnation_ = root.incarnation_,
+                            .id_ = id,
+                            .previous_ = id - 1,
+                            .next_ = id == root.last_group_ ? 0 : id + 1,
+                            .entries_ = {}};
+  // An initializer_list exposes const entries, forcing the vector to copy
+  // the segment even when its initializer moves the string.
+  page.entries_.emplace_back().value_ = std::move(bytes);
+  return page;
 }
 
 }  // namespace