Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 13 additions & 0 deletions CMakeLists.txt
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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")
Expand Down
78 changes: 74 additions & 4 deletions include/bycorf/net/tcp_stream.h
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@
#include <span>
#include <string>
#include <string_view>
#include <utility>
#include <vector>

#include "absl/status/statusor.h"
Expand Down Expand Up @@ -57,6 +58,14 @@ class ReadOperation final : public IoCompletion {
void Complete(Worker& worker, int result, unsigned flags) override;

private:
template <typename>
friend class StreamOperation;

void BindAwaiting(std::coroutine_handle<> awaiting) noexcept {
awaiting_ = awaiting;
connection_->read_waiter_ = awaiting;
}

Connection* connection_ = nullptr;
std::span<std::byte> buffer_;
std::optional<absl::Status> immediate_status_;
Expand All @@ -78,6 +87,13 @@ class WriteOperation final : public IoCompletion {
void Complete(Worker& worker, int result, unsigned flags) override;

private:
template <typename>
friend class StreamOperation;

void BindAwaiting(std::coroutine_handle<> awaiting) noexcept {
awaiting_ = awaiting;
}

Connection* connection_ = nullptr;
std::span<const std::byte> buffer_;
int result_ = 0;
Expand All @@ -98,12 +114,58 @@ class WriteVOperation final : public IoCompletion {
void Complete(Worker& worker, int result, unsigned flags) override;

private:
template <typename>
friend class StreamOperation;

void BindAwaiting(std::coroutine_handle<> awaiting) noexcept {
awaiting_ = awaiting;
}

Connection* connection_ = nullptr;
std::span<const iovec> 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 <typename RawOperation>
class StreamOperation {
public:
using Result = absl::StatusOr<std::size_t>;
template <typename... Args>
explicit StreamOperation(Connection* connection, Args&&... args)
: raw_(std::in_place, connection, std::forward<Args>(args)...) {}
explicit StreamOperation(Task<Result> 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<RawOperation> raw_;
Task<Result> 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
Expand Down Expand Up @@ -198,10 +260,18 @@ class TcpStream {
// with their own bounded ingress queues.
absl::Status SetReadAhead(bool enabled) noexcept;

Task<absl::StatusOr<std::size_t>> ReadSome(std::span<std::byte> buffer);
Task<absl::StatusOr<std::size_t>> WriteSome(
std::span<const std::byte> buffer);
Task<absl::StatusOr<std::size_t>> WriteSomeV(std::span<const iovec> buffers);
using ReadAwaitable = StreamOperation<ReadOperation>;
using WriteAwaitable = StreamOperation<WriteOperation>;
using WriteVAwaitable = StreamOperation<WriteVOperation>;

// 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<std::byte> buffer);
WriteAwaitable WriteSome(std::span<const std::byte> buffer);
WriteVAwaitable WriteSomeV(std::span<const iovec> buffers);

// Full writes retain their coroutine loops: each completion schedules the
// writer before it checks the result and submits any remaining bytes.
Task<absl::Status> WriteAll(std::span<const std::byte> buffer);
Task<absl::Status> WriteAllV(std::span<const iovec> buffers);

Expand Down
25 changes: 12 additions & 13 deletions src/net/tcp_stream.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -492,44 +492,43 @@ absl::Status TcpStream::SetReadAhead(bool enabled) noexcept {
return absl::OkStatus();
}

Task<absl::StatusOr<std::size_t>> TcpStream::ReadSome(
std::span<std::byte> buffer) {
TcpStream::ReadAwaitable TcpStream::ReadSome(std::span<std::byte> 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<std::byte> buffer) noexcept {
return ReadOperation(connection_, buffer);
}

Task<absl::StatusOr<std::size_t>> TcpStream::WriteSome(
TcpStream::WriteAwaitable TcpStream::WriteSome(
std::span<const std::byte> 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(
std::span<const std::byte> buffer) noexcept {
return WriteOperation(connection_, buffer);
}

Task<absl::StatusOr<std::size_t>> TcpStream::WriteSomeV(
TcpStream::WriteVAwaitable TcpStream::WriteSomeV(
std::span<const iovec> 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<const std::byte>(
static_cast<const std::byte*>(buffer.iov_base), buffer.iov_len));
static_cast<const std::byte*>(buffer.iov_base), buffer.iov_len)));
}
co_return std::size_t{0};
return WriteVAwaitable(connection_, std::span<const iovec>{});
}
co_return co_await WriteRawSomeV(buffers);
return WriteVAwaitable(connection_, buffers);
}

WriteVOperation TcpStream::WriteRawSomeV(
Expand Down
Loading