Skip to content

manual_reset_event::set() and co_set() wake waiters in bulk batches #263

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
tzcnt merged 1 commit into from
Jun 26, 2026
Merged

manual_reset_event::set() and co_set() wake waiters in bulk batches #263

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
83 changes: 61 additions & 22 deletions include/tmc/detail/manual_reset_event.ipp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,13 +11,59 @@
#include "tmc/detail/waiter_list.hpp"
#include "tmc/manual_reset_event.hpp"

#include <array>
#include <atomic>
#include <coroutine>

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<tmc::work_item, 64> 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;
Expand All @@ -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;
Expand All @@ -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<tmc::detail::waiter_list_node*>(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);
}
Expand All @@ -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<tmc::detail::waiter_list_node*>(h);
while (curr != nullptr) {
auto next = curr->next;
curr->waiter.resume();
curr = next;
return 0;
}

return tmc::detail::wake_waiters_in_batches(
reinterpret_cast<tmc::detail::waiter_list_node*>(h)
);
}

manual_reset_event::~manual_reset_event() noexcept { set(); }
Expand Down
31 changes: 14 additions & 17 deletions include/tmc/manual_reset_event.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"

Expand All @@ -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; }
Expand All @@ -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.
Expand All @@ -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) {}
Expand All @@ -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.
Expand All @@ -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.
Expand Down
Loading