perf(net): add opt-in buffered stream reads

CMakeLists.txt

@@ -423,6 +423,9 @@ if (BYCORF_BUILD_EXAMPLES OR BYCORF_BUILD_TESTS)
   add_executable(bycorf_connection_storage_check example/connection_storage_check.cpp)
   target_link_libraries(bycorf_connection_storage_check PRIVATE bycorf::core)
 
+  add_executable(bycorf_tcp_buffered_read_check tests/tcp_buffered_read_check.cpp)
+  target_link_libraries(bycorf_tcp_buffered_read_check PRIVATE bycorf::core)
+
   add_executable(bycorf_backend_selection_check example/backend_selection_check.cpp)
   target_link_libraries(bycorf_backend_selection_check PRIVATE bycorf::core)
 
@@ -506,8 +509,10 @@ if (BYCORF_BUILD_TESTS)
   endif()
   add_test(NAME bycorf_connect_timer_check COMMAND bycorf_connect_timer_check)
   add_test(NAME bycorf_connection_storage_check COMMAND bycorf_connection_storage_check)
+  add_test(NAME bycorf_tcp_buffered_read_check COMMAND bycorf_tcp_buffered_read_check)
   add_test(NAME bycorf_backend_selection_check COMMAND bycorf_backend_selection_check)
   set_tests_properties(bycorf_connection_storage_check bycorf_backend_selection_check
+    bycorf_tcp_buffered_read_check
     PROPERTIES TIMEOUT 90 LABELS "runtime;net;io_uring")
   set_tests_properties(bycorf_connect_timer_check PROPERTIES
     TIMEOUT 90 LABELS "runtime;net;io_uring"

README.md

@@ -99,11 +99,13 @@ a 30-second timeout; the other checks below have 90-second timeouts:
 | `bycorf_task_transfer_check` | Bounded native stack for sequential and nested awaits, asynchronous completion, cancellation ownership and reentrant callbacks |
 | `bycorf_connect_timer_check` | Early shutdown, timers and cancellation, loopback TCP, refused connections, deadlines, address validation, and retirement of losing deadlines |
 | `bycorf_connection_storage_check` | Connection storage lifetime while suspended coroutines still borrow it |
+| `bycorf_tcp_buffered_read_check` | Bounded reads across queued receive buffers, partial buffers, unchanged ordinary reads, waiting, errors, and EOF |
 | `bycorf_backend_selection_check` | Unsupported backend requests, inactive optional backends, and immutable allocator/backend selection |
 
-Network checks use one unpinned worker and ephemeral loopback ports. The
-connection-timeout probe warns if the host rejects its unreachable destination
-immediately, since that exercises connection failure instead of the deadline.
+Network checks use one unpinned worker and ephemeral loopback ports or Unix
+socket pairs. The connection-timeout probe warns if the host rejects its
+unreachable destination immediately, since that exercises connection failure
+instead of the deadline.
 These software tests also run in a bypass-capable build with the default
 kernel/io_uring selection, without activating EAL or accessing devices.
 

include/bycorf/net/tcp_stream.h

@@ -45,7 +45,8 @@ class TlsState;
 // so they live directly in the awaiting coroutine frame.
 class ReadOperation final : public IoCompletion {
  public:
-  ReadOperation(Connection* connection, std::span<std::byte> buffer) noexcept;
+  ReadOperation(Connection* connection, std::span<std::byte> buffer,
+                bool drain_buffered = false) noexcept;
   ReadOperation(const ReadOperation&) = delete;
   ReadOperation& operator=(const ReadOperation&) = delete;
   ReadOperation(ReadOperation&&) = delete;
@@ -61,6 +62,7 @@ class ReadOperation final : public IoCompletion {
   std::span<std::byte> buffer_;
   std::optional<absl::Status> immediate_status_;
   std::optional<std::size_t> immediate_result_;
+  bool drain_buffered_ = false;
 };
 
 class WriteOperation final : public IoCompletion {
@@ -199,6 +201,14 @@ class TcpStream {
   absl::Status SetReadAhead(bool enabled) noexcept;
 
   Task<absl::StatusOr<std::size_t>> ReadSome(std::span<std::byte> buffer);
+  // Wait for initial data, then copy currently queued receive buffers up to
+  // buffer.size(). Never wait to fill the destination; its size bounds work
+  // in this call, so callers should limit it for scheduling fairness. TLS
+  // retains ReadSome's decrypted-read behavior without draining raw buffers.
+  // EOF returns zero after queued bytes drain. An error after partial copying
+  // is reported on the next read, preserving the successful byte prefix.
+  Task<absl::StatusOr<std::size_t>> ReadSomeBuffered(
+      std::span<std::byte> buffer);
   Task<absl::StatusOr<std::size_t>> WriteSome(
       std::span<const std::byte> buffer);
   Task<absl::StatusOr<std::size_t>> WriteSomeV(std::span<const iovec> buffers);

src/net/tcp_stream.cpp

@@ -133,8 +133,11 @@ class CancelRecvOperation final : public IoCompletion {
 }  // namespace
 
 ReadOperation::ReadOperation(Connection* connection,
-                             std::span<std::byte> buffer) noexcept
-    : connection_(connection), buffer_(buffer) {}
+                             std::span<std::byte> buffer,
+                             bool drain_buffered) noexcept
+    : connection_(connection),
+      buffer_(buffer),
+      drain_buffered_(drain_buffered) {}
 
 bool ReadOperation::await_suspend(std::coroutine_handle<> awaiting) {
   awaiting_ = awaiting;
@@ -195,25 +198,36 @@ absl::StatusOr<std::size_t> ReadOperation::await_resume() noexcept {
                         "no received data available");
   }
 
-  auto& received = connection_->received_buffers_.front();
-  const std::size_t available =
-      static_cast<std::size_t>(received.size_ - received.offset_);
-  const std::size_t to_copy = std::min(buffer_.size(), available);
-  auto chunk = connection_->worker_->ViewMultishotBuffer(
-      connection_, received.buffer_id_, received.offset_, to_copy);
-  if (chunk.size() != to_copy) {
-    return absl::Status(absl::StatusCode::kInternal,
-                        "invalid multishot buffer view");
-  }
+  std::size_t copied = 0;
+  do {
+    auto& received = connection_->received_buffers_.front();
+    const std::size_t available =
+        static_cast<std::size_t>(received.size_ - received.offset_);
+    const std::size_t to_copy = std::min(buffer_.size() - copied, available);
+    auto chunk = connection_->worker_->ViewMultishotBuffer(
+        connection_, received.buffer_id_, received.offset_, to_copy);
+    if (chunk.size() != to_copy) {
+      auto status = absl::InternalError("invalid multishot buffer view");
+      if (copied == 0) return status;
+      // Earlier buffers have already been consumed. Return that valid prefix
+      // before reporting the error; otherwise the caller would lose bytes.
+      connection_->last_error_ = std::move(status);
+      return copied;
+    }
 
-  std::memcpy(buffer_.data(), chunk.data(), chunk.size());
-  received.offset_ += static_cast<std::uint32_t>(to_copy);
-  if (received.offset_ == received.size_) {
-    const auto buffer_id = received.buffer_id_;
-    connection_->received_buffers_.pop_front();
-    connection_->worker_->ReleaseReceivedBuffer(connection_, buffer_id);
-  }
-  return to_copy;
+    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_) {
+      const auto buffer_id = received.buffer_id_;
+      connection_->received_buffers_.pop_front();
+      connection_->worker_->ReleaseReceivedBuffer(connection_, buffer_id);
+    }
+    // Do not poll or suspend between buffers. Only already queued data belongs
+    // to this read, and ordinary ReadSome retains its single-buffer behavior.
+  } while (drain_buffered_ && copied < buffer_.size() &&
+           !connection_->received_buffers_.empty());
+  return copied;
 }
 
 void ReadOperation::Complete(Worker& worker, int result, unsigned flags) {
@@ -500,6 +514,14 @@ Task<absl::StatusOr<std::size_t>> TcpStream::ReadSome(
   co_return co_await ReadRawSome(buffer);
 }
 
+Task<absl::StatusOr<std::size_t>> TcpStream::ReadSomeBuffered(
+    std::span<std::byte> buffer) {
+  if (tls_ != nullptr) {
+    co_return co_await tls_->ReadSome(*this, buffer);
+  }
+  co_return co_await ReadOperation(connection_, buffer, true);
+}
+
 ReadOperation TcpStream::ReadRawSome(std::span<std::byte> buffer) noexcept {
   return ReadOperation(connection_, buffer);
 }

tests/tcp_buffered_read_check.cpp

@@ -0,0 +1,240 @@
+/*
+ * 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.
+ */
+
+#include <sys/socket.h>
+#include <unistd.h>
+
+#include <array>
+#include <chrono>
+#include <cstddef>
+#include <cstring>
+#include <iostream>
+#include <limits>
+#include <string>
+#include <string_view>
+
+#include "absl/cleanup/cleanup.h"
+#include "bycorf/io/storage.h"
+#include "bycorf/net/tcp_stream.h"
+#include "bycorf/runtime/runtime.h"
+
+namespace {
+
+using namespace std::chrono_literals;
+
+absl::Status Check(bool passed, std::string_view message) {
+  // Keep the last completed assertion visible if CTest kills a stalled read.
+  std::cout << (passed ? "PASS " : "FAIL ") << message << std::endl;
+  return passed ? absl::OkStatus() : absl::InternalError(message);
+}
+
+absl::Status Send(int fd, std::string_view bytes) {
+  while (!bytes.empty()) {
+    const auto sent = send(fd, bytes.data(), bytes.size(), MSG_NOSIGNAL);
+    if (sent <= 0) return absl::InternalError("send failed");
+    bytes.remove_prefix(sent);
+  }
+  return absl::OkStatus();
+}
+
+std::size_t QueuedBytes(const bycorf::Connection& connection) {
+  std::size_t bytes = 0;
+  for (const auto& buffer : connection.received_buffers_)
+    bytes += buffer.size_ - buffer.offset_;
+  return bytes;
+}
+
+bycorf::Task<absl::Status> SendLater(bycorf::Worker& worker, int fd) {
+  auto waited = co_await bycorf::SleepFor(worker, 1ms);
+  if (!waited.ok()) co_return waited;
+  co_return Send(fd, "later");
+}
+
+bycorf::Task<absl::Status> CheckReads(bycorf::Worker& worker) {
+  int sockets[2];
+  if (socketpair(AF_UNIX, SOCK_STREAM | SOCK_NONBLOCK | SOCK_CLOEXEC, 0,
+                 sockets) != 0)
+    co_return absl::InternalError("socketpair failed");
+  absl::Cleanup close_peer = [&] { close(sockets[1]); };
+  bycorf::Connection initial;
+  initial.worker_ = &worker;
+  initial.file_.fd_ = sockets[0];
+  initial.closed_ = false;
+  auto* connection = worker.AddConnection(std::move(initial));
+  bycorf::TcpStream stream(connection);
+  auto storage = stream.BorrowStorage();
+  absl::Cleanup close_stream = [&] { (void)stream.Close(); };
+  std::string payload(32 * 1024, '\0');
+  for (std::size_t i = 0; i < payload.size(); ++i)
+    payload[i] = static_cast<char>(i * 31 + i / 257);
+  // Use distinct views of the one-shot backing store to make queue boundaries
+  // deterministic, including on kernels without provided-buffer support. The
+  // reads still use the real backend's checked views and recycling path. Live
+  // socket receives below separately cover suspension and terminal events.
+  connection->recv_mode_ = bycorf::RecvMode::kOneShot;
+  connection->read_buffer_.resize(payload.size());
+  std::memcpy(connection->read_buffer_.data(), payload.data(), payload.size());
+  for (std::uint32_t offset = 0; offset < payload.size(); offset += 4096) {
+    connection->received_buffers_.push_back(
+        {.buffer_id_ = 0, .size_ = offset + 4096, .offset_ = offset});
+  }
+
+  std::array<std::byte, 64 * 1024> output;
+  auto read = co_await stream.ReadSomeBuffered({});
+  auto status = Check(
+      read.ok() && *read == 0 && QueuedBytes(*connection) == payload.size(),
+      "empty destination preserves queued data");
+  if (!status.ok()) co_return status;
+
+  const auto first_size = connection->received_buffers_.front().size_;
+  const auto partial = first_size / 2;
+  read = co_await stream.ReadSome(std::span(output).first(partial));
+  status = Check(read.ok() && *read == partial,
+                 "ordinary read leaves a partial front buffer");
+  if (!status.ok()) co_return status;
+  std::string received(reinterpret_cast<const char*>(output.data()), *read);
+  const auto& queue = connection->received_buffers_;
+  const auto third_partial = (queue[2].size_ - queue[2].offset_) / 2;
+  const auto third_offset = queue[2].offset_ + third_partial;
+  const auto limit = queue[0].size_ - queue[0].offset_ + queue[1].size_ -
+                     queue[1].offset_ + third_partial;
+  read = co_await stream.ReadSomeBuffered(std::span(output).first(limit));
+  status =
+      Check(read.ok() && *read == limit &&
+                connection->received_buffers_.front().offset_ == third_offset,
+            "buffered read crosses buffers and stops within the last");
+  if (!status.ok()) co_return status;
+  received.append(reinterpret_cast<const char*>(output.data()), *read);
+
+  const auto front_remaining = connection->received_buffers_.front().size_ -
+                               connection->received_buffers_.front().offset_;
+  read = co_await stream.ReadSome(output);
+  status = Check(read.ok() && *read == front_remaining &&
+                     !connection->received_buffers_.empty(),
+                 "ordinary ReadSome still consumes only the front buffer");
+  if (!status.ok()) co_return status;
+  received.append(reinterpret_cast<const char*>(output.data()), *read);
+
+  const auto queued = QueuedBytes(*connection);
+  connection->last_error_ = absl::UnavailableError("injected receive error");
+  read = co_await stream.ReadSomeBuffered(output);
+  status = Check(absl::IsUnavailable(read.status()) &&
+                     QueuedBytes(*connection) == queued &&
+                     connection->last_error_.ok(),
+                 "pending receive error preserves queued data");
+  if (!status.ok()) co_return status;
+
+  // A later invalid view must not discard the successfully consumed prefix.
+  // Restore the real buffer identity after observing the deferred error.
+  status = Check(connection->received_buffers_.size() >= 2,
+                 "multiple remaining buffers for partial-error check");
+  if (!status.ok()) co_return status;
+  auto& invalid_buffer = connection->received_buffers_[1];
+  const auto saved_id = invalid_buffer.buffer_id_;
+  invalid_buffer.buffer_id_ = std::numeric_limits<std::uint16_t>::max();
+  const auto valid_prefix = connection->received_buffers_[0].size_ -
+                            connection->received_buffers_[0].offset_;
+  read = co_await stream.ReadSomeBuffered(output);
+  const bool prefix_ok = read.ok() && *read == valid_prefix;
+  if (prefix_ok)
+    received.append(reinterpret_cast<const char*>(output.data()), *read);
+  auto failed = co_await stream.ReadSome(output);
+  invalid_buffer.buffer_id_ = saved_id;
+  status = Check(prefix_ok && absl::IsInternal(failed.status()),
+                 "partial progress is returned before a later buffer error");
+  if (!status.ok()) co_return status;
+
+  // Deterministically cover data and EOF already queued together without
+  // relying on the kernel to deliver both completions in a particular batch.
+  connection->recv_eof_ = true;
+  read = co_await stream.ReadSomeBuffered(output);
+  status = Check(read.ok() && *read == payload.size() - received.size() &&
+                     connection->received_buffers_.empty(),
+                 "queued data precedes EOF without filling output");
+  if (!status.ok()) co_return status;
+  received.append(reinterpret_cast<const char*>(output.data()), *read);
+  status =
+      Check(received == payload, "all bytes retain order across read modes");
+  if (!status.ok()) co_return status;
+  read = co_await stream.ReadSomeBuffered(output);
+  status = Check(read.ok() && *read == 0,
+                 "queued EOF returns zero after buffered data drains");
+  if (!status.ok()) co_return status;
+
+  connection->recv_eof_ = false;
+  connection->recv_mode_ = bycorf::RecvMode::kMultishot;
+  connection->read_buffer_.clear();
+  // No bytes are queued now: the first read must suspend, but once the peer
+  // sends five bytes it must return without waiting for the rest of output.
+  worker.Spawn(SendLater(worker, sockets[1]));
+  read = co_await stream.ReadSomeBuffered(output);
+  status =
+      Check(read.ok() && *read == 5 &&
+                std::string_view(reinterpret_cast<const char*>(output.data()),
+                                 *read) == "later",
+            "buffered read waits only for initial data");
+  if (!status.ok()) co_return status;
+
+  status = Send(sockets[1], "before eof");
+  if (!status.ok()) co_return status;
+  read = co_await stream.ReadSomeBuffered(output);
+  status =
+      Check(read.ok() && *read == 10 &&
+                std::string_view(reinterpret_cast<const char*>(output.data()),
+                                 *read) == "before eof",
+            "live receive returns the peer payload");
+  if (!status.ok()) co_return status;
+  // Older kernels can lose the AF_UNIX multishot EOF wake when send and
+  // shutdown run back-to-back (Linux fix a68ed2df7213). Consume the payload
+  // first so this tests live EOF independently of that kernel race; queued
+  // data plus EOF ordering is covered explicitly above.
+  if (shutdown(sockets[1], SHUT_WR) != 0)
+    co_return absl::InternalError("peer shutdown failed");
+  read = co_await stream.ReadSomeBuffered(output);
+  co_return Check(read.ok() && *read == 0,
+                  "live EOF returns zero after the queue drains");
+}
+
+bycorf::Task<absl::Status> RunChecks(bycorf::Worker& worker, bool& passed) {
+  const auto status = co_await CheckReads(worker);
+  passed = status.ok();
+  if (!status.ok()) std::cerr << status << '\n';
+  worker.RequestStop();
+  co_return status;
+}
+
+}  // namespace
+
+int main() {
+  bycorf::Runtime runtime;
+  runtime.Start(
+      1,
+      [](unsigned, bycorf::Worker& worker) {
+        auto status = worker.Init();
+        if (!status.ok()) {
+          std::cerr << status << '\n';
+          return 1;
+        }
+        bool passed = false;
+        worker.Spawn(RunChecks(worker, passed));
+        worker.Run();
+        worker.Shutdown();
+        return passed ? 0 : 1;
+      },
+      false);
+  runtime.WaitUntilStopped();
+  return runtime.exit_code();
+}