diff --git a/CMakeLists.txt b/CMakeLists.txt index 7364d9d..38df6dc 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -443,6 +443,16 @@ if (BYCORF_BUILD_EXAMPLES OR BYCORF_BUILD_TESTS) 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) + 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 +491,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..cc9aef8 100644 --- a/include/bycorf/net/tcp_stream.h +++ b/include/bycorf/net/tcp_stream.h @@ -29,6 +29,7 @@ #include #include #include +#include #include #include "absl/status/statusor.h" @@ -57,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_; @@ -78,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; @@ -98,12 +114,58 @@ 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. +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)) {} + + 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(); + return raw_->await_resume(); + } + + private: + std::optional raw_; + Task tls_; +}; + // Outbound async connect (IORING_OP_CONNECT) on a not-yet-registered fd, raced // against an optional deadline. The awaitable owns both SQEs; the first // completion to arrive submits io_uring_prep_cancel for the loser, and the @@ -198,10 +260,18 @@ 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); + 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); + + // 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); diff --git a/src/net/tcp_stream.cpp b/src/net/tcp_stream.cpp index 94f605a..74cae07 100644 --- a/src/net/tcp_stream.cpp +++ b/src/net/tcp_stream.cpp @@ -492,24 +492,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 +516,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( diff --git a/tests/stream_awaitable_check.cpp b/tests/stream_awaitable_check.cpp new file mode 100644 index 0000000..de3eb51 --- /dev/null +++ b/tests/stream_awaitable_check.cpp @@ -0,0 +1,253 @@ +/* + * 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 "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)); +} + +// 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, + 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 (TcpPair(sockets)) { + 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"); + Check(shutdown(sockets[1], SHUT_WR) == 0, "half-close TCP peer"); + 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, "create TCP pair"); + } + 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); + // 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; + worker.Spawn(CheckStream(worker)); + worker.Run(); + worker.Shutdown(); + return failures == 0 ? 0 : 1; + }); + runtime.WaitUntilStopped(); + return runtime.exit_code(); +} diff --git a/tests/tls_awaitable_check.cpp b/tests/tls_awaitable_check.cpp new file mode 100644 index 0000000..ce0f021 --- /dev/null +++ b/tests/tls_awaitable_check.cpp @@ -0,0 +1,312 @@ +/* + * 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::atomic allow_receive{!duplex}; + 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 = [&] { + // 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()) { + 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"); + allow_receive.store(true); + } + 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); + allow_receive.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; +}