Use a resizable receive ring with one inline descriptor

CMakeLists.txt

@@ -407,6 +407,9 @@ if (BYCORF_BUILD_EXAMPLES)
 endif()
 
 if (BYCORF_BUILD_EXAMPLES OR BYCORF_BUILD_TESTS)
+  add_executable(bycorf_received_buffer_queue_check tests/received_buffer_queue_check.cpp)
+  target_link_libraries(bycorf_received_buffer_queue_check PRIVATE bycorf::core)
+
   add_executable(bycorf_task_transfer_check tests/task_transfer_check.cpp)
   target_include_directories(bycorf_task_transfer_check PRIVATE include)
   # Exercise the public compile policy with this build's actual optimization
@@ -487,6 +490,9 @@ if (BYCORF_BUILD_TESTS)
     endforeach()
   endif()
   enable_testing()
+  add_test(NAME bycorf_received_buffer_queue_check COMMAND bycorf_received_buffer_queue_check)
+  set_tests_properties(bycorf_received_buffer_queue_check PROPERTIES
+    TIMEOUT 30 LABELS "runtime;net")
   add_test(NAME bycorf_task_transfer_check COMMAND bycorf_task_transfer_check)
   set_tests_properties(bycorf_task_transfer_check PROPERTIES
     TIMEOUT 30 LABELS "runtime;task")

include/bycorf/base/inline_ring_queue.h

@@ -0,0 +1,238 @@
+/*
+ * Copyright (C) 2026 EloqData Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#ifndef BYCORF_BASE_INLINE_RING_QUEUE_H_
+#define BYCORF_BASE_INLINE_RING_QUEUE_H_
+
+#include <algorithm>
+#include <cassert>
+#include <cstdint>
+#include <memory>
+#include <new>
+#include <type_traits>
+#include <utility>
+
+namespace bycorf {
+
+// Single-threaded resizable FIFO for trivially copyable values. InlineCapacity
+// must be a positive power of two; heap storage starts at max(8, 2 * inline).
+// Push/pop and moves may relocate elements and invalidate references. This
+// container does not manage resources referenced by its values.
+template <typename T, std::uint32_t InlineCapacity>
+class InlineRingQueue {
+  static_assert(
+      std::is_trivially_copyable_v<T> && std::is_trivially_destructible_v<T> &&
+          std::is_nothrow_copy_constructible_v<T>,
+      "InlineRingQueue requires trivial values with nonthrowing copy");
+  static_assert(InlineCapacity > 0 &&
+                    (InlineCapacity & (InlineCapacity - 1)) == 0 &&
+                    InlineCapacity <= (1u << 29),
+                "InlineCapacity must be a power of two in [1, 2^29]");
+
+ public:
+  // User-provided so value-initializing the queue does not zero the unused
+  // inline slots before their first values are constructed.
+  InlineRingQueue() noexcept {}
+  ~InlineRingQueue() {
+    if (capacity_ > kInlineCapacity) delete[] storage_.data_;
+  }
+  InlineRingQueue(const InlineRingQueue&) = delete;
+  InlineRingQueue& operator=(const InlineRingQueue&) = delete;
+  InlineRingQueue(InlineRingQueue&& other) noexcept {
+    *this = std::move(other);
+  }
+  InlineRingQueue& operator=(InlineRingQueue&& other) noexcept {
+    if (this != &other) {
+      if (other.capacity_ == kInlineCapacity) {
+        // Inline elements belong to the queue object and must be copied before
+        // the moved-from queue can be reused or destroyed.
+        if (capacity_ > kInlineCapacity) {
+          delete[] storage_.data_;
+          std::construct_at(&storage_.inline_);
+        }
+        other.CopyTo(storage_.inline_.slots_);
+        head_ = 0;
+      } else {
+        if (capacity_ > kInlineCapacity) delete[] storage_.data_;
+        // Assignment starts the trivial pointer member's lifetime when needed.
+        storage_.data_ = other.storage_.data_;
+        head_ = other.head_;
+        std::construct_at(&other.storage_.inline_);
+      }
+      other.head_ = 0;
+      size_ = std::exchange(other.size_, 0);
+      capacity_ = std::exchange(other.capacity_, kInlineCapacity);
+      low_water_pops_ = std::exchange(other.low_water_pops_, 0);
+    }
+    return *this;
+  }
+
+  bool empty() const noexcept { return size_ == 0; }
+  std::uint32_t size() const noexcept { return size_; }
+  std::uint32_t capacity() const noexcept { return capacity_; }
+  T& operator[](std::uint32_t index) noexcept {
+    assert(index < size_);
+    return Storage()[(head_ + index) & (capacity_ - 1)].value_;
+  }
+  const T& operator[](std::uint32_t index) const noexcept {
+    assert(index < size_);
+    return Storage()[(head_ + index) & (capacity_ - 1)].value_;
+  }
+  T& front() noexcept {
+    assert(!empty());
+    return Storage()[head_].value_;
+  }
+  const T& front() const noexcept {
+    assert(!empty());
+    return Storage()[head_].value_;
+  }
+
+  void push_back(T value) {
+    if (size_ == capacity_) Grow();
+    std::construct_at(&Storage()[(head_ + size_) & (capacity_ - 1)].value_,
+                      value);
+    ++size_;
+    if (capacity_ > kInlineCapacity && size_ > LowWaterLimit())
+      low_water_pops_ = 0;
+  }
+
+  void pop_front() noexcept {
+    assert(!empty());
+    head_ = (head_ + 1) & (capacity_ - 1);
+    --size_;
+    // Inline storage cannot shrink; its low-water count stays zero.
+    if (capacity_ == kInlineCapacity) return;
+    if (size_ <= LowWaterLimit()) {
+      // Require a full capacity's worth of low-water pops before halving.
+      // Brief bursts and ordinary empty/nonempty alternation at the minimum
+      // capacity do not allocate. Shrinking is best-effort: reads and close
+      // cleanup must keep working even when allocation fails.
+      if (++low_water_pops_ == capacity_) Shrink();
+    } else {
+      low_water_pops_ = 0;
+    }
+  }
+
+  // Retains heap capacity. Callers must release any resources referenced by
+  // the values before discarding them.
+  void clear() noexcept { head_ = size_ = low_water_pops_ = 0; }
+
+ private:
+  // An empty slot has no active value. Construct only live values, both
+  // inline and on the heap; a union's empty constructor avoids
+  // value-initializing every unused slot during growth. Trivial values
+  // need no destruction when popped, cleared, overwritten, or released with the
+  // slot array.
+  union Slot {
+    T value_;
+    Slot() noexcept {}
+  };
+  static_assert(std::is_trivially_destructible_v<Slot>);
+
+  struct InlineStorage {
+    Slot slots_[InlineCapacity];
+    // User-provided to avoid zeroing unused inline slots on activation.
+    InlineStorage() noexcept {}
+  };
+  union StorageUnion {
+    InlineStorage inline_;
+    Slot* data_;
+    StorageUnion() noexcept : inline_() {}
+  };
+
+  Slot* Storage() noexcept {
+    return capacity_ == kInlineCapacity ? storage_.inline_.slots_
+                                        : storage_.data_;
+  }
+  const Slot* Storage() const noexcept {
+    return capacity_ == kInlineCapacity ? storage_.inline_.slots_
+                                        : storage_.data_;
+  }
+
+  std::uint32_t LowWaterLimit() const noexcept {
+    // The smallest heap allocation contracts directly to the inline slot.
+    // Only count a sustained depth that actually fits the next capacity.
+    return capacity_ == kMinHeapCapacity ? kInlineCapacity : capacity_ / 4;
+  }
+
+  // Keep allocation and value relocation out of inlined push/pop callers.
+  [[gnu::noinline]] void Grow() {
+    // Allocation failure leaves the source values and ownership intact.
+    if (capacity_ == kMaxCapacity) throw std::bad_alloc();
+    const auto capacity =
+        capacity_ == kInlineCapacity ? kMinHeapCapacity : capacity_ * 2;
+    auto* data = new Slot[capacity];
+    CopyTo(data);
+    // Copying cannot throw. Keep the old storage until all live values have
+    // been copied, then transfer sole ownership to the active pointer member.
+    if (capacity_ > kInlineCapacity) delete[] storage_.data_;
+    storage_.data_ = data;
+    head_ = 0;
+    capacity_ = capacity;
+    low_water_pops_ = 0;
+  }
+
+  [[gnu::noinline]] void Shrink() noexcept {
+    low_water_pops_ = 0;
+    const auto capacity =
+        capacity_ == kMinHeapCapacity ? kInlineCapacity : capacity_ / 2;
+    if (capacity == kInlineCapacity) {
+      // Save the owned array before inline slots overwrite the pointer.
+      // CopyFrom uses the old head/capacity and cannot throw.
+      auto* data = storage_.data_;
+      std::construct_at(&storage_.inline_);
+      CopyFrom(data, storage_.inline_.slots_);
+      delete[] data;
+    } else {
+      auto* data = new (std::nothrow) Slot[capacity];
+      if (!data) return;
+      CopyTo(data);
+      delete[] storage_.data_;
+      storage_.data_ = data;
+    }
+    head_ = 0;
+    capacity_ = capacity;
+  }
+
+  void CopyTo(Slot* destination) const noexcept {
+    CopyFrom(Storage(), destination);
+  }
+
+  void CopyFrom(const Slot* source, Slot* destination) const noexcept {
+    const auto first = std::min(size_, capacity_ - head_);
+    for (std::uint32_t i = 0; i < first; ++i)
+      std::construct_at(&destination[i].value_, source[head_ + i].value_);
+    for (std::uint32_t i = first; i < size_; ++i)
+      std::construct_at(&destination[i].value_, source[i - first].value_);
+  }
+
+  static constexpr std::uint32_t kInlineCapacity = InlineCapacity;
+  static constexpr std::uint32_t kMinHeapCapacity =
+      std::max(8u, InlineCapacity * 2);
+  static constexpr std::uint32_t kMaxCapacity = 1u << 30;
+  // Inline values and heap ownership are mutually exclusive. capacity_ is the
+  // discriminator and the queue is the sole owner of data_ in heap mode.
+  // Every transition starts the new union member's lifetime. Slot construction
+  // and value copying cannot throw, so allocation is the only failure point.
+  StorageUnion storage_;
+  std::uint32_t head_ = 0;
+  std::uint32_t size_ = 0;
+  std::uint32_t capacity_ = kInlineCapacity;
+  std::uint32_t low_water_pops_ = 0;
+};
+
+}  // namespace bycorf
+#endif  // BYCORF_BASE_INLINE_RING_QUEUE_H_

include/bycorf/net/connection.h

@@ -20,13 +20,13 @@
 #include <coroutine>
 #include <cstdint>
 #include <cstdlib>
-#include <deque>
 #include <limits>
 #include <memory>
 #include <vector>
 
 #include "absl/cleanup/cleanup.h"
 #include "absl/status/statusor.h"
+#include "bycorf/base/inline_ring_queue.h"
 
 namespace bycorf {
 
@@ -73,7 +73,7 @@ struct Connection {
   // With 64-byte-aligned allocations, these writes touch one cache line
   // instead of two. This alignment depends on the allocator, not alignof;
   // recheck actual addresses and generated stores when changing the layout.
-  // Normal receive completion still writes the deque tail on another line,
+  // Normal receive completion still writes queue bookkeeping on another line,
   // so this does not establish fewer lines for the whole receive/GET path.
   // Smaller sizeof or fewer touched lines alone does not establish a speedup:
   // require correctness and paired throughput/tail-latency regression tests.
@@ -109,7 +109,15 @@ struct Connection {
   // reclaimed memory. Borrow and release run on the owning worker thread (or
   // before the first await that could observe retirement).
   std::uint32_t storage_borrows_ = 0;
-  std::deque<ReceivedBuffer> received_buffers_;
+  // Owner-worker FIFO of receive descriptors, not the network bytes themselves.
+  // Callers own recycling backend buffers, including before clear(). References
+  // must not survive push/pop or moving the queue.
+  // The inline descriptor and mutable counters occupy the first 32 bytes. In
+  // the audited x86 layout the queue starts at Connection+96, so these fit one
+  // cache line when the Connection is 64-byte aligned. The heap owner shares
+  // storage with the inline descriptor. Recheck allocator addresses and
+  // throughput/tails after layout changes.
+  InlineRingQueue<ReceivedBuffer, 1> received_buffers_;
 
   std::uint64_t id_ = 0;
   std::uint64_t generation_ = 0;

src/net/tcp_stream.cpp

@@ -217,11 +217,13 @@ absl::StatusOr<std::size_t> ReadOperation::await_resume() noexcept {
 
     std::memcpy(buffer_.data() + copied, chunk.data(), chunk.size());
     copied += to_copy;
-    received.offset_ += static_cast<std::uint32_t>(to_copy);
-    if (received.offset_ == received.size_) {
+    if (to_copy == available) {
+      // The descriptor is about to leave the queue; no offset store is needed.
       const auto buffer_id = received.buffer_id_;
       connection_->received_buffers_.pop_front();
       connection_->worker_->ReleaseReceivedBuffer(connection_, buffer_id);
+    } else {
+      received.offset_ += static_cast<std::uint32_t>(to_copy);
     }
     // Do not poll or suspend between buffers. Only already queued data belongs
     // to this read, and ordinary ReadSome retains its single-buffer behavior.

src/runtime/worker.cpp

@@ -790,11 +790,14 @@ void Worker::DiscardReceivedBuffers(Connection* connection) {
   if (connection == nullptr) {
     return;
   }
-  while (!connection->received_buffers_.empty()) {
-    auto received = connection->received_buffers_.front();
-    connection->received_buffers_.pop_front();
-    backend_.ReleaseRecvBuffer(connection, received.buffer_id_);
-  }
+  auto& received = connection->received_buffers_;
+  const auto count = received.size();
+  // ReleaseRecvBuffer only recycles backend storage; it does not mutate this
+  // queue or resume readers. Clear once to avoid shrinking while discarding.
+  for (std::uint32_t i = 0; i < count; ++i) {
+    backend_.ReleaseRecvBuffer(connection, received[i].buffer_id_);
+  }
+  received.clear();
 }
 
 void Worker::CheckIdleConnections() {