diff --git a/include/tmc/detail/manual_reset_event.ipp b/include/tmc/detail/manual_reset_event.ipp index 5254a54c..bca43795 100644 --- a/include/tmc/detail/manual_reset_event.ipp +++ b/include/tmc/detail/manual_reset_event.ipp @@ -11,13 +11,59 @@ #include "tmc/detail/waiter_list.hpp" #include "tmc/manual_reset_event.hpp" +#include #include #include namespace tmc { -bool aw_manual_reset_event::await_suspend( - std::coroutine_handle<> Outer -) noexcept { +namespace detail { +// Submits a chain of waiters to their continuation executors, grouping +// consecutive waiters that share an executor and priority into bulk posts. +// Returns the number of waiters woken. +inline size_t wake_waiters_in_batches(tmc::detail::waiter_list_node* curr) noexcept { + std::array batch; + size_t batchSize = 0; + tmc::ex_any* continuationExecutor = nullptr; + size_t continuationPriority = 0; + size_t wakeCount = 0; + + while (curr != nullptr) { + auto next = curr->next; + + // A batch can only be posted to a single executor at a single priority. + // If this waiter's executor or priority differs from the batch in + // progress, post that batch before starting a new one for this waiter. + if (batchSize != 0 && (curr->waiter.continuation_executor != continuationExecutor || + curr->waiter.continuation_priority != continuationPriority)) { + continuationExecutor->post_bulk(batch.data(), batchSize, continuationPriority); + batchSize = 0; + } + + if (batchSize == 0) { + continuationExecutor = curr->waiter.continuation_executor; + continuationPriority = curr->waiter.continuation_priority; + } + + batch[batchSize] = curr->waiter.continuation; + ++batchSize; + ++wakeCount; + if (batchSize == batch.size()) { + continuationExecutor->post_bulk(batch.data(), batchSize, continuationPriority); + batchSize = 0; + } + + curr = next; + } + + if (batchSize != 0) { + continuationExecutor->post_bulk(batch.data(), batchSize, continuationPriority); + } + + return wakeCount; +} +} // namespace detail + +bool aw_manual_reset_event::await_suspend(std::coroutine_handle<> Outer) noexcept { me.waiter.continuation = Outer; me.waiter.continuation_executor = tmc::detail::this_thread::executor(); me.waiter.continuation_priority = tmc::detail::this_thread::this_task().prio; @@ -35,11 +81,9 @@ bool aw_manual_reset_event::await_suspend( return true; } -std::coroutine_handle<> aw_manual_reset_event_co_set::await_suspend( - std::coroutine_handle<> Outer -) noexcept { - auto h = - parent.head.exchange(manual_reset_event::READY, std::memory_order_acq_rel); +std::coroutine_handle<> +aw_manual_reset_event_co_set::await_suspend(std::coroutine_handle<> Outer) noexcept { + auto h = parent.head.exchange(manual_reset_event::READY, std::memory_order_acq_rel); if (manual_reset_event::READY == h || manual_reset_event::NOT_READY == h) { // It was ready, or there are no waiters - just resume return Outer; @@ -48,12 +92,9 @@ std::coroutine_handle<> aw_manual_reset_event_co_set::await_suspend( // Save the first / most recently added waiter for symmetric transfer. auto toWake = reinterpret_cast(h); - auto curr = toWake->next; - while (curr != nullptr) { - auto next = curr->next; - curr->waiter.resume(); - curr = next; - } + // Wake the remaining waiters in batches, then count toWake itself, which is + // woken below by symmetric transfer. + wakeCount = 1 + tmc::detail::wake_waiters_in_batches(toWake->next); return toWake->waiter.try_symmetric_transfer(Outer); } @@ -72,18 +113,16 @@ size_t manual_reset_event::waiter_count() noexcept { return count; } -void manual_reset_event::set() noexcept { +size_t manual_reset_event::set() noexcept { auto h = head.exchange(READY, std::memory_order_acq_rel); if (READY == h) { // It was ready, nothing to wake - return; - } - auto curr = reinterpret_cast(h); - while (curr != nullptr) { - auto next = curr->next; - curr->waiter.resume(); - curr = next; + return 0; } + + return tmc::detail::wake_waiters_in_batches( + reinterpret_cast(h) + ); } manual_reset_event::~manual_reset_event() noexcept { set(); } diff --git a/include/tmc/manual_reset_event.hpp b/include/tmc/manual_reset_event.hpp index fd9b9b44..561d6816 100644 --- a/include/tmc/manual_reset_event.hpp +++ b/include/tmc/manual_reset_event.hpp @@ -6,6 +6,7 @@ #pragma once #include "tmc/detail/impl.hpp" // IWYU pragma: keep +#include "tmc/detail/compat.hpp" #include "tmc/detail/concepts_awaitable.hpp" #include "tmc/detail/waiter_list.hpp" @@ -26,8 +27,7 @@ class aw_manual_reset_event { friend class manual_reset_event; - inline aw_manual_reset_event(manual_reset_event& Parent) noexcept - : parent(Parent) {} + inline aw_manual_reset_event(manual_reset_event& Parent) noexcept : parent(Parent) {} public: inline bool await_ready() noexcept { return false; } @@ -47,27 +47,26 @@ class [[nodiscard( "You must co_await aw_manual_reset_event_co_set for it to have any effect." )]] aw_manual_reset_event_co_set : tmc::detail::AwaitTagNoGroupAsIs { manual_reset_event& parent; + size_t wakeCount; friend class manual_reset_event; inline aw_manual_reset_event_co_set(manual_reset_event& Parent) noexcept - : parent(Parent) {} + : parent(Parent), wakeCount(0) {} public: inline bool await_ready() noexcept { return false; } - TMC_DECL std::coroutine_handle<> - await_suspend(std::coroutine_handle<> Outer) noexcept; + TMC_DECL std::coroutine_handle<> await_suspend(std::coroutine_handle<> Outer) noexcept; - inline void await_resume() noexcept {} + /// Returns the number of awaiters that were woken. + inline size_t await_resume() noexcept { return wakeCount; } // Copy/move constructors *could* be implemented, but why? aw_manual_reset_event_co_set(aw_manual_reset_event_co_set const&) = delete; - aw_manual_reset_event_co_set& - operator=(aw_manual_reset_event_co_set const&) = delete; + aw_manual_reset_event_co_set& operator=(aw_manual_reset_event_co_set const&) = delete; aw_manual_reset_event_co_set(aw_manual_reset_event_co_set&&) = delete; - aw_manual_reset_event_co_set& - operator=(aw_manual_reset_event_co_set&&) = delete; + aw_manual_reset_event_co_set& operator=(aw_manual_reset_event_co_set&&) = delete; }; /// An async version of Windows ManualResetEvent. @@ -93,8 +92,7 @@ class manual_reset_event { public: /// The Ready parameter controls the initial state. - inline manual_reset_event(bool Ready) noexcept - : head(Ready ? READY : NOT_READY) {} + inline manual_reset_event(bool Ready) noexcept : head(Ready ? READY : NOT_READY) {} /// The initial state will be not-set / not-ready. inline manual_reset_event() noexcept : manual_reset_event(false) {} @@ -108,9 +106,7 @@ class manual_reset_event { /// If the event state is already reset, this will do nothing. inline void reset() noexcept { auto expected = READY; - head.compare_exchange_strong( - expected, NOT_READY, std::memory_order_acq_rel - ); + head.compare_exchange_strong(expected, NOT_READY, std::memory_order_acq_rel); // Don't need to check the result of the operation - it becomes not ready // in any case. If there were already waiters, they will not be removed from // the list. @@ -119,8 +115,9 @@ class manual_reset_event { /// All current awaiters will be resumed. /// Any future awaiters will resume immediately. /// If the event state is already set, this will do nothing. - /// Does not symmetric transfer; awaiters will be posted to their executors. - TMC_DECL void set() noexcept; + /// Does not symmetric transfer; awaiters will be posted to their executors. Returns the + /// number of awaiters that were woken. + TMC_DECL size_t set() noexcept; /// All current awaiters will be resumed. /// Any future awaiters will resume immediately.