diff --git a/CMakeLists.txt b/CMakeLists.txt index 7364d9d..7407857 100644 --- a/CMakeLists.txt +++ b/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" diff --git a/README.md b/README.md index 1655633..2163ffc 100644 --- a/README.md +++ b/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. diff --git a/include/bycorf/net/tcp_stream.h b/include/bycorf/net/tcp_stream.h index a645b7f..12a495e 100644 --- a/include/bycorf/net/tcp_stream.h +++ b/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 buffer) noexcept; + ReadOperation(Connection* connection, std::span 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 buffer_; std::optional immediate_status_; std::optional immediate_result_; + bool drain_buffered_ = false; }; class WriteOperation final : public IoCompletion { @@ -199,6 +201,14 @@ class TcpStream { absl::Status SetReadAhead(bool enabled) noexcept; Task> ReadSome(std::span 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> ReadSomeBuffered( + std::span buffer); Task> WriteSome( std::span buffer); Task> WriteSomeV(std::span buffers); diff --git a/src/net/tcp_stream.cpp b/src/net/tcp_stream.cpp index 94f605a..a59ae3a 100644 --- a/src/net/tcp_stream.cpp +++ b/src/net/tcp_stream.cpp @@ -133,8 +133,11 @@ class CancelRecvOperation final : public IoCompletion { } // namespace ReadOperation::ReadOperation(Connection* connection, - std::span buffer) noexcept - : connection_(connection), buffer_(buffer) {} + std::span 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 ReadOperation::await_resume() noexcept { "no received data available"); } - auto& received = connection_->received_buffers_.front(); - const std::size_t available = - static_cast(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(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(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(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> TcpStream::ReadSome( co_return co_await ReadRawSome(buffer); } +Task> TcpStream::ReadSomeBuffered( + std::span 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 buffer) noexcept { return ReadOperation(connection_, buffer); } diff --git a/tests/tcp_buffered_read_check.cpp b/tests/tcp_buffered_read_check.cpp new file mode 100644 index 0000000..dba06f8 --- /dev/null +++ b/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 +#include + +#include +#include +#include +#include +#include +#include +#include +#include + +#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 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 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(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 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(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(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(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::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(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(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(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(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 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(); +}