From 61fb7e94d2dc23d1cea28ace228f11b88006ec5b Mon Sep 17 00:00:00 2001 From: Jerry Zhao Date: Mon, 12 Oct 2026 08:17:38 +0000 Subject: [PATCH 1/6] Avoid child coroutine frames in TCP stream I/O --- CMakeLists.txt | 7 + include/bycorf/net/tcp_stream.h | 108 ++++++++++++-- src/io/io_uring_backend.cpp | 30 ++-- src/net/tcp_stream.cpp | 112 ++++++++++---- tests/stream_awaitable_check.cpp | 248 +++++++++++++++++++++++++++++++ 5 files changed, 448 insertions(+), 57 deletions(-) create mode 100644 tests/stream_awaitable_check.cpp diff --git a/CMakeLists.txt b/CMakeLists.txt index 7364d9d..b987805 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -443,6 +443,10 @@ if (BYCORF_BUILD_EXAMPLES OR BYCORF_BUILD_TESTS) endif() if (BYCORF_BUILD_TESTS) + find_package(Threads REQUIRED) + add_executable(bycorf_stream_awaitable_check tests/stream_awaitable_check.cpp) + target_link_libraries(bycorf_stream_awaitable_check PRIVATE bycorf::core Threads::Threads) + if (BYCORF_KERNEL_BYPASS) add_executable(bycorf_dpdk_adaptive_check tests/dpdk_adaptive_check.cpp) target_link_libraries(bycorf_dpdk_adaptive_check PRIVATE bycorf::core) @@ -481,6 +485,9 @@ if (BYCORF_BUILD_TESTS) endforeach() endif() enable_testing() + add_test(NAME bycorf_stream_awaitable_check COMMAND bycorf_stream_awaitable_check) + set_tests_properties(bycorf_stream_awaitable_check PROPERTIES + TIMEOUT 30 LABELS "runtime;net;io_uring") 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") diff --git a/include/bycorf/net/tcp_stream.h b/include/bycorf/net/tcp_stream.h index a645b7f..8709bfe 100644 --- a/include/bycorf/net/tcp_stream.h +++ b/include/bycorf/net/tcp_stream.h @@ -29,6 +29,8 @@ #include #include #include +#include +#include #include #include "absl/status/statusor.h" @@ -57,6 +59,14 @@ class ReadOperation final : public IoCompletion { void Complete(Worker& worker, int result, unsigned flags) override; private: + template + friend class StreamOperation; + + void BindAwaiting(std::coroutine_handle<> awaiting) noexcept { + awaiting_ = awaiting; + connection_->read_waiter_ = awaiting; + } + Connection* connection_ = nullptr; std::span buffer_; std::optional immediate_status_; @@ -65,8 +75,8 @@ class ReadOperation final : public IoCompletion { class WriteOperation final : public IoCompletion { public: - WriteOperation(Connection* connection, - std::span buffer) noexcept; + WriteOperation(Connection* connection, std::span buffer, + bool write_all = false) noexcept; WriteOperation(const WriteOperation&) = delete; WriteOperation& operator=(const WriteOperation&) = delete; WriteOperation(WriteOperation&&) = delete; @@ -78,15 +88,24 @@ class WriteOperation final : public IoCompletion { void Complete(Worker& worker, int result, unsigned flags) override; private: + template + friend class StreamOperation; + + void BindAwaiting(std::coroutine_handle<> awaiting) noexcept { + awaiting_ = awaiting; + } + Connection* connection_ = nullptr; std::span buffer_; int result_ = 0; + bool write_all_ = false; + std::size_t written_ = 0; }; class WriteVOperation final : public IoCompletion { public: - WriteVOperation(Connection* connection, - std::span buffers) noexcept; + WriteVOperation(Connection* connection, std::span buffers, + bool write_all = false); WriteVOperation(const WriteVOperation&) = delete; WriteVOperation& operator=(const WriteVOperation&) = delete; WriteVOperation(WriteVOperation&&) = delete; @@ -98,10 +117,63 @@ class WriteVOperation final : public IoCompletion { void Complete(Worker& worker, int result, unsigned flags) override; private: + template + friend class StreamOperation; + + void BindAwaiting(std::coroutine_handle<> awaiting) noexcept { + awaiting_ = awaiting; + } + Connection* connection_ = nullptr; std::span buffers_; msghdr message_{}; int result_ = 0; + bool write_all_ = false; + std::size_t written_ = 0; + // Full writes advance a private descriptor copy; payload storage is borrowed. + std::vector remaining_; +}; + +// Selects raw I/O without a child frame, or owns the TLS coroutine until its +// result has been consumed. Operations borrow the stream and payload through +// completion and must be awaited on the connection's worker. They are immovable +// because the backend may retain the raw operation's address as its completion +// tag. Raw I/O starts in await_ready so immediate results require no coroutine +// transfer (returning the caller's own handle can grow the native stack under +// sanitizers). The owner worker cannot dispatch the new operation between +// await_ready and await_suspend, which binds its continuation before yielding. +// TLS retains Task's symmetric transfer and child ownership. +template +class StreamOperation { + public: + template + explicit StreamOperation(Connection* connection, Args&&... args) + : raw_(std::in_place, connection, std::forward(args)...) {} + explicit StreamOperation(Task tls) : tls_(std::move(tls)) {} + + bool await_ready() { + if (!raw_) return tls_.done(); + return raw_->await_ready() || !raw_->await_suspend({}); + } + std::coroutine_handle<> await_suspend(std::coroutine_handle<> awaiting) { + if (raw_) { + raw_->BindAwaiting(awaiting); + return std::noop_coroutine(); + } + return std::move(tls_).operator co_await().await_suspend(awaiting); + } + Result await_resume() { + if (!raw_) return std::move(tls_).operator co_await().await_resume(); + if constexpr (std::is_same_v) { + return raw_->await_resume().status(); + } else { + return raw_->await_resume(); + } + } + + private: + std::optional raw_; + Task tls_; }; // Outbound async connect (IORING_OP_CONNECT) on a not-yet-registered fd, raced @@ -198,12 +270,24 @@ class TcpStream { // with their own bounded ingress queues. absl::Status SetReadAhead(bool enabled) noexcept; - Task> ReadSome(std::span buffer); - Task> WriteSome( - std::span buffer); - Task> WriteSomeV(std::span buffers); - Task WriteAll(std::span buffer); - Task WriteAllV(std::span buffers); + using ReadAwaitable = + StreamOperation>; + using WriteAwaitable = + StreamOperation>; + using WriteVAwaitable = + StreamOperation>; + using WriteAllAwaitable = StreamOperation; + using WriteAllVAwaitable = StreamOperation; + + // Awaitable byte-stream I/O. Buffers and the stream must outlive completion. + // Plaintext operations live in the caller's frame; TLS uses lazy Tasks. + // Some operations return the transferred byte count. All operations retry + // short writes and report an error if a nonempty write makes no progress. + ReadAwaitable ReadSome(std::span buffer); + WriteAwaitable WriteSome(std::span buffer); + WriteVAwaitable WriteSomeV(std::span buffers); + WriteAllAwaitable WriteAll(std::span buffer); + WriteAllVAwaitable WriteAllV(std::span buffers); Task StartTls(const std::shared_ptr& context, bool server, std::string_view peer_name = {}); @@ -232,7 +316,9 @@ class TcpStream { ReadOperation ReadRawSome(std::span buffer) noexcept; WriteOperation WriteRawSome(std::span buffer) noexcept; WriteVOperation WriteRawSomeV(std::span buffers) noexcept; - Task WriteRawAll(std::span buffer); + WriteAllAwaitable WriteRawAll(std::span buffer); + Task WriteAllTls(std::span buffer); + Task WriteAllVTls(std::span buffers); Connection* connection_ = nullptr; std::shared_ptr tls_; diff --git a/src/io/io_uring_backend.cpp b/src/io/io_uring_backend.cpp index 82f09b4..16566d8 100644 --- a/src/io/io_uring_backend.cpp +++ b/src/io/io_uring_backend.cpp @@ -800,18 +800,21 @@ void IoUringBackend::DispatchCqe(io_uring_cqe* cqe) { } bool IoUringBackend::ReapCompletions() { - unsigned head = 0; - io_uring_cqe* cqe = nullptr; - unsigned processed = 0; - io_uring_for_each_cqe(&ring_, head, cqe) { - DispatchCqe(cqe); - ++processed; - } - if (processed != 0) { - io_uring_cq_advance(&ring_, processed); - return true; + bool processed = false; + unsigned budget = io_uring_cq_ready(&ring_); + while (budget-- != 0) { + io_uring_cqe* cqe = nullptr; + if (io_uring_peek_cqe(&ring_, &cqe) != 0) break; + // A short-write callback can submit again and AcquireSqe may reap + // recursively when the ring is full. Retire this CQE before dispatch, and + // copy it because the kernel can reuse the slot during that submission. + // Peek afresh after each callback: reentry may have consumed later CQEs. + io_uring_cqe completion = *cqe; + io_uring_cqe_seen(&ring_, cqe); + DispatchCqe(&completion); + processed = true; } - return false; + return processed; } void IoUringBackend::DrainRecvRearm() { @@ -877,9 +880,8 @@ bool IoUringBackend::Wait(int timeout_ms) { spdlog::error("backend wait_cqe failed rc={}", rc); return false; } - DispatchCqe(cqe); - io_uring_cqe_seen(&ring_, cqe); - Poll(); // drain any other ready completions + re-arm recvs + // Poll retires CQEs before callbacks, including the one that woke this wait. + Poll(); return true; } diff --git a/src/net/tcp_stream.cpp b/src/net/tcp_stream.cpp index 94f605a..3bde8b7 100644 --- a/src/net/tcp_stream.cpp +++ b/src/net/tcp_stream.cpp @@ -223,8 +223,9 @@ void ReadOperation::Complete(Worker& worker, int result, unsigned flags) { } WriteOperation::WriteOperation(Connection* connection, - std::span buffer) noexcept - : connection_(connection), buffer_(buffer) {} + std::span buffer, + bool write_all) noexcept + : connection_(connection), buffer_(buffer), write_all_(write_all) {} bool WriteOperation::await_suspend(std::coroutine_handle<> awaiting) { awaiting_ = awaiting; @@ -256,7 +257,11 @@ bool WriteOperation::await_suspend(std::coroutine_handle<> awaiting) { } absl::StatusOr WriteOperation::await_resume() noexcept { - if (result_ >= 0) return static_cast(result_); + if (write_all_ && result_ == 0 && !buffer_.empty()) { + return absl::InternalError("WriteSome returned 0"); + } + if (result_ >= 0) + return write_all_ ? written_ : static_cast(result_); return ErrnoToStatus(-result_, "send failed"); } @@ -267,12 +272,32 @@ void WriteOperation::Complete(Worker& worker, int result, unsigned flags) { connection_->write_inflight_ = false; connection_->inflight_ops_ -= 1; } + if (write_all_ && result > 0) { + const auto bytes = static_cast(result); + if (bytes > buffer_.size()) { + result_ = -EIO; + } else { + written_ += bytes; + buffer_ = buffer_.subspan(bytes); + // Retire each send before submitting its successor, keeping the same + // completion tag alive. Only the final completion resumes the caller. + if (!buffer_.empty() && await_suspend(awaiting_)) return; + } + } worker.Enqueue(awaiting_); } WriteVOperation::WriteVOperation(Connection* connection, - std::span buffers) noexcept - : connection_(connection), buffers_(buffers) {} + std::span buffers, bool write_all) + : connection_(connection), buffers_(buffers), write_all_(write_all) { + if (write_all_) { + remaining_.reserve(buffers.size()); + for (const auto& buffer : buffers) { + if (buffer.iov_len != 0) remaining_.push_back(buffer); + } + buffers_ = remaining_; + } +} bool WriteVOperation::await_suspend(std::coroutine_handle<> awaiting) { awaiting_ = awaiting; @@ -306,7 +331,11 @@ bool WriteVOperation::await_suspend(std::coroutine_handle<> awaiting) { } absl::StatusOr WriteVOperation::await_resume() noexcept { - if (result_ >= 0) return static_cast(result_); + if (write_all_ && result_ == 0 && !buffers_.empty()) { + return absl::InternalError("WriteSomeV returned 0"); + } + if (result_ >= 0) + return write_all_ ? written_ : static_cast(result_); return ErrnoToStatus(-result_, "sendmsg failed"); } @@ -317,6 +346,22 @@ void WriteVOperation::Complete(Worker& worker, int result, unsigned flags) { connection_->write_inflight_ = false; connection_->inflight_ops_ -= 1; } + if (write_all_ && result > 0) { + auto bytes = static_cast(result); + written_ += bytes; + while (!buffers_.empty() && bytes >= buffers_.front().iov_len) { + bytes -= buffers_.front().iov_len; + buffers_ = buffers_.subspan(1); + } + if (!buffers_.empty()) { + auto& first = remaining_[remaining_.size() - buffers_.size()]; + first.iov_base = static_cast(first.iov_base) + bytes; + first.iov_len -= bytes; + if (await_suspend(awaiting_)) return; + } else if (bytes != 0) { + result_ = -EIO; + } + } worker.Enqueue(awaiting_); } @@ -492,24 +537,23 @@ absl::Status TcpStream::SetReadAhead(bool enabled) noexcept { return absl::OkStatus(); } -Task> TcpStream::ReadSome( - std::span buffer) { +TcpStream::ReadAwaitable TcpStream::ReadSome(std::span buffer) { if (tls_ != nullptr) { - co_return co_await tls_->ReadSome(*this, buffer); + return ReadAwaitable(tls_->ReadSome(*this, buffer)); } - co_return co_await ReadRawSome(buffer); + return ReadAwaitable(connection_, buffer); } ReadOperation TcpStream::ReadRawSome(std::span buffer) noexcept { return ReadOperation(connection_, buffer); } -Task> TcpStream::WriteSome( +TcpStream::WriteAwaitable TcpStream::WriteSome( std::span buffer) { if (tls_ != nullptr) { - co_return co_await tls_->WriteSome(*this, buffer); + return WriteAwaitable(tls_->WriteSome(*this, buffer)); } - co_return co_await WriteRawSome(buffer); + return WriteAwaitable(connection_, buffer); } WriteOperation TcpStream::WriteRawSome( @@ -517,19 +561,19 @@ WriteOperation TcpStream::WriteRawSome( return WriteOperation(connection_, buffer); } -Task> TcpStream::WriteSomeV( +TcpStream::WriteVAwaitable TcpStream::WriteSomeV( std::span buffers) { if (tls_ != nullptr) { - for (const iovec& buffer : buffers) { + for (const auto& buffer : buffers) { if (buffer.iov_len == 0) continue; - co_return co_await tls_->WriteSome( + return WriteVAwaitable(tls_->WriteSome( *this, std::span( - static_cast(buffer.iov_base), buffer.iov_len)); + static_cast(buffer.iov_base), buffer.iov_len))); } - co_return std::size_t{0}; + return WriteVAwaitable(connection_, std::span{}); } - co_return co_await WriteRawSomeV(buffers); + return WriteVAwaitable(connection_, buffers); } WriteVOperation TcpStream::WriteRawSomeV( @@ -537,20 +581,24 @@ WriteVOperation TcpStream::WriteRawSomeV( return WriteVOperation(connection_, buffers); } -Task TcpStream::WriteRawAll(std::span buffer) { - std::size_t written = 0; - while (written < buffer.size()) { - auto result = co_await WriteRawSome(buffer.subspan(written)); - if (!result.ok()) co_return result.status(); - if (*result == 0) { - co_return absl::InternalError("raw WriteSome returned 0"); - } - written += *result; - } - co_return absl::OkStatus(); +TcpStream::WriteAllAwaitable TcpStream::WriteRawAll( + std::span buffer) { + return WriteAllAwaitable(connection_, buffer, true); +} + +TcpStream::WriteAllAwaitable TcpStream::WriteAll( + std::span buffer) { + if (tls_ != nullptr) return WriteAllAwaitable(WriteAllTls(buffer)); + return WriteRawAll(buffer); +} + +TcpStream::WriteAllVAwaitable TcpStream::WriteAllV( + std::span buffers) { + if (tls_ != nullptr) return WriteAllVAwaitable(WriteAllVTls(buffers)); + return WriteAllVAwaitable(connection_, buffers, true); } -Task TcpStream::WriteAll(std::span buffer) { +Task TcpStream::WriteAllTls(std::span buffer) { std::size_t written = 0; while (written < buffer.size()) { auto result = co_await WriteSome(buffer.subspan(written)); @@ -566,7 +614,7 @@ Task TcpStream::WriteAll(std::span buffer) { co_return absl::OkStatus(); } -Task TcpStream::WriteAllV(std::span buffers) { +Task TcpStream::WriteAllVTls(std::span buffers) { std::vector remaining; remaining.reserve(buffers.size()); for (const iovec& buffer : buffers) { diff --git a/tests/stream_awaitable_check.cpp b/tests/stream_awaitable_check.cpp new file mode 100644 index 0000000..b413c72 --- /dev/null +++ b/tests/stream_awaitable_check.cpp @@ -0,0 +1,248 @@ +/* + * 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 "bycorf/io/io_uring_backend.h" +#include "bycorf/net/tcp_stream.h" +#include "bycorf/runtime/runtime.h" + +namespace { +using namespace std::chrono_literals; +using bycorf::Task; +using bycorf::TcpStream; +int failures = 0; + +void Check(bool passed, const char* message) { + std::cout << (passed ? "PASS " : "FAIL ") << message << '\n'; + if (!passed) ++failures; +} + +std::span Bytes(const std::string& value) { + return std::as_bytes(std::span(value)); +} + +// Model AcquireSqe's recursive reap directly, without depending on kernel +// timing to fill both rings. The current CQE must already be retired when a +// completion submits work and reenters polling. +void CheckCompletionReentry(bool wait) { + bycorf::Worker worker; + bycorf::IoUringBackend backend; + bycorf::IoBackendOptions options; + options.ring_entries_ = 8; + options.recv_buffer_count_ = 0; + const auto initialized = backend.Init(options, &worker, -1); + Check(initialized.ok(), "initialize completion reentry test"); + if (!initialized.ok()) return; + struct Completion final : bycorf::IoCompletion { + bycorf::IoUringBackend* backend = nullptr; + unsigned calls = 0; + void Complete(bycorf::Worker&, int, unsigned) override { + if (++calls == 1) backend->Poll(); + } + } completion; + completion.backend = &backend; + __kernel_timespec timeout{0, 1}; + Check(backend.SubmitTimeout(timeout, &completion).ok(), + "submit reentry timer"); + Check(backend.Submit().ok(), "flush reentry timer"); + if (wait) Check(backend.Wait(1000), "wait dispatches the reentry timer"); + const auto deadline = std::chrono::steady_clock::now() + 1s; + while (completion.calls == 0 && std::chrono::steady_clock::now() < deadline) { + // DEFER_TASKRUN needs an enter to publish timer task-work into the CQ. + (void)backend.Submit(); + backend.Poll(); + } + Check(completion.calls == 1, + "recursive polling dispatches each CQE exactly once"); + backend.Shutdown(); +} + +// A tiny send buffer and a delayed reader force the full-write operations to +// survive multiple completions while retaining their payload and iovec state. +Task CheckWrite(bycorf::Worker& worker, bool vectored, + bool disconnect) { + int sockets[2]; + if (socketpair(AF_UNIX, SOCK_STREAM | SOCK_CLOEXEC, 0, sockets) != 0) { + Check(false, "socketpair"); + co_return absl::InternalError("socketpair failed"); + } + const int capacity = 4096; + Check(setsockopt(sockets[0], SOL_SOCKET, SO_SNDBUF, &capacity, + sizeof(capacity)) == 0, + "small send buffer"); + Check(fcntl(sockets[0], F_SETFL, O_NONBLOCK) == 0, "nonblocking sender"); + bycorf::Connection initial; + initial.worker_ = &worker; + initial.closed_ = false; + initial.file_.fd_ = sockets[0]; + auto* connection = worker.AddConnection(std::move(initial)); + TcpStream stream(connection); + auto borrow = stream.BorrowStorage(); + + std::string payload(1024 * 1024, '\0'); + for (std::size_t i = 0; i < payload.size(); ++i) payload[i] = char(i % 251); + std::string received; + std::thread reader([&] { + std::this_thread::sleep_for(20ms); + std::array buffer; + while (received.size() < payload.size()) { + const auto n = read(sockets[1], buffer.data(), buffer.size()); + if (n < 0 && errno == EINTR) continue; + if (n <= 0) break; + received.append(buffer.data(), static_cast(n)); + if (disconnect) break; + } + close(sockets[1]); + }); + + absl::Status status; + if (vectored) { + // Include empty entries and cuts inside and across vector boundaries. + std::array vectors{ + {{nullptr, 0}, + {payload.data(), 13}, + {nullptr, 0}, + {payload.data() + 13, 65539}, + {payload.data() + 65552, payload.size() - 65552}}}; + const auto original = vectors; + status = co_await stream.WriteAllV(vectors); + bool unchanged = true; + for (std::size_t i = 0; i < vectors.size(); ++i) { + unchanged &= vectors[i].iov_base == original[i].iov_base && + vectors[i].iov_len == original[i].iov_len; + } + Check(unchanged, "WriteAllV preserves caller descriptors"); + } else { + status = co_await stream.WriteAll(Bytes(payload)); + } + // Wake the peer even if a regression returns before the payload is complete. + shutdown(sockets[0], SHUT_WR); + reader.join(); + Check(disconnect ? !status.ok() : status.ok(), + disconnect ? "disconnect during full write returns an error" + : "full write succeeds under backpressure"); + if (!disconnect) + Check(received == payload, "short writes preserve every byte"); + Check(!connection->write_inflight_ && connection->inflight_ops_ == 0, + "full write retires all sends"); + Check(stream.Close().ok(), "close after full write"); + co_return absl::OkStatus(); +} + +Task CheckStream(bycorf::Worker& worker) { + TcpStream unbound; + std::array buffer{}; + Check((co_await unbound.ReadSome({})).value() == 0, "empty read"); + Check((co_await unbound.WriteSome({})).value() == 0, "empty write"); + Check((co_await unbound.WriteSomeV({})).value() == 0, "empty vector write"); + Check((co_await unbound.WriteAll({})).ok(), "empty full write"); + std::array empty_vectors{}; + Check((co_await unbound.WriteAllV(empty_vectors)).ok(), + "all-empty full vector write"); + Check(!(co_await unbound.ReadSome(buffer)).ok(), + "unbound read fails immediately"); + Check(!(co_await unbound.WriteAll(buffer)).ok(), + "unbound full write fails immediately"); + iovec vector{buffer.data(), buffer.size()}; + Check(!(co_await unbound.WriteAllV(std::span(&vector, 1))).ok(), + "unbound full vector write fails immediately"); + + bool immediate_errors = true; + for (unsigned i = 0; i < 100000; ++i) { + immediate_errors &= !(co_await unbound.ReadSome(buffer)).ok(); + } + Check(immediate_errors, + "repeated immediate completion keeps native stack bounded"); + + int sockets[2]; + if (socketpair(AF_UNIX, SOCK_STREAM | SOCK_NONBLOCK | SOCK_CLOEXEC, 0, + sockets) == 0) { + bycorf::Connection initial; + initial.worker_ = &worker; + initial.closed_ = false; + initial.file_.fd_ = sockets[0]; + auto* connection = worker.AddConnection(std::move(initial)); + TcpStream stream(connection); + auto borrow = stream.BorrowStorage(); + Check(write(sockets[1], "abcdefgh", 8) == 8, "queue received bytes"); + shutdown(sockets[1], SHUT_WR); + std::size_t read_bytes = 0; + while (read_bytes < buffer.size()) { + auto result = + co_await stream.ReadSome(std::span(buffer).subspan(read_bytes)); + if (!result.ok() || *result == 0) break; + read_bytes += *result; + } + Check(read_bytes == 8 && std::string(reinterpret_cast(buffer.data()), + 8) == "abcdefgh", + "read resumes with the received bytes"); + auto eof = co_await stream.ReadSome(buffer); + Check(eof.ok() && *eof == 0, "read reports EOF"); + auto sent = co_await stream.WriteSome(buffer); + Check(sent.ok() && *sent == buffer.size(), + "WriteSome reports transferred bytes"); + std::array response{}; + Check(read(sockets[1], response.data(), response.size()) == 8, + "WriteSome reaches the peer"); + std::array vectors{ + {{buffer.data(), 3}, {nullptr, 0}, {buffer.data() + 3, 5}}}; + sent = co_await stream.WriteSomeV(vectors); + Check(sent.ok() && *sent == buffer.size(), + "WriteSomeV reports transferred bytes"); + Check(read(sockets[1], response.data(), response.size()) == 8, + "WriteSomeV reaches the peer"); + Check(stream.Close().ok(), "close read stream"); + Check(!(co_await stream.WriteSome(buffer)).ok(), + "closed write fails immediately"); + Check(!(co_await stream.ReadSome(buffer)).ok(), + "closed read fails immediately"); + close(sockets[1]); + } else { + Check(false, "read socketpair"); + } + for (bool vectored : {false, true}) { + for (bool disconnect : {false, true}) { + (void)co_await CheckWrite(worker, vectored, disconnect); + } + } + worker.RequestStop(); + co_return absl::OkStatus(); +} +} // namespace + +int main() { + std::signal(SIGPIPE, SIG_IGN); + CheckCompletionReentry(false); + CheckCompletionReentry(true); + bycorf::Runtime runtime; + runtime.Start(1, [](unsigned, bycorf::Worker& worker) { + if (!worker.Init().ok()) return 1; + worker.Spawn(CheckStream(worker)); + worker.Run(); + worker.Shutdown(); + return failures == 0 ? 0 : 1; + }); + runtime.WaitUntilStopped(); + return runtime.exit_code(); +} From 4c388338e26e655ef35a98b8aae799db3c2b250a Mon Sep 17 00:00:00 2001 From: Jerry Zhao Date: Mon, 12 Oct 2026 10:05:55 +0000 Subject: [PATCH 2/6] Limit awaitable conversion to single-operation stream wrappers --- include/bycorf/net/tcp_stream.h | 60 ++++++++-------------- src/io/io_uring_backend.cpp | 30 +++++------ src/net/tcp_stream.cpp | 87 +++++++------------------------- tests/stream_awaitable_check.cpp | 39 -------------- 4 files changed, 55 insertions(+), 161 deletions(-) diff --git a/include/bycorf/net/tcp_stream.h b/include/bycorf/net/tcp_stream.h index 8709bfe..cc9aef8 100644 --- a/include/bycorf/net/tcp_stream.h +++ b/include/bycorf/net/tcp_stream.h @@ -29,7 +29,6 @@ #include #include #include -#include #include #include @@ -59,7 +58,7 @@ class ReadOperation final : public IoCompletion { void Complete(Worker& worker, int result, unsigned flags) override; private: - template + template friend class StreamOperation; void BindAwaiting(std::coroutine_handle<> awaiting) noexcept { @@ -75,8 +74,8 @@ class ReadOperation final : public IoCompletion { class WriteOperation final : public IoCompletion { public: - WriteOperation(Connection* connection, std::span buffer, - bool write_all = false) noexcept; + WriteOperation(Connection* connection, + std::span buffer) noexcept; WriteOperation(const WriteOperation&) = delete; WriteOperation& operator=(const WriteOperation&) = delete; WriteOperation(WriteOperation&&) = delete; @@ -88,7 +87,7 @@ class WriteOperation final : public IoCompletion { void Complete(Worker& worker, int result, unsigned flags) override; private: - template + template friend class StreamOperation; void BindAwaiting(std::coroutine_handle<> awaiting) noexcept { @@ -98,14 +97,12 @@ class WriteOperation final : public IoCompletion { Connection* connection_ = nullptr; std::span buffer_; int result_ = 0; - bool write_all_ = false; - std::size_t written_ = 0; }; class WriteVOperation final : public IoCompletion { public: - WriteVOperation(Connection* connection, std::span buffers, - bool write_all = false); + WriteVOperation(Connection* connection, + std::span buffers) noexcept; WriteVOperation(const WriteVOperation&) = delete; WriteVOperation& operator=(const WriteVOperation&) = delete; WriteVOperation(WriteVOperation&&) = delete; @@ -117,7 +114,7 @@ class WriteVOperation final : public IoCompletion { void Complete(Worker& worker, int result, unsigned flags) override; private: - template + template friend class StreamOperation; void BindAwaiting(std::coroutine_handle<> awaiting) noexcept { @@ -128,10 +125,6 @@ class WriteVOperation final : public IoCompletion { std::span buffers_; msghdr message_{}; int result_ = 0; - bool write_all_ = false; - std::size_t written_ = 0; - // Full writes advance a private descriptor copy; payload storage is borrowed. - std::vector remaining_; }; // Selects raw I/O without a child frame, or owns the TLS coroutine until its @@ -143,9 +136,10 @@ class WriteVOperation final : public IoCompletion { // sanitizers). The owner worker cannot dispatch the new operation between // await_ready and await_suspend, which binds its continuation before yielding. // TLS retains Task's symmetric transfer and child ownership. -template +template class StreamOperation { public: + using Result = absl::StatusOr; template explicit StreamOperation(Connection* connection, Args&&... args) : raw_(std::in_place, connection, std::forward(args)...) {} @@ -164,11 +158,7 @@ class StreamOperation { } Result await_resume() { if (!raw_) return std::move(tls_).operator co_await().await_resume(); - if constexpr (std::is_same_v) { - return raw_->await_resume().status(); - } else { - return raw_->await_resume(); - } + return raw_->await_resume(); } private: @@ -270,24 +260,20 @@ class TcpStream { // with their own bounded ingress queues. absl::Status SetReadAhead(bool enabled) noexcept; - using ReadAwaitable = - StreamOperation>; - using WriteAwaitable = - StreamOperation>; - using WriteVAwaitable = - StreamOperation>; - using WriteAllAwaitable = StreamOperation; - using WriteAllVAwaitable = StreamOperation; - - // Awaitable byte-stream I/O. Buffers and the stream must outlive completion. - // Plaintext operations live in the caller's frame; TLS uses lazy Tasks. - // Some operations return the transferred byte count. All operations retry - // short writes and report an error if a nonempty write makes no progress. + using ReadAwaitable = StreamOperation; + using WriteAwaitable = StreamOperation; + using WriteVAwaitable = StreamOperation; + + // Single-operation byte-stream I/O. Buffers and the stream must outlive + // completion. Plaintext awaits use the caller's frame; TLS owns a lazy Task. ReadAwaitable ReadSome(std::span buffer); WriteAwaitable WriteSome(std::span buffer); WriteVAwaitable WriteSomeV(std::span buffers); - WriteAllAwaitable WriteAll(std::span buffer); - WriteAllVAwaitable WriteAllV(std::span buffers); + + // Full writes retain their coroutine loops: each completion schedules the + // writer before it checks the result and submits any remaining bytes. + Task WriteAll(std::span buffer); + Task WriteAllV(std::span buffers); Task StartTls(const std::shared_ptr& context, bool server, std::string_view peer_name = {}); @@ -316,9 +302,7 @@ class TcpStream { ReadOperation ReadRawSome(std::span buffer) noexcept; WriteOperation WriteRawSome(std::span buffer) noexcept; WriteVOperation WriteRawSomeV(std::span buffers) noexcept; - WriteAllAwaitable WriteRawAll(std::span buffer); - Task WriteAllTls(std::span buffer); - Task WriteAllVTls(std::span buffers); + Task WriteRawAll(std::span buffer); Connection* connection_ = nullptr; std::shared_ptr tls_; diff --git a/src/io/io_uring_backend.cpp b/src/io/io_uring_backend.cpp index 16566d8..82f09b4 100644 --- a/src/io/io_uring_backend.cpp +++ b/src/io/io_uring_backend.cpp @@ -800,21 +800,18 @@ void IoUringBackend::DispatchCqe(io_uring_cqe* cqe) { } bool IoUringBackend::ReapCompletions() { - bool processed = false; - unsigned budget = io_uring_cq_ready(&ring_); - while (budget-- != 0) { - io_uring_cqe* cqe = nullptr; - if (io_uring_peek_cqe(&ring_, &cqe) != 0) break; - // A short-write callback can submit again and AcquireSqe may reap - // recursively when the ring is full. Retire this CQE before dispatch, and - // copy it because the kernel can reuse the slot during that submission. - // Peek afresh after each callback: reentry may have consumed later CQEs. - io_uring_cqe completion = *cqe; - io_uring_cqe_seen(&ring_, cqe); - DispatchCqe(&completion); - processed = true; + unsigned head = 0; + io_uring_cqe* cqe = nullptr; + unsigned processed = 0; + io_uring_for_each_cqe(&ring_, head, cqe) { + DispatchCqe(cqe); + ++processed; } - return processed; + if (processed != 0) { + io_uring_cq_advance(&ring_, processed); + return true; + } + return false; } void IoUringBackend::DrainRecvRearm() { @@ -880,8 +877,9 @@ bool IoUringBackend::Wait(int timeout_ms) { spdlog::error("backend wait_cqe failed rc={}", rc); return false; } - // Poll retires CQEs before callbacks, including the one that woke this wait. - Poll(); + DispatchCqe(cqe); + io_uring_cqe_seen(&ring_, cqe); + Poll(); // drain any other ready completions + re-arm recvs return true; } diff --git a/src/net/tcp_stream.cpp b/src/net/tcp_stream.cpp index 3bde8b7..74cae07 100644 --- a/src/net/tcp_stream.cpp +++ b/src/net/tcp_stream.cpp @@ -223,9 +223,8 @@ void ReadOperation::Complete(Worker& worker, int result, unsigned flags) { } WriteOperation::WriteOperation(Connection* connection, - std::span buffer, - bool write_all) noexcept - : connection_(connection), buffer_(buffer), write_all_(write_all) {} + std::span buffer) noexcept + : connection_(connection), buffer_(buffer) {} bool WriteOperation::await_suspend(std::coroutine_handle<> awaiting) { awaiting_ = awaiting; @@ -257,11 +256,7 @@ bool WriteOperation::await_suspend(std::coroutine_handle<> awaiting) { } absl::StatusOr WriteOperation::await_resume() noexcept { - if (write_all_ && result_ == 0 && !buffer_.empty()) { - return absl::InternalError("WriteSome returned 0"); - } - if (result_ >= 0) - return write_all_ ? written_ : static_cast(result_); + if (result_ >= 0) return static_cast(result_); return ErrnoToStatus(-result_, "send failed"); } @@ -272,32 +267,12 @@ void WriteOperation::Complete(Worker& worker, int result, unsigned flags) { connection_->write_inflight_ = false; connection_->inflight_ops_ -= 1; } - if (write_all_ && result > 0) { - const auto bytes = static_cast(result); - if (bytes > buffer_.size()) { - result_ = -EIO; - } else { - written_ += bytes; - buffer_ = buffer_.subspan(bytes); - // Retire each send before submitting its successor, keeping the same - // completion tag alive. Only the final completion resumes the caller. - if (!buffer_.empty() && await_suspend(awaiting_)) return; - } - } worker.Enqueue(awaiting_); } WriteVOperation::WriteVOperation(Connection* connection, - std::span buffers, bool write_all) - : connection_(connection), buffers_(buffers), write_all_(write_all) { - if (write_all_) { - remaining_.reserve(buffers.size()); - for (const auto& buffer : buffers) { - if (buffer.iov_len != 0) remaining_.push_back(buffer); - } - buffers_ = remaining_; - } -} + std::span buffers) noexcept + : connection_(connection), buffers_(buffers) {} bool WriteVOperation::await_suspend(std::coroutine_handle<> awaiting) { awaiting_ = awaiting; @@ -331,11 +306,7 @@ bool WriteVOperation::await_suspend(std::coroutine_handle<> awaiting) { } absl::StatusOr WriteVOperation::await_resume() noexcept { - if (write_all_ && result_ == 0 && !buffers_.empty()) { - return absl::InternalError("WriteSomeV returned 0"); - } - if (result_ >= 0) - return write_all_ ? written_ : static_cast(result_); + if (result_ >= 0) return static_cast(result_); return ErrnoToStatus(-result_, "sendmsg failed"); } @@ -346,22 +317,6 @@ void WriteVOperation::Complete(Worker& worker, int result, unsigned flags) { connection_->write_inflight_ = false; connection_->inflight_ops_ -= 1; } - if (write_all_ && result > 0) { - auto bytes = static_cast(result); - written_ += bytes; - while (!buffers_.empty() && bytes >= buffers_.front().iov_len) { - bytes -= buffers_.front().iov_len; - buffers_ = buffers_.subspan(1); - } - if (!buffers_.empty()) { - auto& first = remaining_[remaining_.size() - buffers_.size()]; - first.iov_base = static_cast(first.iov_base) + bytes; - first.iov_len -= bytes; - if (await_suspend(awaiting_)) return; - } else if (bytes != 0) { - result_ = -EIO; - } - } worker.Enqueue(awaiting_); } @@ -581,24 +536,20 @@ WriteVOperation TcpStream::WriteRawSomeV( return WriteVOperation(connection_, buffers); } -TcpStream::WriteAllAwaitable TcpStream::WriteRawAll( - std::span buffer) { - return WriteAllAwaitable(connection_, buffer, true); -} - -TcpStream::WriteAllAwaitable TcpStream::WriteAll( - std::span buffer) { - if (tls_ != nullptr) return WriteAllAwaitable(WriteAllTls(buffer)); - return WriteRawAll(buffer); -} - -TcpStream::WriteAllVAwaitable TcpStream::WriteAllV( - std::span buffers) { - if (tls_ != nullptr) return WriteAllVAwaitable(WriteAllVTls(buffers)); - return WriteAllVAwaitable(connection_, buffers, true); +Task TcpStream::WriteRawAll(std::span buffer) { + std::size_t written = 0; + while (written < buffer.size()) { + auto result = co_await WriteRawSome(buffer.subspan(written)); + if (!result.ok()) co_return result.status(); + if (*result == 0) { + co_return absl::InternalError("raw WriteSome returned 0"); + } + written += *result; + } + co_return absl::OkStatus(); } -Task TcpStream::WriteAllTls(std::span buffer) { +Task TcpStream::WriteAll(std::span buffer) { std::size_t written = 0; while (written < buffer.size()) { auto result = co_await WriteSome(buffer.subspan(written)); @@ -614,7 +565,7 @@ Task TcpStream::WriteAllTls(std::span buffer) { co_return absl::OkStatus(); } -Task TcpStream::WriteAllVTls(std::span buffers) { +Task TcpStream::WriteAllV(std::span buffers) { std::vector remaining; remaining.reserve(buffers.size()); for (const iovec& buffer : buffers) { diff --git a/tests/stream_awaitable_check.cpp b/tests/stream_awaitable_check.cpp index b413c72..b3bb643 100644 --- a/tests/stream_awaitable_check.cpp +++ b/tests/stream_awaitable_check.cpp @@ -23,7 +23,6 @@ #include #include -#include "bycorf/io/io_uring_backend.h" #include "bycorf/net/tcp_stream.h" #include "bycorf/runtime/runtime.h" @@ -42,42 +41,6 @@ std::span Bytes(const std::string& value) { return std::as_bytes(std::span(value)); } -// Model AcquireSqe's recursive reap directly, without depending on kernel -// timing to fill both rings. The current CQE must already be retired when a -// completion submits work and reenters polling. -void CheckCompletionReentry(bool wait) { - bycorf::Worker worker; - bycorf::IoUringBackend backend; - bycorf::IoBackendOptions options; - options.ring_entries_ = 8; - options.recv_buffer_count_ = 0; - const auto initialized = backend.Init(options, &worker, -1); - Check(initialized.ok(), "initialize completion reentry test"); - if (!initialized.ok()) return; - struct Completion final : bycorf::IoCompletion { - bycorf::IoUringBackend* backend = nullptr; - unsigned calls = 0; - void Complete(bycorf::Worker&, int, unsigned) override { - if (++calls == 1) backend->Poll(); - } - } completion; - completion.backend = &backend; - __kernel_timespec timeout{0, 1}; - Check(backend.SubmitTimeout(timeout, &completion).ok(), - "submit reentry timer"); - Check(backend.Submit().ok(), "flush reentry timer"); - if (wait) Check(backend.Wait(1000), "wait dispatches the reentry timer"); - const auto deadline = std::chrono::steady_clock::now() + 1s; - while (completion.calls == 0 && std::chrono::steady_clock::now() < deadline) { - // DEFER_TASKRUN needs an enter to publish timer task-work into the CQ. - (void)backend.Submit(); - backend.Poll(); - } - Check(completion.calls == 1, - "recursive polling dispatches each CQE exactly once"); - backend.Shutdown(); -} - // A tiny send buffer and a delayed reader force the full-write operations to // survive multiple completions while retaining their payload and iovec state. Task CheckWrite(bycorf::Worker& worker, bool vectored, @@ -233,8 +196,6 @@ Task CheckStream(bycorf::Worker& worker) { int main() { std::signal(SIGPIPE, SIG_IGN); - CheckCompletionReentry(false); - CheckCompletionReentry(true); bycorf::Runtime runtime; runtime.Start(1, [](unsigned, bycorf::Worker& worker) { if (!worker.Init().ok()) return 1; From 57a0f956f4fe554484d6bd08ba0f18b2aa7b9bad Mon Sep 17 00:00:00 2001 From: Jerry Zhao Date: Mon, 12 Oct 2026 10:34:32 +0000 Subject: [PATCH 3/6] Use native TLS read and write awaitables --- CMakeLists.txt | 6 + include/bycorf/net/tcp_stream.h | 81 ++++----- include/bycorf/net/tls.h | 53 +++++- src/net/tcp_stream.cpp | 10 +- src/net/tls.cpp | 131 +++++++++----- tests/tls_awaitable_check.cpp | 305 ++++++++++++++++++++++++++++++++ 6 files changed, 490 insertions(+), 96 deletions(-) create mode 100644 tests/tls_awaitable_check.cpp diff --git a/CMakeLists.txt b/CMakeLists.txt index b987805..38df6dc 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -444,6 +444,12 @@ endif() if (BYCORF_BUILD_TESTS) find_package(Threads REQUIRED) + add_executable(bycorf_tls_awaitable_check tests/tls_awaitable_check.cpp) + target_link_libraries(bycorf_tls_awaitable_check PRIVATE bycorf::core Threads::Threads) + add_test(NAME bycorf_tls_awaitable_check COMMAND bycorf_tls_awaitable_check) + set_tests_properties(bycorf_tls_awaitable_check PROPERTIES + LABELS "net;io_uring;runtime;tls" TIMEOUT 60) + add_executable(bycorf_stream_awaitable_check tests/stream_awaitable_check.cpp) target_link_libraries(bycorf_stream_awaitable_check PRIVATE bycorf::core Threads::Threads) diff --git a/include/bycorf/net/tcp_stream.h b/include/bycorf/net/tcp_stream.h index cc9aef8..7e1e85b 100644 --- a/include/bycorf/net/tcp_stream.h +++ b/include/bycorf/net/tcp_stream.h @@ -35,6 +35,7 @@ #include "absl/status/statusor.h" #include "bycorf/io/completion.h" #include "bycorf/net/connection.h" +#include "bycorf/net/tls.h" #include "bycorf/runtime/task.h" namespace bycorf { @@ -58,14 +59,6 @@ class ReadOperation final : public IoCompletion { void Complete(Worker& worker, int result, unsigned flags) override; private: - template - friend class StreamOperation; - - void BindAwaiting(std::coroutine_handle<> awaiting) noexcept { - awaiting_ = awaiting; - connection_->read_waiter_ = awaiting; - } - Connection* connection_ = nullptr; std::span buffer_; std::optional immediate_status_; @@ -87,13 +80,6 @@ class WriteOperation final : public IoCompletion { void Complete(Worker& worker, int result, unsigned flags) override; private: - template - friend class StreamOperation; - - void BindAwaiting(std::coroutine_handle<> awaiting) noexcept { - awaiting_ = awaiting; - } - Connection* connection_ = nullptr; std::span buffer_; int result_ = 0; @@ -114,56 +100,56 @@ class WriteVOperation final : public IoCompletion { void Complete(Worker& worker, int result, unsigned flags) override; private: - template - friend class StreamOperation; - - void BindAwaiting(std::coroutine_handle<> awaiting) noexcept { - awaiting_ = awaiting; - } - Connection* connection_ = nullptr; std::span buffers_; msghdr message_{}; int result_ = 0; }; -// Selects raw I/O without a child frame, or owns the TLS coroutine until its -// result has been consumed. Operations borrow the stream and payload through -// completion and must be awaited on the connection's worker. They are immovable -// because the backend may retain the raw operation's address as its completion -// tag. Raw I/O starts in await_ready so immediate results require no coroutine -// transfer (returning the caller's own handle can grow the native stack under -// sanitizers). The owner worker cannot dispatch the new operation between -// await_ready and await_suspend, which binds its continuation before yielding. -// TLS retains Task's symmetric transfer and child ownership. +// Selects the transport while keeping the active awaitable directly in the +// caller's frame. Neither branch allocates a wrapper coroutine. The operation +// is immovable because a submitted raw I/O may retain its address as a tag. template class StreamOperation { public: using Result = absl::StatusOr; template explicit StreamOperation(Connection* connection, Args&&... args) - : raw_(std::in_place, connection, std::forward(args)...) {} - explicit StreamOperation(Task tls) : tls_(std::move(tls)) {} + : is_tls_(false), raw_(connection, std::forward(args)...) {} + StreamOperation(TlsState& state, TcpStream& stream, + std::span buffer) + : is_tls_(true), tls_(state.ReadSome(stream, buffer)) {} + StreamOperation(TlsState& state, TcpStream& stream, + std::span buffer) + : is_tls_(true), tls_(state.WriteSome(stream, buffer)) {} + StreamOperation(const StreamOperation&) = delete; + StreamOperation& operator=(const StreamOperation&) = delete; + StreamOperation(StreamOperation&&) = delete; + StreamOperation& operator=(StreamOperation&&) = delete; + ~StreamOperation() { + if (is_tls_) + std::destroy_at(&tls_); + else + std::destroy_at(&raw_); + } - bool await_ready() { - if (!raw_) return tls_.done(); - return raw_->await_ready() || !raw_->await_suspend({}); + bool await_ready() const noexcept { + return is_tls_ ? tls_.await_ready() : raw_.await_ready(); } - std::coroutine_handle<> await_suspend(std::coroutine_handle<> awaiting) { - if (raw_) { - raw_->BindAwaiting(awaiting); - return std::noop_coroutine(); - } - return std::move(tls_).operator co_await().await_suspend(awaiting); + bool await_suspend(std::coroutine_handle<> awaiting) { + return is_tls_ ? tls_.await_suspend(awaiting) + : raw_.await_suspend(awaiting); } Result await_resume() { - if (!raw_) return std::move(tls_).operator co_await().await_resume(); - return raw_->await_resume(); + return is_tls_ ? tls_.await_resume() : raw_.await_resume(); } private: - std::optional raw_; - Task tls_; + bool is_tls_; + union { + RawOperation raw_; + TlsState::IoAwaitable tls_; + }; }; // Outbound async connect (IORING_OP_CONNECT) on a not-yet-registered fd, raced @@ -265,7 +251,8 @@ class TcpStream { using WriteVAwaitable = StreamOperation; // Single-operation byte-stream I/O. Buffers and the stream must outlive - // completion. Plaintext awaits use the caller's frame; TLS owns a lazy Task. + // completion. Both transports keep their operation in the caller's frame; + // TLS retains child Tasks for ciphertext transport. ReadAwaitable ReadSome(std::span buffer); WriteAwaitable WriteSome(std::span buffer); WriteVAwaitable WriteSomeV(std::span buffers); diff --git a/include/bycorf/net/tls.h b/include/bycorf/net/tls.h index 3904a28..e871135 100644 --- a/include/bycorf/net/tls.h +++ b/include/bycorf/net/tls.h @@ -19,6 +19,7 @@ #include #include +#include #include #include #include @@ -75,6 +76,50 @@ class TlsContext { // pair. This also makes the TLS state transferable with a handed-off fd. class TlsState { public: + // Stores the SSL read/write loop in its caller's frame. Ciphertext transport + // still uses the existing worker-local Tasks. The stream and payload must + // outlive completion; this operation may only be awaited on their worker. + class IoAwaitable { + public: + IoAwaitable(TlsState& state, TcpStream& stream, std::span buffer) + : IoAwaitable(state, stream, std::span(buffer), true) { + } + IoAwaitable(TlsState& state, TcpStream& stream, + std::span buffer) + : IoAwaitable(state, stream, buffer, false) {} + IoAwaitable(const IoAwaitable&) = delete; + IoAwaitable& operator=(const IoAwaitable&) = delete; + IoAwaitable(IoAwaitable&&) = delete; + IoAwaitable& operator=(IoAwaitable&&) = delete; + + bool await_ready() const noexcept { return buffer_.empty(); } + bool await_suspend(std::coroutine_handle<> awaiting); + absl::StatusOr await_resume(); + + private: + enum class Phase { kSsl, kFlushing, kReading }; + IoAwaitable(TlsState& state, TcpStream& stream, + std::span buffer, bool read) + : state_(state), stream_(stream), buffer_(buffer), read_(read) {} + bool Advance(absl::Status status); + void OnComplete(); + + TlsState& state_; + TcpStream& stream_; + std::span buffer_; + std::coroutine_handle<> awaiting_{}; + // A Task's callback runs inside final_suspend. Keep that frame alive until + // the callback returns, even when Advance starts another child immediately. + Task completing_; + Task pending_; + std::optional> result_; + std::size_t size_ = 0; + int error_ = 0; + Phase phase_ = Phase::kSsl; + bool read_; + bool advancing_ = false; + }; + ~TlsState(); TlsState(const TlsState&) = delete; @@ -89,10 +134,8 @@ class TlsState { std::string_view peer_name); Task Handshake(TcpStream& stream); - Task> ReadSome(TcpStream& stream, - std::span buffer); - Task> WriteSome( - TcpStream& stream, std::span buffer); + IoAwaitable ReadSome(TcpStream& stream, std::span buffer); + IoAwaitable WriteSome(TcpStream& stream, std::span buffer); Task Shutdown(TcpStream& stream); absl::StatusOr> PeerCertificateUriSans() const; @@ -100,6 +143,8 @@ class TlsState { Task ReadCiphertext(TcpStream& stream); friend class TcpStream; + template + friend class StreamOperation; std::unique_ptr impl_; }; diff --git a/src/net/tcp_stream.cpp b/src/net/tcp_stream.cpp index 74cae07..827b132 100644 --- a/src/net/tcp_stream.cpp +++ b/src/net/tcp_stream.cpp @@ -494,7 +494,7 @@ absl::Status TcpStream::SetReadAhead(bool enabled) noexcept { TcpStream::ReadAwaitable TcpStream::ReadSome(std::span buffer) { if (tls_ != nullptr) { - return ReadAwaitable(tls_->ReadSome(*this, buffer)); + return ReadAwaitable(*tls_, *this, buffer); } return ReadAwaitable(connection_, buffer); } @@ -506,7 +506,7 @@ ReadOperation TcpStream::ReadRawSome(std::span buffer) noexcept { TcpStream::WriteAwaitable TcpStream::WriteSome( std::span buffer) { if (tls_ != nullptr) { - return WriteAwaitable(tls_->WriteSome(*this, buffer)); + return WriteAwaitable(*tls_, *this, buffer); } return WriteAwaitable(connection_, buffer); } @@ -521,10 +521,10 @@ TcpStream::WriteVAwaitable TcpStream::WriteSomeV( if (tls_ != nullptr) { for (const auto& buffer : buffers) { if (buffer.iov_len == 0) continue; - return WriteVAwaitable(tls_->WriteSome( - *this, + return WriteVAwaitable( + *tls_, *this, std::span( - static_cast(buffer.iov_base), buffer.iov_len))); + static_cast(buffer.iov_base), buffer.iov_len)); } return WriteVAwaitable(connection_, std::span{}); } diff --git a/src/net/tls.cpp b/src/net/tls.cpp index bd20674..0364b4e 100644 --- a/src/net/tls.cpp +++ b/src/net/tls.cpp @@ -367,54 +367,105 @@ Task TlsState::Handshake(TcpStream& stream) { co_return absl::OkStatus(); } -Task> TlsState::ReadSome( - TcpStream& stream, std::span buffer) { - if (buffer.empty()) co_return std::size_t{0}; - while (true) { - std::size_t size = 0; - ERR_clear_error(); - const int result = - SSL_read_ex(impl_->ssl_, buffer.data(), buffer.size(), &size); - const int error = - result == 1 ? SSL_ERROR_NONE : SSL_get_error(impl_->ssl_, result); - absl::Status flushed = co_await FlushOutput(stream); - if (!flushed.ok()) co_return flushed; - if (result == 1) co_return size; - if (error == SSL_ERROR_ZERO_RETURN) co_return std::size_t{0}; - if (error == SSL_ERROR_WANT_WRITE) continue; - if (error == SSL_ERROR_WANT_READ) { - absl::Status read = co_await ReadCiphertext(stream); - if (!read.ok()) co_return read; - continue; - } - co_return TlsError("TLS read failed"); +TlsState::IoAwaitable TlsState::ReadSome(TcpStream& stream, + std::span buffer) { + return IoAwaitable(*this, stream, buffer); +} + +TlsState::IoAwaitable TlsState::WriteSome(TcpStream& stream, + std::span buffer) { + return IoAwaitable(*this, stream, buffer); +} + +bool TlsState::IoAwaitable::await_suspend(std::coroutine_handle<> awaiting) { + awaiting_ = awaiting; + return Advance(absl::OkStatus()); +} + +absl::StatusOr TlsState::IoAwaitable::await_resume() { + if (buffer_.empty()) return std::size_t{0}; + return std::move(*result_); +} + +void TlsState::IoAwaitable::OnComplete() { + // Immediate child completions are consumed by the running loop after + // resume() returns. This prevents synchronous retries from growing the stack. + if (advancing_) return; + completing_ = std::move(pending_); + if (!Advance(std::move(completing_).TakeResult())) { + // Do not resume the parent inline: it destroys this awaitable and its + // completing_ Task, whose final_suspend callback is still on the stack. + stream_.connection_->worker_->Enqueue(awaiting_); } } -Task> TlsState::WriteSome( - TcpStream& stream, std::span buffer) { - if (buffer.empty()) co_return std::size_t{0}; +bool TlsState::IoAwaitable::Advance(absl::Status status) { + advancing_ = true; while (true) { - std::size_t size = 0; - ERR_clear_error(); - const int result = - SSL_write_ex(impl_->ssl_, buffer.data(), buffer.size(), &size); - const int error = - result == 1 ? SSL_ERROR_NONE : SSL_get_error(impl_->ssl_, result); - absl::Status flushed = co_await FlushOutput(stream); - if (!flushed.ok()) co_return flushed; - if (result == 1) co_return size; - if (error == SSL_ERROR_WANT_WRITE) continue; - if (error == SSL_ERROR_WANT_READ) { - absl::Status read = co_await ReadCiphertext(stream); - if (!read.ok()) co_return read; + if (!status.ok()) { + result_.emplace(std::move(status)); + break; + } + if (phase_ == Phase::kSsl) { + size_ = 0; + ERR_clear_error(); + // Read constructors accept mutable buffers; write constructors never + // enter SSL_read_ex. The const span shares storage for both operations. + const int result = + read_ ? SSL_read_ex(state_.impl_->ssl_, + const_cast(buffer_.data()), + buffer_.size(), &size_) + : SSL_write_ex(state_.impl_->ssl_, buffer_.data(), + buffer_.size(), &size_); + error_ = result == 1 ? SSL_ERROR_NONE + : SSL_get_error(state_.impl_->ssl_, result); + phase_ = Phase::kFlushing; + pending_ = state_.FlushOutput(stream_); + } else if (phase_ == Phase::kFlushing) { + if (error_ == SSL_ERROR_NONE) { + result_.emplace(size_); + break; + } + if (error_ == SSL_ERROR_ZERO_RETURN) { + if (read_) + result_.emplace(std::size_t{0}); + else + result_.emplace(absl::UnavailableError("TLS peer closed connection")); + break; + } + if (error_ == SSL_ERROR_WANT_WRITE) { + phase_ = Phase::kSsl; + continue; + } + if (error_ != SSL_ERROR_WANT_READ) { + result_.emplace( + TlsError(read_ ? "TLS read failed" : "TLS write failed")); + break; + } + phase_ = Phase::kReading; + pending_ = state_.ReadCiphertext(stream_); + } else { + phase_ = Phase::kSsl; continue; } - if (error == SSL_ERROR_ZERO_RETURN) { - co_return absl::UnavailableError("TLS peer closed connection"); + + pending_.SetCompletionCallback( + this, [](void* context, std::coroutine_handle<>) noexcept { + static_cast(context)->OnComplete(); + }); + // Use Task's awaiter to retain background-class propagation. There is no + // coroutine continuation: the callback advances this operation instead. + auto child = std::move(pending_).operator co_await().await_suspend({}); + child.resume(); + if (!pending_.done()) { + advancing_ = false; + return true; } - co_return TlsError("TLS write failed"); + status = std::move(pending_).TakeResult(); + pending_ = {}; } + advancing_ = false; + return false; } Task TlsState::Shutdown(TcpStream& stream) { diff --git a/tests/tls_awaitable_check.cpp b/tests/tls_awaitable_check.cpp new file mode 100644 index 0000000..b90fa7d --- /dev/null +++ b/tests/tls_awaitable_check.cpp @@ -0,0 +1,305 @@ +/* + * 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 +#include +#include +#include + +#include "bycorf/net/tcp_stream.h" +#include "bycorf/net/tls.h" +#include "bycorf/runtime/runtime.h" + +namespace { +using bycorf::Task; +int failures = 0; +void Check(bool ok, const char* message) { + if (!ok) { + ++failures; + std::cerr << "FAIL " << message << '\n'; + } +} + +// Each process owns its temporary identity; no external openssl executable or +// shared fixture files are needed when sanitizer and normal tests run together. +struct Identity { + char cert_path[64] = "/tmp/bycorf-tls-cert-XXXXXX"; + char key_path[64] = "/tmp/bycorf-tls-key-XXXXXX"; + bool Create() { + EVP_PKEY_CTX* gen = EVP_PKEY_CTX_new_id(EVP_PKEY_RSA, nullptr); + EVP_PKEY* key = nullptr; + if (!gen) return false; + bool ok = EVP_PKEY_keygen_init(gen) > 0 && + EVP_PKEY_CTX_set_rsa_keygen_bits(gen, 2048) > 0 && + EVP_PKEY_keygen(gen, &key) > 0; + EVP_PKEY_CTX_free(gen); + if (!ok) return false; + X509* cert = X509_new(); + X509_set_version(cert, 2); + ASN1_INTEGER_set(X509_get_serialNumber(cert), 1); + X509_gmtime_adj(X509_get_notBefore(cert), -60); + X509_gmtime_adj(X509_get_notAfter(cert), 3600); + X509_set_pubkey(cert, key); + auto* name = X509_get_subject_name(cert); + X509_NAME_add_entry_by_txt( + name, "CN", MBSTRING_ASC, + reinterpret_cast("localhost"), -1, -1, 0); + X509_set_issuer_name(cert, name); + ok = X509_sign(cert, key, EVP_sha256()) > 0; + const int cert_fd = mkstemp(cert_path); + const int key_fd = mkstemp(key_path); + if (cert_fd >= 0) { + FILE* file = fdopen(cert_fd, "w"); + ok &= PEM_write_X509(file, cert) == 1; + fclose(file); + } else { + ok = false; + } + if (key_fd >= 0) { + FILE* file = fdopen(key_fd, "w"); + ok &= PEM_write_PrivateKey(file, key, nullptr, nullptr, 0, nullptr, + nullptr) == 1; + fclose(file); + } else { + ok = false; + } + X509_free(cert); + EVP_PKEY_free(key); + return ok; + } + ~Identity() { + unlink(cert_path); + unlink(key_path); + } +}; + +struct NextRound { + bycorf::Worker& worker; + bool await_ready() const noexcept { return false; } + void await_suspend(std::coroutine_handle<> handle) { + worker.EnqueueNext(handle); + } + void await_resume() const noexcept {} +}; + +Task DuplexWriter(bycorf::TcpStream& stream, + const std::string& payload, bool& done) { + auto status = co_await stream.WriteAll(std::as_bytes(std::span(payload))); + Check(status.ok(), "concurrent TLS writer"); + done = true; + co_return status; +} + +Task Exercise(bycorf::Worker& worker, + std::shared_ptr context, + bool abrupt, bool background, bool duplex = false) { + int sockets[2]; + if (socketpair(AF_UNIX, SOCK_STREAM | SOCK_CLOEXEC, 0, sockets) != 0) + co_return absl::InternalError("socketpair"); + const int capacity = 4096; + setsockopt(sockets[0], SOL_SOCKET, SO_SNDBUF, &capacity, sizeof(capacity)); + fcntl(sockets[0], F_SETFL, O_NONBLOCK); + bycorf::Connection initial; + initial.worker_ = &worker; + initial.closed_ = false; + initial.file_.fd_ = sockets[0]; + auto* connection = worker.AddConnection(std::move(initial)); + bycorf::TcpStream stream(connection); + auto borrow = stream.BorrowStorage(); +#if defined(BYCORF_ASAN) + // Two TLS records still exercise refill and immediate completion without + // turning sanitizer runs into a million-allocation microbenchmark. + constexpr std::size_t kBytes = 32 * 1024; +#else + constexpr std::size_t kBytes = 256 * 1024; +#endif + std::string payload(kBytes, 'x'); + for (std::size_t i = 0; i < payload.size(); ++i) payload[i] = char(i % 251); + bool peer_ok = true; + std::atomic read_finished{false}; + std::thread peer([&] { + SSL_CTX* ctx = SSL_CTX_new(TLS_client_method()); + SSL* ssl = SSL_new(ctx); + SSL_set_fd(ssl, sockets[1]); + peer_ok = SSL_connect(ssl) == 1; + auto send = [&] { + std::size_t offset = 0; + while (peer_ok && offset < payload.size()) { + const int n = SSL_write(ssl, payload.data() + offset, + static_cast(payload.size() - offset)); + if (n <= 0) { + peer_ok = false; + break; + } + offset += n; + } + }; + auto receive = [&] { + std::this_thread::sleep_for(std::chrono::milliseconds(20)); + std::string received(payload.size(), '\0'); + std::size_t offset = 0; + while (peer_ok && offset < received.size()) { + const int n = SSL_read(ssl, received.data() + offset, + static_cast(received.size() - offset)); + if (n <= 0) { + peer_ok = false; + break; + } + offset += n; + } + peer_ok &= received == payload; + }; + if (duplex) receive(); + send(); + if (!abrupt) { + if (!duplex) receive(); + if (peer_ok && SSL_shutdown(ssl) == 0) SSL_shutdown(ssl); + } + if (abrupt) { + while (!read_finished.load()) + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + } + SSL_free(ssl); + SSL_CTX_free(ctx); + close(sockets[1]); + }); + bool writer_done = !duplex; + auto handshake = co_await stream.StartTls(context, true); + Check(handshake.ok(), "TLS handshake"); + if (!handshake.ok()) std::cerr << handshake << '\n'; + if (handshake.ok()) { + Check((co_await stream.ReadSome({})).value() == 0, "empty TLS read"); + Check((co_await stream.WriteSome({})).value() == 0, "empty TLS write"); + std::array empty{}; + Check((co_await stream.WriteSomeV(empty)).value() == 0, "empty TLS vector"); + if (duplex) { + worker.Spawn(DuplexWriter(stream, payload, writer_done)); + co_await NextRound{worker}; + Check(connection->write_inflight_, "TLS writer holds the output lock"); + } + std::string received(payload.size(), '\0'); + auto start = std::chrono::steady_clock::now(); + std::size_t offset = 0; + // Small reads exercise already-decrypted data as well as record refill. +#if defined(BYCORF_ASAN) + constexpr std::size_t kReadBytes = 8; +#else + constexpr std::size_t kReadBytes = 1; +#endif + while (offset < received.size()) { + auto result = co_await stream.ReadSome(std::as_writable_bytes( + std::span(received.data() + offset, + std::min(kReadBytes, received.size() - offset)))); + if (!result.ok() || *result == 0) { + Check(false, "TLS payload read"); + break; + } + offset += *result; + } + read_finished.store(true); + const auto nanos = std::chrono::duration_cast( + std::chrono::steady_clock::now() - start) + .count(); + std::cout << "TLS read ns/byte=" << double(nanos) / payload.size() + << " background=" << background << " abrupt=" << abrupt << '\n'; + Check(received == payload, "TLS record boundaries preserve bytes"); + Check((bycorf::CurrentTaskClass() == bycorf::TaskClass::kBackground) == + background, + "TLS preserves scheduling class"); + if (!abrupt && !duplex) { + std::array vectors{ + {{nullptr, 0}, + {payload.data(), 13}, + {nullptr, 0}, + {payload.data() + 13, payload.size() - 13}}}; + Check((co_await stream.WriteAllV(vectors)).ok(), + "TLS write with backpressure"); + } + std::byte byte; + auto eof = co_await stream.ReadSome(std::span(&byte, 1)); + Check(abrupt ? !eof.ok() : eof.ok() && *eof == 0, + "TLS distinguishes truncation from close_notify"); + if (abrupt) { + Check(!(co_await stream.WriteAll(std::as_bytes(std::span(payload)))).ok(), + "TLS write propagates peer disconnect"); + } else { + Check((co_await stream.ShutdownTls()).ok(), "TLS shutdown"); + } + } + read_finished.store(true); + stream.Close().IgnoreError(); + if (handshake.ok()) { + while (!writer_done) co_await NextRound{worker}; + } + peer.join(); + Check(peer_ok, "TLS peer transfer"); + Check(!connection->write_inflight_, "TLS sends drained"); + co_return absl::OkStatus(); +} + +Task Run(bycorf::Worker& worker, + std::shared_ptr context, + bool background) { + for (bool abrupt : {false, true}) { + Check((co_await Exercise(worker, context, abrupt, background)).ok(), + "TLS exercise"); + } + Check((co_await Exercise(worker, context, false, background, true)).ok(), + "TLS reader waits for concurrent ciphertext flush"); + worker.RequestStop(); + co_return absl::OkStatus(); +} +} // namespace + +int main() { + std::signal(SIGPIPE, SIG_IGN); + std::cout << std::unitbuf; + Identity identity; + if (!identity.Create()) return 1; + auto context = bycorf::TlsContext::CreateServer( + {.cert_file_ = identity.cert_path, .key_file_ = identity.key_path}); + if (!context.ok()) { + std::cerr << context.status() << '\n'; + return 1; + } + std::cout << "awaitable sizes: read=" + << sizeof(bycorf::TcpStream::ReadAwaitable) + << " write=" << sizeof(bycorf::TcpStream::WriteAwaitable) << '\n'; + for (bool background : {false, true}) { + bycorf::Runtime runtime; + runtime.Start(1, [&](unsigned, bycorf::Worker& worker) { + if (!worker.Init().ok()) return 1; + if (background) + worker.SpawnBackground(Run(worker, *context, true)); + else + worker.Spawn(Run(worker, *context, false)); + worker.Run(); + worker.Shutdown(); + return failures == 0 ? 0 : 1; + }); + runtime.WaitUntilStopped(); + if (runtime.exit_code() != 0) return 1; + } + return failures == 0 ? 0 : 1; +} From 74f9e67d357fe88aec234d98a9374f7cef23e355 Mon Sep 17 00:00:00 2001 From: Jerry Zhao Date: Mon, 12 Oct 2026 10:49:07 +0000 Subject: [PATCH 4/6] Flush stream regression progress for CI timeout diagnosis --- tests/stream_awaitable_check.cpp | 2 ++ 1 file changed, 2 insertions(+) diff --git a/tests/stream_awaitable_check.cpp b/tests/stream_awaitable_check.cpp index b3bb643..51d964a 100644 --- a/tests/stream_awaitable_check.cpp +++ b/tests/stream_awaitable_check.cpp @@ -196,6 +196,8 @@ Task CheckStream(bycorf::Worker& worker) { int main() { std::signal(SIGPIPE, SIG_IGN); + // Preserve the last completed check when CTest terminates a stalled run. + std::cout << std::unitbuf; bycorf::Runtime runtime; runtime.Start(1, [](unsigned, bycorf::Worker& worker) { if (!worker.Init().ok()) return 1; From 960a9cca775897e0d3a64908cb7bc6a4a983c3c3 Mon Sep 17 00:00:00 2001 From: Jerry Zhao Date: Mon, 12 Oct 2026 11:12:07 +0000 Subject: [PATCH 5/6] Restore coroutine-based TLS read and write loops --- include/bycorf/net/tcp_stream.h | 81 +++++++++++--------- include/bycorf/net/tls.h | 53 +------------ src/net/tcp_stream.cpp | 10 +-- src/net/tls.cpp | 131 ++++++++++---------------------- tests/tls_awaitable_check.cpp | 9 ++- 5 files changed, 104 insertions(+), 180 deletions(-) diff --git a/include/bycorf/net/tcp_stream.h b/include/bycorf/net/tcp_stream.h index 7e1e85b..cc9aef8 100644 --- a/include/bycorf/net/tcp_stream.h +++ b/include/bycorf/net/tcp_stream.h @@ -35,7 +35,6 @@ #include "absl/status/statusor.h" #include "bycorf/io/completion.h" #include "bycorf/net/connection.h" -#include "bycorf/net/tls.h" #include "bycorf/runtime/task.h" namespace bycorf { @@ -59,6 +58,14 @@ class ReadOperation final : public IoCompletion { void Complete(Worker& worker, int result, unsigned flags) override; private: + template + friend class StreamOperation; + + void BindAwaiting(std::coroutine_handle<> awaiting) noexcept { + awaiting_ = awaiting; + connection_->read_waiter_ = awaiting; + } + Connection* connection_ = nullptr; std::span buffer_; std::optional immediate_status_; @@ -80,6 +87,13 @@ class WriteOperation final : public IoCompletion { void Complete(Worker& worker, int result, unsigned flags) override; private: + template + friend class StreamOperation; + + void BindAwaiting(std::coroutine_handle<> awaiting) noexcept { + awaiting_ = awaiting; + } + Connection* connection_ = nullptr; std::span buffer_; int result_ = 0; @@ -100,56 +114,56 @@ class WriteVOperation final : public IoCompletion { void Complete(Worker& worker, int result, unsigned flags) override; private: + template + friend class StreamOperation; + + void BindAwaiting(std::coroutine_handle<> awaiting) noexcept { + awaiting_ = awaiting; + } + Connection* connection_ = nullptr; std::span buffers_; msghdr message_{}; int result_ = 0; }; -// Selects the transport while keeping the active awaitable directly in the -// caller's frame. Neither branch allocates a wrapper coroutine. The operation -// is immovable because a submitted raw I/O may retain its address as a tag. +// Selects raw I/O without a child frame, or owns the TLS coroutine until its +// result has been consumed. Operations borrow the stream and payload through +// completion and must be awaited on the connection's worker. They are immovable +// because the backend may retain the raw operation's address as its completion +// tag. Raw I/O starts in await_ready so immediate results require no coroutine +// transfer (returning the caller's own handle can grow the native stack under +// sanitizers). The owner worker cannot dispatch the new operation between +// await_ready and await_suspend, which binds its continuation before yielding. +// TLS retains Task's symmetric transfer and child ownership. template class StreamOperation { public: using Result = absl::StatusOr; template explicit StreamOperation(Connection* connection, Args&&... args) - : is_tls_(false), raw_(connection, std::forward(args)...) {} - StreamOperation(TlsState& state, TcpStream& stream, - std::span buffer) - : is_tls_(true), tls_(state.ReadSome(stream, buffer)) {} - StreamOperation(TlsState& state, TcpStream& stream, - std::span buffer) - : is_tls_(true), tls_(state.WriteSome(stream, buffer)) {} - StreamOperation(const StreamOperation&) = delete; - StreamOperation& operator=(const StreamOperation&) = delete; - StreamOperation(StreamOperation&&) = delete; - StreamOperation& operator=(StreamOperation&&) = delete; - ~StreamOperation() { - if (is_tls_) - std::destroy_at(&tls_); - else - std::destroy_at(&raw_); - } + : raw_(std::in_place, connection, std::forward(args)...) {} + explicit StreamOperation(Task tls) : tls_(std::move(tls)) {} - bool await_ready() const noexcept { - return is_tls_ ? tls_.await_ready() : raw_.await_ready(); + bool await_ready() { + if (!raw_) return tls_.done(); + return raw_->await_ready() || !raw_->await_suspend({}); } - bool await_suspend(std::coroutine_handle<> awaiting) { - return is_tls_ ? tls_.await_suspend(awaiting) - : raw_.await_suspend(awaiting); + std::coroutine_handle<> await_suspend(std::coroutine_handle<> awaiting) { + if (raw_) { + raw_->BindAwaiting(awaiting); + return std::noop_coroutine(); + } + return std::move(tls_).operator co_await().await_suspend(awaiting); } Result await_resume() { - return is_tls_ ? tls_.await_resume() : raw_.await_resume(); + if (!raw_) return std::move(tls_).operator co_await().await_resume(); + return raw_->await_resume(); } private: - bool is_tls_; - union { - RawOperation raw_; - TlsState::IoAwaitable tls_; - }; + std::optional raw_; + Task tls_; }; // Outbound async connect (IORING_OP_CONNECT) on a not-yet-registered fd, raced @@ -251,8 +265,7 @@ class TcpStream { using WriteVAwaitable = StreamOperation; // Single-operation byte-stream I/O. Buffers and the stream must outlive - // completion. Both transports keep their operation in the caller's frame; - // TLS retains child Tasks for ciphertext transport. + // completion. Plaintext awaits use the caller's frame; TLS owns a lazy Task. ReadAwaitable ReadSome(std::span buffer); WriteAwaitable WriteSome(std::span buffer); WriteVAwaitable WriteSomeV(std::span buffers); diff --git a/include/bycorf/net/tls.h b/include/bycorf/net/tls.h index e871135..3904a28 100644 --- a/include/bycorf/net/tls.h +++ b/include/bycorf/net/tls.h @@ -19,7 +19,6 @@ #include #include -#include #include #include #include @@ -76,50 +75,6 @@ class TlsContext { // pair. This also makes the TLS state transferable with a handed-off fd. class TlsState { public: - // Stores the SSL read/write loop in its caller's frame. Ciphertext transport - // still uses the existing worker-local Tasks. The stream and payload must - // outlive completion; this operation may only be awaited on their worker. - class IoAwaitable { - public: - IoAwaitable(TlsState& state, TcpStream& stream, std::span buffer) - : IoAwaitable(state, stream, std::span(buffer), true) { - } - IoAwaitable(TlsState& state, TcpStream& stream, - std::span buffer) - : IoAwaitable(state, stream, buffer, false) {} - IoAwaitable(const IoAwaitable&) = delete; - IoAwaitable& operator=(const IoAwaitable&) = delete; - IoAwaitable(IoAwaitable&&) = delete; - IoAwaitable& operator=(IoAwaitable&&) = delete; - - bool await_ready() const noexcept { return buffer_.empty(); } - bool await_suspend(std::coroutine_handle<> awaiting); - absl::StatusOr await_resume(); - - private: - enum class Phase { kSsl, kFlushing, kReading }; - IoAwaitable(TlsState& state, TcpStream& stream, - std::span buffer, bool read) - : state_(state), stream_(stream), buffer_(buffer), read_(read) {} - bool Advance(absl::Status status); - void OnComplete(); - - TlsState& state_; - TcpStream& stream_; - std::span buffer_; - std::coroutine_handle<> awaiting_{}; - // A Task's callback runs inside final_suspend. Keep that frame alive until - // the callback returns, even when Advance starts another child immediately. - Task completing_; - Task pending_; - std::optional> result_; - std::size_t size_ = 0; - int error_ = 0; - Phase phase_ = Phase::kSsl; - bool read_; - bool advancing_ = false; - }; - ~TlsState(); TlsState(const TlsState&) = delete; @@ -134,8 +89,10 @@ class TlsState { std::string_view peer_name); Task Handshake(TcpStream& stream); - IoAwaitable ReadSome(TcpStream& stream, std::span buffer); - IoAwaitable WriteSome(TcpStream& stream, std::span buffer); + Task> ReadSome(TcpStream& stream, + std::span buffer); + Task> WriteSome( + TcpStream& stream, std::span buffer); Task Shutdown(TcpStream& stream); absl::StatusOr> PeerCertificateUriSans() const; @@ -143,8 +100,6 @@ class TlsState { Task ReadCiphertext(TcpStream& stream); friend class TcpStream; - template - friend class StreamOperation; std::unique_ptr impl_; }; diff --git a/src/net/tcp_stream.cpp b/src/net/tcp_stream.cpp index 827b132..74cae07 100644 --- a/src/net/tcp_stream.cpp +++ b/src/net/tcp_stream.cpp @@ -494,7 +494,7 @@ absl::Status TcpStream::SetReadAhead(bool enabled) noexcept { TcpStream::ReadAwaitable TcpStream::ReadSome(std::span buffer) { if (tls_ != nullptr) { - return ReadAwaitable(*tls_, *this, buffer); + return ReadAwaitable(tls_->ReadSome(*this, buffer)); } return ReadAwaitable(connection_, buffer); } @@ -506,7 +506,7 @@ ReadOperation TcpStream::ReadRawSome(std::span buffer) noexcept { TcpStream::WriteAwaitable TcpStream::WriteSome( std::span buffer) { if (tls_ != nullptr) { - return WriteAwaitable(*tls_, *this, buffer); + return WriteAwaitable(tls_->WriteSome(*this, buffer)); } return WriteAwaitable(connection_, buffer); } @@ -521,10 +521,10 @@ TcpStream::WriteVAwaitable TcpStream::WriteSomeV( if (tls_ != nullptr) { for (const auto& buffer : buffers) { if (buffer.iov_len == 0) continue; - return WriteVAwaitable( - *tls_, *this, + return WriteVAwaitable(tls_->WriteSome( + *this, std::span( - static_cast(buffer.iov_base), buffer.iov_len)); + static_cast(buffer.iov_base), buffer.iov_len))); } return WriteVAwaitable(connection_, std::span{}); } diff --git a/src/net/tls.cpp b/src/net/tls.cpp index 0364b4e..bd20674 100644 --- a/src/net/tls.cpp +++ b/src/net/tls.cpp @@ -367,105 +367,54 @@ Task TlsState::Handshake(TcpStream& stream) { co_return absl::OkStatus(); } -TlsState::IoAwaitable TlsState::ReadSome(TcpStream& stream, - std::span buffer) { - return IoAwaitable(*this, stream, buffer); -} - -TlsState::IoAwaitable TlsState::WriteSome(TcpStream& stream, - std::span buffer) { - return IoAwaitable(*this, stream, buffer); -} - -bool TlsState::IoAwaitable::await_suspend(std::coroutine_handle<> awaiting) { - awaiting_ = awaiting; - return Advance(absl::OkStatus()); -} - -absl::StatusOr TlsState::IoAwaitable::await_resume() { - if (buffer_.empty()) return std::size_t{0}; - return std::move(*result_); -} - -void TlsState::IoAwaitable::OnComplete() { - // Immediate child completions are consumed by the running loop after - // resume() returns. This prevents synchronous retries from growing the stack. - if (advancing_) return; - completing_ = std::move(pending_); - if (!Advance(std::move(completing_).TakeResult())) { - // Do not resume the parent inline: it destroys this awaitable and its - // completing_ Task, whose final_suspend callback is still on the stack. - stream_.connection_->worker_->Enqueue(awaiting_); +Task> TlsState::ReadSome( + TcpStream& stream, std::span buffer) { + if (buffer.empty()) co_return std::size_t{0}; + while (true) { + std::size_t size = 0; + ERR_clear_error(); + const int result = + SSL_read_ex(impl_->ssl_, buffer.data(), buffer.size(), &size); + const int error = + result == 1 ? SSL_ERROR_NONE : SSL_get_error(impl_->ssl_, result); + absl::Status flushed = co_await FlushOutput(stream); + if (!flushed.ok()) co_return flushed; + if (result == 1) co_return size; + if (error == SSL_ERROR_ZERO_RETURN) co_return std::size_t{0}; + if (error == SSL_ERROR_WANT_WRITE) continue; + if (error == SSL_ERROR_WANT_READ) { + absl::Status read = co_await ReadCiphertext(stream); + if (!read.ok()) co_return read; + continue; + } + co_return TlsError("TLS read failed"); } } -bool TlsState::IoAwaitable::Advance(absl::Status status) { - advancing_ = true; +Task> TlsState::WriteSome( + TcpStream& stream, std::span buffer) { + if (buffer.empty()) co_return std::size_t{0}; while (true) { - if (!status.ok()) { - result_.emplace(std::move(status)); - break; - } - if (phase_ == Phase::kSsl) { - size_ = 0; - ERR_clear_error(); - // Read constructors accept mutable buffers; write constructors never - // enter SSL_read_ex. The const span shares storage for both operations. - const int result = - read_ ? SSL_read_ex(state_.impl_->ssl_, - const_cast(buffer_.data()), - buffer_.size(), &size_) - : SSL_write_ex(state_.impl_->ssl_, buffer_.data(), - buffer_.size(), &size_); - error_ = result == 1 ? SSL_ERROR_NONE - : SSL_get_error(state_.impl_->ssl_, result); - phase_ = Phase::kFlushing; - pending_ = state_.FlushOutput(stream_); - } else if (phase_ == Phase::kFlushing) { - if (error_ == SSL_ERROR_NONE) { - result_.emplace(size_); - break; - } - if (error_ == SSL_ERROR_ZERO_RETURN) { - if (read_) - result_.emplace(std::size_t{0}); - else - result_.emplace(absl::UnavailableError("TLS peer closed connection")); - break; - } - if (error_ == SSL_ERROR_WANT_WRITE) { - phase_ = Phase::kSsl; - continue; - } - if (error_ != SSL_ERROR_WANT_READ) { - result_.emplace( - TlsError(read_ ? "TLS read failed" : "TLS write failed")); - break; - } - phase_ = Phase::kReading; - pending_ = state_.ReadCiphertext(stream_); - } else { - phase_ = Phase::kSsl; + std::size_t size = 0; + ERR_clear_error(); + const int result = + SSL_write_ex(impl_->ssl_, buffer.data(), buffer.size(), &size); + const int error = + result == 1 ? SSL_ERROR_NONE : SSL_get_error(impl_->ssl_, result); + absl::Status flushed = co_await FlushOutput(stream); + if (!flushed.ok()) co_return flushed; + if (result == 1) co_return size; + if (error == SSL_ERROR_WANT_WRITE) continue; + if (error == SSL_ERROR_WANT_READ) { + absl::Status read = co_await ReadCiphertext(stream); + if (!read.ok()) co_return read; continue; } - - pending_.SetCompletionCallback( - this, [](void* context, std::coroutine_handle<>) noexcept { - static_cast(context)->OnComplete(); - }); - // Use Task's awaiter to retain background-class propagation. There is no - // coroutine continuation: the callback advances this operation instead. - auto child = std::move(pending_).operator co_await().await_suspend({}); - child.resume(); - if (!pending_.done()) { - advancing_ = false; - return true; + if (error == SSL_ERROR_ZERO_RETURN) { + co_return absl::UnavailableError("TLS peer closed connection"); } - status = std::move(pending_).TakeResult(); - pending_ = {}; + co_return TlsError("TLS write failed"); } - advancing_ = false; - return false; } Task TlsState::Shutdown(TcpStream& stream) { diff --git a/tests/tls_awaitable_check.cpp b/tests/tls_awaitable_check.cpp index b90fa7d..ce0f021 100644 --- a/tests/tls_awaitable_check.cpp +++ b/tests/tls_awaitable_check.cpp @@ -138,6 +138,7 @@ Task Exercise(bycorf::Worker& worker, for (std::size_t i = 0; i < payload.size(); ++i) payload[i] = char(i % 251); bool peer_ok = true; std::atomic read_finished{false}; + std::atomic allow_receive{!duplex}; std::thread peer([&] { SSL_CTX* ctx = SSL_CTX_new(TLS_client_method()); SSL* ssl = SSL_new(ctx); @@ -156,7 +157,11 @@ Task Exercise(bycorf::Worker& worker, } }; auto receive = [&] { - std::this_thread::sleep_for(std::chrono::milliseconds(20)); + // The duplex case must keep the writer suspended until the worker has + // observed its output lock. A fixed sleep is unreliable under sanitizers. + while (!allow_receive.load()) + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + if (!duplex) std::this_thread::sleep_for(std::chrono::milliseconds(20)); std::string received(payload.size(), '\0'); std::size_t offset = 0; while (peer_ok && offset < received.size()) { @@ -197,6 +202,7 @@ Task Exercise(bycorf::Worker& worker, worker.Spawn(DuplexWriter(stream, payload, writer_done)); co_await NextRound{worker}; Check(connection->write_inflight_, "TLS writer holds the output lock"); + allow_receive.store(true); } std::string received(payload.size(), '\0'); auto start = std::chrono::steady_clock::now(); @@ -248,6 +254,7 @@ Task Exercise(bycorf::Worker& worker, } } read_finished.store(true); + allow_receive.store(true); stream.Close().IgnoreError(); if (handshake.ok()) { while (!writer_done) co_await NextRound{worker}; From 1c9ba9bcca132397111ae87c72060b342505bada Mon Sep 17 00:00:00 2001 From: Jerry Zhao Date: Mon, 12 Oct 2026 11:12:07 +0000 Subject: [PATCH 6/6] Exercise stream half-close and EOF over loopback TCP --- tests/stream_awaitable_check.cpp | 50 +++++++++++++++++++++++++++++--- 1 file changed, 46 insertions(+), 4 deletions(-) diff --git a/tests/stream_awaitable_check.cpp b/tests/stream_awaitable_check.cpp index 51d964a..de3eb51 100644 --- a/tests/stream_awaitable_check.cpp +++ b/tests/stream_awaitable_check.cpp @@ -12,6 +12,7 @@ */ #include +#include #include #include @@ -41,6 +42,48 @@ std::span Bytes(const std::string& value) { return std::as_bytes(std::span(value)); } +// Exercise TCP half-close semantics on the transport TcpStream serves. An +// AF_UNIX socketpair takes a different kernel multishot-receive path. +bool TcpPair(int (&sockets)[2]) { + const int listener = socket(AF_INET, SOCK_STREAM | SOCK_CLOEXEC, 0); + if (listener < 0) return false; + sockaddr_in address{}; + address.sin_family = AF_INET; + address.sin_addr.s_addr = htonl(INADDR_LOOPBACK); + if (bind(listener, reinterpret_cast(&address), sizeof(address)) != + 0 || + listen(listener, 1) != 0) { + close(listener); + return false; + } + socklen_t size = sizeof(address); + if (getsockname(listener, reinterpret_cast(&address), &size) != + 0) { + close(listener); + return false; + } + const int peer = socket(AF_INET, SOCK_STREAM | SOCK_CLOEXEC, 0); + if (peer < 0) { + close(listener); + return false; + } + if (connect(peer, reinterpret_cast(&address), size) != 0) { + close(peer); + close(listener); + return false; + } + const int local = + accept4(listener, nullptr, nullptr, SOCK_NONBLOCK | SOCK_CLOEXEC); + close(listener); + if (local < 0) { + close(peer); + return false; + } + sockets[0] = local; + sockets[1] = peer; + return true; +} + // A tiny send buffer and a delayed reader force the full-write operations to // survive multiple completions while retaining their payload and iovec state. Task CheckWrite(bycorf::Worker& worker, bool vectored, @@ -139,8 +182,7 @@ Task CheckStream(bycorf::Worker& worker) { "repeated immediate completion keeps native stack bounded"); int sockets[2]; - if (socketpair(AF_UNIX, SOCK_STREAM | SOCK_NONBLOCK | SOCK_CLOEXEC, 0, - sockets) == 0) { + if (TcpPair(sockets)) { bycorf::Connection initial; initial.worker_ = &worker; initial.closed_ = false; @@ -149,7 +191,7 @@ Task CheckStream(bycorf::Worker& worker) { TcpStream stream(connection); auto borrow = stream.BorrowStorage(); Check(write(sockets[1], "abcdefgh", 8) == 8, "queue received bytes"); - shutdown(sockets[1], SHUT_WR); + Check(shutdown(sockets[1], SHUT_WR) == 0, "half-close TCP peer"); std::size_t read_bytes = 0; while (read_bytes < buffer.size()) { auto result = @@ -182,7 +224,7 @@ Task CheckStream(bycorf::Worker& worker) { "closed read fails immediately"); close(sockets[1]); } else { - Check(false, "read socketpair"); + Check(false, "create TCP pair"); } for (bool vectored : {false, true}) { for (bool disconnect : {false, true}) {