diff --git a/CHANGELOG.md b/CHANGELOG.md index 6226909e..60ad4998 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -61,6 +61,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed +- **Smaller channel send coroutine frames**: `channel::send(...)` now lets + its internal waiter borrow the by-value payload already owned by the send + coroutine frame instead of retaining a second moved-from/moved-to `T` pair. + Directly constructed public send awaiters remain owning, and queue, + cancellation, close, and dequeue-versus-destruction behavior is unchanged + (#1047). - **Allocation-free ready semaphore acquires**: Non-cancellable acquires now defer shared wake-state allocation until the ready permit check fails. Entering `await_suspend` allocates before taking the semaphore queue lock, diff --git a/examples/CMakeLists.txt b/examples/CMakeLists.txt index c61a9e80..4e96d8a5 100644 --- a/examples/CMakeLists.txt +++ b/examples/CMakeLists.txt @@ -66,6 +66,13 @@ add_executable(bench_channel bench_channel.cpp) target_link_libraries(bench_channel PRIVATE elio) target_link_options(bench_channel PRIVATE ${STATIC_LINK_FLAGS}) +add_executable(bench_channel_send_frame + bench_channel_send_frame.cpp + channel_send_frame_bench_factory.cpp +) +target_link_libraries(bench_channel_send_frame PRIVATE elio) +target_link_options(bench_channel_send_frame PRIVATE ${STATIC_LINK_FLAGS}) + # Signal handling example add_executable(signal_handling signal_handling.cpp) target_link_libraries(signal_handling PRIVATE elio) diff --git a/examples/bench_channel_send_frame.cpp b/examples/bench_channel_send_frame.cpp new file mode 100644 index 00000000..15c5b74b --- /dev/null +++ b/examples/bench_channel_send_frame.cpp @@ -0,0 +1,387 @@ +#include "channel_send_frame_bench_factory.hpp" + +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#if defined(__GLIBC__) +#include +#endif + +using namespace elio; +using namespace std::chrono; + +namespace { + +struct allocation_record { + void* address = nullptr; + size_t requested = 0; +}; + +thread_local bool record_allocations = false; +thread_local std::array allocation_records{}; +thread_local size_t stored_allocation_count = 0; +thread_local size_t total_allocation_count = 0; +thread_local size_t total_requested_bytes = 0; + +void record_allocation(void* address, size_t requested) noexcept { + if (!record_allocations) return; + const size_t index = total_allocation_count++; + total_requested_bytes += requested; + if (index < allocation_records.size()) { + allocation_records[index] = {address, requested}; + ++stored_allocation_count; + } +} + +class allocation_recording_scope { +public: + allocation_recording_scope() noexcept { + stored_allocation_count = 0; + total_allocation_count = 0; + total_requested_bytes = 0; + record_allocations = true; + } + + ~allocation_recording_scope() { stop(); } + + void stop() noexcept { record_allocations = false; } +}; + +allocation_record find_frame_allocation(void* frame_address) { + const auto frame = reinterpret_cast(frame_address); + allocation_record match; + size_t matches = 0; + for (size_t i = 0; i < stored_allocation_count; ++i) { + const auto base = reinterpret_cast( + allocation_records[i].address); + const size_t size = allocation_records[i].requested; + if (frame >= base && frame - base < size) { + match = allocation_records[i]; + ++matches; + } + } + if (matches != 1) { + std::abort(); + } + return match; +} + +size_t allocator_usable_bytes(void* allocation) noexcept { +#if defined(__GLIBC__) + return malloc_usable_size(allocation); +#else + (void)allocation; + return 0; +#endif +} + +struct send_frame_measurement { + size_t frame_bytes = 0; + size_t usable_bytes = 0; + size_t allocations = 0; + size_t allocated_bytes = 0; +}; + +template +send_frame_measurement measure_send_frame(bool with_token) { + sync::channel> channel(1); + coro::cancel_source source; + auto token = source.get_token(); + + allocation_recording_scope recording; + auto frame = with_token + ? bench::make_cancellable_send_frame(channel, token) + : bench::make_send_frame(channel); + recording.stop(); + + const auto frame_allocation = find_frame_allocation(frame.address()); + return { + frame_allocation.requested, + allocator_usable_bytes(frame_allocation.address), + total_allocation_count, + total_requested_bytes}; +} + +template +double measure_send_task_construction_ns(bool with_token, size_t operations) { + sync::channel> channel(1); + coro::cancel_source source; + auto token = source.get_token(); + + const auto start = steady_clock::now(); + for (size_t i = 0; i < operations; ++i) { + if (with_token) { + auto frame = bench::make_cancellable_send_frame(channel, token); + std::atomic_signal_fence(std::memory_order_seq_cst); + (void)frame; + } else { + auto frame = bench::make_send_frame(channel); + std::atomic_signal_fence(std::memory_order_seq_cst); + (void)frame; + } + } + const auto end = steady_clock::now(); + return duration(end - start).count() / operations; +} + +template +coro::task ready_bounded_send_loop( + sync::channel>& channel, + size_t operations) { + for (size_t i = 0; i < operations; ++i) { + if (!co_await channel.send(bench::inline_payload{})) { + std::abort(); + } + if (!channel.try_recv().has_value()) { + std::abort(); + } + } +} + +template +coro::task ready_unbounded_send_loop( + sync::channel>& channel, + size_t operations) { + for (size_t i = 0; i < operations; ++i) { + if (!co_await channel.send(bench::inline_payload{})) { + std::abort(); + } + if (!channel.try_recv().has_value()) { + std::abort(); + } + } +} + +template +coro::task ready_token_send_loop( + sync::channel>& channel, + coro::cancel_token token, size_t operations) { + for (size_t i = 0; i < operations; ++i) { + const auto result = co_await channel.send( + bench::inline_payload{}, token); + if (!result.success() || !channel.try_recv().has_value()) { + std::abort(); + } + } +} + +template +double measure_ready_send_ns(Factory&& factory, size_t operations) { + auto operation = factory(); + auto handle = coro::detail::task_access::handle(operation); + const auto start = steady_clock::now(); + { + coro::detail::frame_context_scope frame_scope( + std::addressof(handle.promise())); + handle.resume(); + } + const auto end = steady_clock::now(); + if (!handle.done()) { + std::abort(); + } + return duration(end - start).count() / operations; +} + +template +coro::task forced_sender( + sync::channel>& channel, + coro::cancel_token token, size_t operations) { + for (size_t i = 0; i < operations; ++i) { + if constexpr (WithToken) { + const auto result = co_await channel.send( + bench::inline_payload{}, token); + if (!result.success()) { + std::abort(); + } + } else if (!co_await channel.send( + bench::inline_payload{})) { + std::abort(); + } + } +} + +template +coro::task forced_receiver( + sync::channel>& channel, + size_t operations) { + for (size_t i = 0; i < operations; ++i) { + const auto result = co_await channel.recv(); + if (!result.has_value()) { + std::abort(); + } + } +} + +template +double measure_forced_handoff_ns(size_t capacity, size_t operations) { + using payload = bench::inline_payload<256>; + sync::channel channel(capacity); + const bool bounded_full = capacity == 1; + if (bounded_full && !channel.try_send(payload{})) { + std::abort(); + } + + coro::cancel_source source; + auto sender = forced_sender<256, WithToken>( + channel, source.get_token(), operations); + auto receiver = forced_receiver( + channel, operations + static_cast(bounded_full)); + auto sender_handle = coro::detail::task_access::handle(sender); + auto receiver_handle = coro::detail::task_access::handle(receiver); + + { + coro::detail::frame_context_scope frame_scope( + std::addressof(sender_handle.promise())); + sender_handle.resume(); + } + if (sender_handle.done()) { + std::abort(); + } + + const auto start = steady_clock::now(); + { + coro::detail::frame_context_scope frame_scope( + std::addressof(receiver_handle.promise())); + receiver_handle.resume(); + } + const auto end = steady_clock::now(); + + if (!sender_handle.done() || !receiver_handle.done() || !channel.empty()) { + std::abort(); + } + return duration(end - start).count() / operations; +} + +void print_rate(const char* name, double ns_per_send) { + std::cout << std::setw(42) << std::left << name + << std::setw(12) << std::right << std::fixed + << std::setprecision(2) << ns_per_send << " ns/send " + << (1e9 / ns_per_send) << " sends/s\n"; +} + +} // namespace + +#if defined(__GNUC__) || defined(__clang__) +#define ELIO_BENCH_NOINLINE __attribute__((noinline)) +#else +#define ELIO_BENCH_NOINLINE +#endif + +ELIO_BENCH_NOINLINE void* operator new(std::size_t size) { + void* address = std::malloc(size == 0 ? 1 : size); + if (!address) throw std::bad_alloc(); + record_allocation(address, size); + return address; +} + +ELIO_BENCH_NOINLINE void* operator new[](std::size_t size) { + return ::operator new(size); +} + +ELIO_BENCH_NOINLINE void operator delete(void* address) noexcept { + std::free(address); +} +ELIO_BENCH_NOINLINE void operator delete( + void* address, std::size_t) noexcept { + std::free(address); +} +ELIO_BENCH_NOINLINE void operator delete[](void* address) noexcept { + std::free(address); +} +ELIO_BENCH_NOINLINE void operator delete[]( + void* address, std::size_t) noexcept { + std::free(address); +} + +#undef ELIO_BENCH_NOINLINE + +int main(int argc, char** argv) { + bool smoke = false; + if (argc == 2 && std::string_view(argv[1]) == "--smoke") { + smoke = true; + } else if (argc != 1) { + std::cerr << "usage: bench_channel_send_frame [--smoke]\n"; + return 2; + } + + log::logger::instance().set_level(log::level::error); + + std::cout << "=== Channel Send Frame Benchmark ===\n"; + if (smoke) { + std::cout << "Smoke mode: reduced iteration counts for Debug " + "validation.\n"; + } + std::cout << "Naturally aligned inline payloads; usable bytes are zero " + "when the allocator does not expose them.\n"; + std::cout << "Payload Token Frame bytes Usable bytes Allocations " + "Allocated bytes Construct ns\n"; + + auto print_frame = [](size_t payload_bytes, bool with_token, + send_frame_measurement measured, + double construction_ns) { + std::cout << std::setw(7) << payload_bytes << " " + << std::setw(5) << (with_token ? "yes" : "no") << " " + << std::setw(11) << measured.frame_bytes << " " + << std::setw(12) << measured.usable_bytes << " " + << std::setw(11) << measured.allocations << " " + << std::setw(15) << measured.allocated_bytes << " " + << std::fixed << std::setprecision(2) << construction_ns + << '\n'; + }; + + const size_t construction_operations = smoke ? 100 : 50000; + auto measure_payload = [&]() { + print_frame(Bytes, false, measure_send_frame(false), + measure_send_task_construction_ns( + false, construction_operations)); + print_frame(Bytes, true, measure_send_frame(true), + measure_send_task_construction_ns( + true, construction_operations)); + }; + + measure_payload.template operator()<8>(); + measure_payload.template operator()<64>(); + measure_payload.template operator()<256>(); + measure_payload.template operator()<1024>(); + + const size_t ready_operations = smoke ? 100 : 100000; + sync::channel> bounded(1); + auto unbounded = sync::channel>::unbounded(); + coro::cancel_source source; + auto token = source.get_token(); + + print_rate("ready bounded send (256 bytes)", measure_ready_send_ns( + [&] { return ready_bounded_send_loop(bounded, ready_operations); }, + ready_operations)); + print_rate("ready unbounded send (256 bytes)", measure_ready_send_ns( + [&] { return ready_unbounded_send_loop(unbounded, ready_operations); }, + ready_operations)); + print_rate("ready active-token send (256 bytes)", measure_ready_send_ns( + [&] { return ready_token_send_loop( + bounded, token, ready_operations); }, ready_operations)); + + const size_t handoff_operations = smoke ? 100 : 100000; + print_rate("forced bounded-full handoff (256 bytes)", + measure_forced_handoff_ns(1, handoff_operations)); + print_rate("forced bounded-full token handoff (256 bytes)", + measure_forced_handoff_ns(1, handoff_operations)); + print_rate("forced rendezvous handoff (256 bytes)", + measure_forced_handoff_ns(0, handoff_operations)); + print_rate("forced rendezvous token handoff (256 bytes)", + measure_forced_handoff_ns(0, handoff_operations)); + + return 0; +} diff --git a/examples/channel_send_frame_bench_factory.cpp b/examples/channel_send_frame_bench_factory.cpp new file mode 100644 index 00000000..623a8fa7 --- /dev/null +++ b/examples/channel_send_frame_bench_factory.cpp @@ -0,0 +1,53 @@ +#include "channel_send_frame_bench_factory.hpp" + +#include + +namespace elio::bench { + +#if defined(__GNUC__) || defined(__clang__) +#define ELIO_BENCH_NOINLINE __attribute__((noinline)) +#else +#define ELIO_BENCH_NOINLINE +#endif + +template +ELIO_BENCH_NOINLINE unstarted_frame make_send_frame( + sync::channel>& channel) { + auto operation = channel.send(inline_payload{}); + auto handle = coro::detail::task_access::release(std::move(operation)); + return unstarted_frame( + std::coroutine_handle<>::from_address(handle.address())); +} + +template +ELIO_BENCH_NOINLINE unstarted_frame make_cancellable_send_frame( + sync::channel>& channel, + coro::cancel_token token) { + auto operation = channel.send( + inline_payload{}, std::move(token)); + auto handle = coro::detail::task_access::release(std::move(operation)); + return unstarted_frame( + std::coroutine_handle<>::from_address(handle.address())); +} + +template unstarted_frame make_send_frame<8>( + sync::channel>&); +template unstarted_frame make_send_frame<64>( + sync::channel>&); +template unstarted_frame make_send_frame<256>( + sync::channel>&); +template unstarted_frame make_send_frame<1024>( + sync::channel>&); + +template unstarted_frame make_cancellable_send_frame<8>( + sync::channel>&, coro::cancel_token); +template unstarted_frame make_cancellable_send_frame<64>( + sync::channel>&, coro::cancel_token); +template unstarted_frame make_cancellable_send_frame<256>( + sync::channel>&, coro::cancel_token); +template unstarted_frame make_cancellable_send_frame<1024>( + sync::channel>&, coro::cancel_token); + +#undef ELIO_BENCH_NOINLINE + +} // namespace elio::bench diff --git a/examples/channel_send_frame_bench_factory.hpp b/examples/channel_send_frame_bench_factory.hpp new file mode 100644 index 00000000..a5219381 --- /dev/null +++ b/examples/channel_send_frame_bench_factory.hpp @@ -0,0 +1,80 @@ +#pragma once + +#include +#include +#include +#include + +#include +#include + +namespace elio::bench { + +template +struct inline_payload { + std::array bytes{}; + + inline_payload() = default; + inline_payload(const inline_payload&) = delete; + inline_payload& operator=(const inline_payload&) = delete; + inline_payload(inline_payload&&) noexcept = default; + inline_payload& operator=(inline_payload&&) noexcept = default; +}; + +class unstarted_frame { +public: + explicit unstarted_frame(std::coroutine_handle<> handle) noexcept + : handle_(handle) {} + + unstarted_frame(const unstarted_frame&) = delete; + unstarted_frame& operator=(const unstarted_frame&) = delete; + + unstarted_frame(unstarted_frame&& other) noexcept + : handle_(std::exchange(other.handle_, {})) {} + + unstarted_frame& operator=(unstarted_frame&& other) noexcept { + if (this != &other) { + if (handle_) handle_.destroy(); + handle_ = std::exchange(other.handle_, {}); + } + return *this; + } + + ~unstarted_frame() { + if (handle_) handle_.destroy(); + } + + void* address() const noexcept { return handle_.address(); } + +private: + std::coroutine_handle<> handle_; +}; + +template +unstarted_frame make_send_frame( + sync::channel>& channel); + +template +unstarted_frame make_cancellable_send_frame( + sync::channel>& channel, + coro::cancel_token token); + +extern template unstarted_frame make_send_frame<8>( + sync::channel>&); +extern template unstarted_frame make_send_frame<64>( + sync::channel>&); +extern template unstarted_frame make_send_frame<256>( + sync::channel>&); +extern template unstarted_frame make_send_frame<1024>( + sync::channel>&); + +extern template unstarted_frame make_cancellable_send_frame<8>( + sync::channel>&, coro::cancel_token); +extern template unstarted_frame make_cancellable_send_frame<64>( + sync::channel>&, coro::cancel_token); +extern template unstarted_frame make_cancellable_send_frame<256>( + sync::channel>&, coro::cancel_token); +extern template unstarted_frame make_cancellable_send_frame<1024>( + sync::channel>&, coro::cancel_token); + +} // namespace elio::bench diff --git a/include/elio/sync/channel.hpp b/include/elio/sync/channel.hpp index ed053c67..fdce1a24 100644 --- a/include/elio/sync/channel.hpp +++ b/include/elio/sync/channel.hpp @@ -33,7 +33,7 @@ inline std::atomic bounded_recv_paused_after_failed_pop_for_test{false}; template class channel { public: - // Forward declarations for intrusive_list + // Public awaitable declarations class send_awaitable; class cancellable_send_awaitable; class recv_awaitable; @@ -101,16 +101,17 @@ class channel { channel(channel&&) = delete; channel& operator=(channel&&) = delete; - /// Send awaitable — handles both bounded and rendezvous channels. - /// Stores handle + value. Inherits intrusive_list_node for safe unlinking. - class send_awaitable : public elio::detail::intrusive_list_node { +private: + class send_waiter_core + : public elio::detail::intrusive_list_node { public: - send_awaitable(channel& ch, T value) + send_waiter_core(channel& ch, T& value, bool cancellable = false) : ch_(ch) - , value_(std::move(value)) - , wake_state_(detail::make_wake_state()) {} + , value_(std::addressof(value)) + , wake_state_(detail::make_wake_state()) + , cancellable_(cancellable) {} - ~send_awaitable() { + ~send_waiter_core() { // Fast path: if we never suspended, we were never enqueued if (!suspended_) return; @@ -145,14 +146,14 @@ class channel { if (auto* receiver = ch_.claim_receiver_locked()) { to_schedule = receiver->wake_state_; if (claim_completion()) { - ch_.queue_.push(std::move(value_)); + ch_.queue_.push(std::move(*value_)); success_ = true; } should_suspend = false; } } else if (ch_.is_unbounded()) { if (claim_completion()) { - ch_.queue_.push(std::move(value_)); + ch_.queue_.push(std::move(*value_)); success_ = true; if (auto* receiver = ch_.claim_receiver_locked()) { to_schedule = receiver->wake_state_; @@ -164,7 +165,7 @@ class channel { if (ch_.ring_->size() < ch_.capacity_ && ch_.ring_->can_push()) { if (claim_completion()) { - const bool pushed = ch_.ring_->try_push(value_); + const bool pushed = ch_.ring_->try_push(*value_); assert(pushed); (void)pushed; success_ = true; @@ -219,13 +220,6 @@ class channel { return success_; } - protected: - send_awaitable(channel& ch, T value, bool cancellable) - : ch_(ch) - , value_(std::move(value)) - , wake_state_(detail::make_wake_state()) - , cancellable_(cancellable) {} - const detail::wake_state_ptr& cancellation_wake_state() const noexcept { return wake_state_; } @@ -246,21 +240,96 @@ class channel { return {success_, coro::cancel_result::completed}; } - private: bool claim_completion() noexcept { return !cancellable_ || detail::claim_wake_state(wake_state_) != detail::wake_action::rejected; } + T& value() noexcept { return *value_; } + void mark_success() noexcept { success_ = true; } + const detail::wake_state_ptr& wake_state() const noexcept { + return wake_state_; + } + bool is_cancellable() const noexcept { return cancellable_; } + + private: channel& ch_; - T value_; + T* value_; detail::wake_state_ptr wake_state_; bool success_ = false; bool suspended_ = false; // True if enqueued in send_waiters_ bool cancellable_ = false; + }; - friend class channel; + class borrowed_cancellable_send_awaitable { + public: + borrowed_cancellable_send_awaitable( + channel& ch, T& value, coro::cancel_token token) + : core_(ch, value, true) { + cancel_registration_ = token.on_cancel( + [state = core_.cancellation_wake_state()] { + state->request_cancel(); + }); + } + + ~borrowed_cancellable_send_awaitable() { + cancel_registration_.unregister(); + } + + bool await_ready() const noexcept { + return core_.cancellation_wake_state()->was_cancelled(); + } + + bool await_suspend(std::coroutine_handle<> h) noexcept { + return core_.await_suspend(h); + } + + cancellable_send_result await_resume() noexcept { + cancel_registration_.unregister(); + return core_.await_resume_cancellable(); + } + + private: + send_waiter_core core_; + coro::cancel_token::registration cancel_registration_; + }; + +public: + /// Send awaitable — handles both bounded and rendezvous channels. + /// Directly constructed awaiters own their value independently. + class send_awaitable + : public elio::detail::intrusive_list_node { + public: + send_awaitable(channel& ch, T value) + : value_(std::move(value)) + , core_(ch, value_) {} + + bool await_ready() const noexcept { return core_.await_ready(); } + + bool await_suspend(std::coroutine_handle<> h) noexcept { + return core_.await_suspend(h); + } + + bool await_resume() noexcept { return core_.await_resume(); } + bool is_linked() const noexcept { return core_.is_linked(); } + + protected: + send_awaitable(channel& ch, T value, bool cancellable) + : value_(std::move(value)) + , core_(ch, value_, cancellable) {} + + const detail::wake_state_ptr& cancellation_wake_state() const noexcept { + return core_.cancellation_wake_state(); + } + + cancellable_send_result await_resume_cancellable() noexcept { + return core_.await_resume_cancellable(); + } + + private: + T value_; + send_waiter_core core_; }; class cancellable_send_awaitable : public send_awaitable { @@ -492,8 +561,8 @@ class channel { } } - // Suspend via send_awaitable - send_awaitable awaitable{*this, std::move(value)}; + // The coroutine frame owns value for the complete suspended lifetime. + send_waiter_core awaitable{*this, value}; bool pushed = co_await awaitable; if (pushed) { @@ -506,8 +575,8 @@ class channel { /// A cancellation winner does not transfer the value into the channel. coro::task send( T value, coro::cancel_token token) { - co_return co_await cancellable_send_awaitable( - *this, std::move(value), std::move(token)); + co_return co_await borrowed_cancellable_send_awaitable( + *this, value, std::move(token)); } /// Try to send without waiting @@ -583,9 +652,9 @@ class channel { { std::lock_guard guard(mutex_); if (auto* sender = claim_sender_locked()) { - result = std::optional(std::move(sender->value_)); - sender->success_ = true; - sender_handle = sender->wake_state_; + result = std::optional(std::move(sender->value())); + sender->mark_success(); + sender_handle = sender->wake_state(); } else if (closed_.load(std::memory_order_acquire)) { if (!queue_.empty()) { result = std::move(queue_.front()); @@ -623,9 +692,9 @@ class channel { auto* sender = claim_sender_locked(); if (sender) { result = std::optional( - std::move(sender->value_)); - sender->success_ = true; - sender_handle = sender->wake_state_; + std::move(sender->value())); + sender->mark_success(); + sender_handle = sender->wake_state(); } else if (closed_.load(std::memory_order_acquire)) { result = std::nullopt; } else { @@ -713,9 +782,9 @@ class channel { resolved = true; } else if (auto* sender = claim_sender_locked()) { result = std::optional( - std::move(sender->value_)); - sender->success_ = true; - sender_handle = sender->wake_state_; + std::move(sender->value())); + sender->mark_success(); + sender_handle = sender->wake_state(); resolved = true; } else { retry = true; @@ -739,9 +808,9 @@ class channel { if (!awaitable.claim_completion()) { resolved = true; } else if (auto* sender = claim_sender_locked()) { - result = std::optional(std::move(sender->value_)); - sender->success_ = true; - sender_handle = sender->wake_state_; + result = std::optional(std::move(sender->value())); + sender->mark_success(); + sender_handle = sender->wake_state(); resolved = true; } else { retry = true; @@ -836,9 +905,9 @@ class channel { if (!sender) { return std::nullopt; } - result = std::optional(std::move(sender->value_)); - sender->success_ = true; - sender_handle = sender->wake_state_; + result = std::optional(std::move(sender->value())); + sender->mark_success(); + sender_handle = sender->wake_state(); } else { return std::nullopt; } @@ -873,18 +942,18 @@ class channel { queue_.push(std::move(*val)); } while (auto* sender = claim_sender_locked()) { - queue_.push(std::move(sender->value_)); - sender->success_ = true; // Value was delivered to queue - to_schedule.push_back(sender->wake_state_); + queue_.push(std::move(sender->value())); + sender->mark_success(); // Value was delivered to queue + to_schedule.push_back(sender->wake_state()); } } // Drain rendezvous send_waiters_ if (is_rendezvous()) { while (auto* sender = claim_sender_locked()) { - queue_.push(std::move(sender->value_)); - sender->success_ = true; // Value was delivered to queue - to_schedule.push_back(sender->wake_state_); + queue_.push(std::move(sender->value())); + sender->mark_success(); // Value was delivered to queue + to_schedule.push_back(sender->wake_state()); } } @@ -947,11 +1016,11 @@ class channel { return nullptr; } - send_awaitable* claim_sender_locked() noexcept { + send_waiter_core* claim_sender_locked() noexcept { while (!send_waiters_.empty()) { auto* sender = send_waiters_.pop_front(); - if (sender->cancellable_ && - detail::claim_wake_state(sender->wake_state_) == + if (sender->is_cancellable() && + detail::claim_wake_state(sender->wake_state()) == detail::wake_action::rejected) { continue; } @@ -990,11 +1059,11 @@ class channel { return; } - const bool pushed = ring_->try_push(sender->value_); + const bool pushed = ring_->try_push(sender->value()); assert(pushed); (void)pushed; - sender->success_ = true; - sender_handle = sender->wake_state_; + sender->mark_success(); + sender_handle = sender->wake_state(); if (auto* receiver = claim_receiver_locked()) { receiver_handle = receiver->wake_state_; } @@ -1024,7 +1093,7 @@ class channel { std::unique_ptr> ring_; std::queue queue_; elio::detail::intrusive_list recv_waiters_; - elio::detail::intrusive_list send_waiters_; + elio::detail::intrusive_list send_waiters_; size_t capacity_; std::atomic closed_; }; diff --git a/tests/unit/test_channel_success.cpp b/tests/unit/test_channel_success.cpp index 4e92b881..27b6bee3 100644 --- a/tests/unit/test_channel_success.cpp +++ b/tests/unit/test_channel_success.cpp @@ -22,10 +22,14 @@ #include #include +#include +#include +#include #include #include #include #include +#include using namespace elio::sync; using namespace elio::coro; @@ -82,6 +86,52 @@ struct gated_value { } }; +struct send_move_observation { + bool armed = false; + size_t frame_transfers = 0; +}; + +template +struct inline_move_only_value { + std::array storage{}; + int id = 0; + send_move_observation* observation = nullptr; + bool frame_origin = true; + + inline_move_only_value(int value, send_move_observation* observed) noexcept + : id(value), observation(observed) {} + + inline_move_only_value(const inline_move_only_value&) = delete; + inline_move_only_value& operator=(const inline_move_only_value&) = delete; + + inline_move_only_value(inline_move_only_value&& other) noexcept + : storage(other.storage) + , id(other.id) + , observation(other.observation) + , frame_origin(other.frame_origin) { + finish_move(other); + } + + inline_move_only_value& operator=(inline_move_only_value&& other) noexcept { + storage = other.storage; + id = other.id; + observation = other.observation; + frame_origin = other.frame_origin; + finish_move(other); + return *this; + } + +private: + void finish_move(inline_move_only_value& other) noexcept { + if (frame_origin && observation && observation->armed) { + ++observation->frame_transfers; + frame_origin = false; + } + other.frame_origin = false; + other.id = -1; + } +}; + bool wait_for_true(std::atomic& flag, std::chrono::milliseconds timeout = std::chrono::milliseconds(2000)) { @@ -117,6 +167,308 @@ bool wait_for_condition(Predicate&& predicate, } // namespace +TEST_CASE("channel send frames transfer move-only payloads once", + "[sync][channel][coro][frame_allocation]") { + using payload = inline_move_only_value<>; + + SECTION("ready bounded send") { + channel ch(1); + send_move_observation observed; + auto send_task = ch.send(payload(11, &observed)); + observed.armed = true; + + auto handle = elio::coro::detail::task_access::handle(send_task); + handle.resume(); + + REQUIRE(handle.done()); + REQUIRE(handle.promise().value_.value()); + REQUIRE(observed.frame_transfers == 1); + auto received = ch.try_recv(); + REQUIRE(received.has_value()); + REQUIRE(received->id == 11); + REQUIRE(observed.frame_transfers == 1); + } + + SECTION("ready unbounded send") { + auto ch = channel::unbounded(); + send_move_observation observed; + auto send_task = ch.send(payload(17, &observed)); + observed.armed = true; + + auto handle = elio::coro::detail::task_access::handle(send_task); + handle.resume(); + + REQUIRE(handle.done()); + REQUIRE(handle.promise().value_.value()); + REQUIRE(observed.frame_transfers == 1); + auto received = ch.try_recv(); + REQUIRE(received.has_value()); + REQUIRE(received->id == 17); + REQUIRE(observed.frame_transfers == 1); + } + + SECTION("bounded parked sender refill") { + channel ch(1); + send_move_observation filler_observed; + send_move_observation observed; + REQUIRE(ch.try_send(payload(1, &filler_observed))); + + auto send_task = ch.send(payload(12, &observed)); + observed.armed = true; + auto handle = elio::coro::detail::task_access::handle(send_task); + handle.resume(); + REQUIRE_FALSE(handle.done()); + REQUIRE(observed.frame_transfers == 0); + + auto first = ch.try_recv(); + REQUIRE(first.has_value()); + REQUIRE(first->id == 1); + REQUIRE(handle.done()); + REQUIRE(observed.frame_transfers == 1); + auto second = ch.try_recv(); + REQUIRE(second.has_value()); + REQUIRE(second->id == 12); + REQUIRE(observed.frame_transfers == 1); + } + + SECTION("rendezvous receiver steals parked sender") { + channel ch; + send_move_observation observed; + auto send_task = ch.send(payload(13, &observed)); + observed.armed = true; + auto handle = elio::coro::detail::task_access::handle(send_task); + handle.resume(); + REQUIRE_FALSE(handle.done()); + REQUIRE(observed.frame_transfers == 0); + + auto received = ch.try_recv(); + REQUIRE(received.has_value()); + REQUIRE(received->id == 13); + REQUIRE(handle.done()); + REQUIRE(observed.frame_transfers == 1); + } + + SECTION("close drains parked sender") { + channel ch; + send_move_observation observed; + auto send_task = ch.send(payload(14, &observed)); + observed.armed = true; + auto handle = elio::coro::detail::task_access::handle(send_task); + handle.resume(); + REQUIRE_FALSE(handle.done()); + + ch.close(); + REQUIRE(handle.done()); + REQUIRE(handle.promise().value_.value()); + REQUIRE(observed.frame_transfers == 1); + auto received = ch.try_recv(); + REQUIRE(received.has_value()); + REQUIRE(received->id == 14); + REQUIRE(observed.frame_transfers == 1); + } + + SECTION("cancellation winner leaves frame payload untouched") { + channel ch; + cancel_source source; + send_move_observation observed; + auto send_task = ch.send( + payload(15, &observed), source.get_token()); + observed.armed = true; + auto handle = elio::coro::detail::task_access::handle(send_task); + handle.resume(); + REQUIRE_FALSE(handle.done()); + REQUIRE(observed.frame_transfers == 0); + + source.cancel(); + REQUIRE(handle.done()); + REQUIRE(handle.promise().value_->was_cancelled()); + REQUIRE(observed.frame_transfers == 0); + REQUIRE_FALSE(ch.try_recv().has_value()); + } + + SECTION("active-token ready send transfers once") { + channel ch(1); + cancel_source source; + send_move_observation observed; + auto send_task = ch.send( + payload(16, &observed), source.get_token()); + observed.armed = true; + auto handle = elio::coro::detail::task_access::handle(send_task); + handle.resume(); + + REQUIRE(handle.done()); + REQUIRE(handle.promise().value_->success()); + REQUIRE(observed.frame_transfers == 1); + auto received = ch.try_recv(); + REQUIRE(received.has_value()); + REQUIRE(received->id == 16); + REQUIRE(observed.frame_transfers == 1); + } + + SECTION("active-token parked send transfers once after refill") { + channel ch(1); + send_move_observation filler_observed; + send_move_observation observed; + REQUIRE(ch.try_send(payload(1, &filler_observed))); + cancel_source source; + auto send_task = ch.send( + payload(18, &observed), source.get_token()); + observed.armed = true; + auto handle = elio::coro::detail::task_access::handle(send_task); + handle.resume(); + REQUIRE_FALSE(handle.done()); + REQUIRE(observed.frame_transfers == 0); + + auto first = ch.try_recv(); + REQUIRE(first.has_value()); + REQUIRE(first->id == 1); + REQUIRE(handle.done()); + REQUIRE(handle.promise().value_->success()); + REQUIRE(observed.frame_transfers == 1); + auto second = ch.try_recv(); + REQUIRE(second.has_value()); + REQUIRE(second->id == 18); + REQUIRE(observed.frame_transfers == 1); + } +} + +TEST_CASE("channel send races move frame payload only for delivery winners", + "[sync][channel][cancellation][race][frame_allocation]") { + using payload = inline_move_only_value<64>; + + SECTION("cancellation versus bounded refill") { + for (int iteration = 0; iteration < 64; ++iteration) { + channel ch(1); + send_move_observation filler_observed; + send_move_observation observed; + REQUIRE(ch.try_send(payload(1, &filler_observed))); + cancel_source source; + auto send_task = ch.send( + payload(30 + iteration, &observed), source.get_token()); + observed.armed = true; + auto handle = elio::coro::detail::task_access::handle(send_task); + handle.resume(); + REQUIRE_FALSE(handle.done()); + + std::optional first; + std::barrier start(3); + std::thread canceller([&] { + start.arrive_and_wait(); + source.cancel(); + }); + std::thread receiver([&] { + start.arrive_and_wait(); + first = ch.try_recv(); + }); + start.arrive_and_wait(); + canceller.join(); + receiver.join(); + + REQUIRE(first.has_value()); + REQUIRE(first->id == 1); + REQUIRE(handle.done()); + const auto& result = handle.promise().value_.value(); + if (result.success()) { + REQUIRE(observed.frame_transfers == 1); + auto delivered = ch.try_recv(); + REQUIRE(delivered.has_value()); + REQUIRE(delivered->id == 30 + iteration); + } else { + REQUIRE(result.was_cancelled()); + REQUIRE(observed.frame_transfers == 0); + REQUIRE_FALSE(ch.try_recv().has_value()); + } + } + } + + SECTION("cancellation versus close drain") { + for (int iteration = 0; iteration < 64; ++iteration) { + channel ch; + send_move_observation observed; + cancel_source source; + auto send_task = ch.send( + payload(100 + iteration, &observed), source.get_token()); + observed.armed = true; + auto handle = elio::coro::detail::task_access::handle(send_task); + handle.resume(); + REQUIRE_FALSE(handle.done()); + + std::barrier start(3); + std::thread canceller([&] { + start.arrive_and_wait(); + source.cancel(); + }); + std::thread closer([&] { + start.arrive_and_wait(); + ch.close(); + }); + start.arrive_and_wait(); + canceller.join(); + closer.join(); + + REQUIRE(handle.done()); + const auto& result = handle.promise().value_.value(); + auto delivered = ch.try_recv(); + if (result.success()) { + REQUIRE(observed.frame_transfers == 1); + REQUIRE(delivered.has_value()); + REQUIRE(delivered->id == 100 + iteration); + } else { + REQUIRE(result.was_cancelled()); + REQUIRE(observed.frame_transfers == 0); + REQUIRE_FALSE(delivered.has_value()); + } + } + } +} + +TEST_CASE("public channel send awaiters retain owned move-only values", + "[sync][channel][cancellation][lifetime]") { + using payload = inline_move_only_value<64>; + using owning_awaiter = channel::send_awaitable; + STATIC_REQUIRE((std::is_base_of_v< + elio::detail::intrusive_list_node, owning_awaiter>)); + + SECTION("ordinary awaiter") { + channel ch; + send_move_observation observed; + std::optional::send_awaitable> sender; + { + payload source(21, &observed); + sender.emplace(ch, std::move(source)); + } + + REQUIRE_FALSE(sender->await_ready()); + REQUIRE(sender->await_suspend(std::noop_coroutine())); + REQUIRE(sender->is_linked()); + auto received = ch.try_recv(); + REQUIRE(received.has_value()); + REQUIRE(received->id == 21); + REQUIRE_FALSE(sender->is_linked()); + REQUIRE(sender->await_resume()); + } + + SECTION("cancellable awaiter remains an owning send awaiter") { + channel ch; + send_move_observation observed; + std::optional::cancellable_send_awaitable> sender; + { + payload source(22, &observed); + sender.emplace(ch, std::move(source), cancel_token{}); + } + + channel::send_awaitable& base = *sender; + REQUIRE_FALSE(base.await_ready()); + REQUIRE(sender->await_suspend(std::noop_coroutine())); + REQUIRE(base.is_linked()); + auto received = ch.try_recv(); + REQUIRE(received.has_value()); + REQUIRE(received->id == 22); + REQUIRE_FALSE(base.is_linked()); + REQUIRE(sender->await_resume().success()); + } +} + // --------------------------------------------------------------------------- // Test 1: recv() direct steal from blocked sender on bounded channel // --------------------------------------------------------------------------- diff --git a/wiki/API-Reference.md b/wiki/API-Reference.md index 820fe7df..cade8be1 100644 --- a/wiki/API-Reference.md +++ b/wiki/API-Reference.md @@ -3514,6 +3514,15 @@ public: }; ``` +Both `send` overloads take their payload by value, so the returned task owns it +in the coroutine frame until delivery, cancellation, closure, or task +destruction resolves the operation. The internal queued waiter borrows that +frame-owned object and transfers it only after normal completion wins. It does +not add a second `T` to the frame. This is an implementation detail of the task +wrappers: directly constructed `send_awaitable` and +`cancellable_send_awaitable` objects continue to own their payload and may +outlive the expression that supplied it. + ### `semaphore` Counting semaphore. diff --git a/wiki/Performance-Tuning.md b/wiki/Performance-Tuning.md index 1b9f927d..44909ad5 100644 --- a/wiki/Performance-Tuning.md +++ b/wiki/Performance-Tuning.md @@ -536,6 +536,25 @@ bool pushed = ring.try_push(value); auto popped = ring.try_pop(); ``` +The `channel::send(...)` task stores one by-value `T` in its coroutine frame. +When a bounded or rendezvous send must wait, its intrusive waiter borrows that +same frame-owned object rather than moving it into a second `T` subobject. The +reduction is approximately one payload region for large inline types and +applies to both ordinary and token-aware sends. It does not change the delivery +move: the payload is transferred only after normal completion wins. Public +awaiter objects constructed directly by callers still own one independent +payload so their lifetime does not depend on a constructor argument. + +`bench_channel_send_frame` reports requested coroutine-frame bytes, +allocator-usable bytes where the platform exposes them, +construction/destruction cost for naturally aligned 8/64/256/1024-byte inline +payloads, ready bounded/unbounded sends, and forced bounded-full/rendezvous +handoffs with and without active tokens. Its allocation recorder is isolated +from the ordinary `bench_channel` executable. Compare Release builds with +pinned, interleaved baseline/candidate samples; the benchmark intentionally +does not impose timing thresholds on shared CI runners. Pass `--smoke` for +reduced iteration counts when validating a Debug build. + ## Network Performance ### Connection Pooling