From b73aabc174b720f9473e226973950e72645f2379 Mon Sep 17 00:00:00 2001 From: tzcnt Date: Thu, 25 Jun 2026 20:48:24 -0700 Subject: [PATCH] tests for manual_reset_event set() / co_set() bulk post behavior --- tests/test_manual_reset_event.cpp | 169 ++++++++++++++++++++++++-- tests/test_spawn_func_many_common.hpp | 2 + tests/test_spawn_many_common.hpp | 2 + 3 files changed, 160 insertions(+), 13 deletions(-) diff --git a/tests/test_manual_reset_event.cpp b/tests/test_manual_reset_event.cpp index dcf7c23..1811691 100644 --- a/tests/test_manual_reset_event.cpp +++ b/tests/test_manual_reset_event.cpp @@ -7,27 +7,64 @@ #include #include +#include #define CATEGORY test_manual_reset_event class CATEGORY : public testing::Test { protected: + // A second executor, used to exercise waking waiters that resume on + // different executors. + static inline tmc::ex_cpu otherExec; + static void SetUpTestSuite() { tmc::cpu_executor().set_thread_count(4).set_priority_count(2).init(); + otherExec.set_thread_count(4).set_priority_count(2).init(); } - static void TearDownTestSuite() { tmc::cpu_executor().teardown(); } + static void TearDownTestSuite() { + otherExec.teardown(); + tmc::cpu_executor().teardown(); + } static tmc::ex_cpu& ex() { return tmc::cpu_executor(); } using waiter_count_accessor = tmc::tests::waiter_count_accessor; }; +// Suspends on Event, then increments AA when resumed. +static tmc::task +wait_and_inc(tmc::manual_reset_event& Event, atomic_awaitable& AA) { + co_await Event; + AA.inc(); +} + +// Suspends on Event, then asserts it resumed on ExpectedExec before +// incrementing AA. +static tmc::task wait_check_exec_and_inc( + tmc::manual_reset_event& Event, atomic_awaitable& AA, + tmc::ex_any* ExpectedExec +) { + co_await Event; + EXPECT_EQ(tmc::current_executor(), ExpectedExec); + AA.inc(); +} + +// Suspends on Event, then asserts it resumed at ExpectedPrio before +// incrementing AA. +static tmc::task wait_check_prio_and_inc( + tmc::manual_reset_event& Event, atomic_awaitable& AA, size_t ExpectedPrio +) { + co_await Event; + EXPECT_EQ(tmc::current_priority(), ExpectedPrio); + AA.inc(); +} + TEST_F(CATEGORY, no_waiters) { test_async_main(ex(), []() -> tmc::task { tmc::manual_reset_event event(false); EXPECT_FALSE(event.is_set()); - event.set(); + EXPECT_EQ(event.set(), 0u); EXPECT_TRUE(event.is_set()); co_return; }()); @@ -37,7 +74,7 @@ TEST_F(CATEGORY, no_waiters_co_set) { test_async_main(ex(), []() -> tmc::task { tmc::manual_reset_event event(false); EXPECT_FALSE(event.is_set()); - co_await event.co_set(); + EXPECT_EQ(co_await event.co_set(), 0u); EXPECT_TRUE(event.is_set()); co_return; }()); @@ -53,11 +90,11 @@ TEST_F(CATEGORY, nonblocking) { EXPECT_EQ(event.is_set(), false); event.reset(); EXPECT_EQ(event.is_set(), false); - event.set(); + EXPECT_EQ(event.set(), 0u); EXPECT_EQ(event.is_set(), true); - event.set(); + EXPECT_EQ(event.set(), 0u); EXPECT_EQ(event.is_set(), true); - co_await event.co_set(); + EXPECT_EQ(co_await event.co_set(), 0u); EXPECT_EQ(event.is_set(), true); co_await event; EXPECT_EQ(event.is_set(), true); @@ -83,7 +120,7 @@ TEST_F(CATEGORY, one_waiter) { co_await waiter_count_accessor::wait_for_waiter_count(event, 1); EXPECT_EQ(event.is_set(), false); EXPECT_EQ(aa.load(), 0); - event.set(); + EXPECT_EQ(event.set(), 1u); co_await aa; co_await std::move(t); } @@ -102,7 +139,7 @@ TEST_F(CATEGORY, one_waiter) { co_await waiter_count_accessor::wait_for_waiter_count(event, 1); EXPECT_EQ(event.is_set(), false); EXPECT_EQ(aa.load(), 0); - co_await event.co_set(); + EXPECT_EQ(co_await event.co_set(), 1u); co_await aa; co_await std::move(t); } @@ -127,7 +164,7 @@ TEST_F(CATEGORY, multi_waiter) { auto t = tmc::spawn_many(tasks).fork(); co_await waiter_count_accessor::wait_for_waiter_count(event, 5); EXPECT_EQ(aa.load(), 0); - event.set(); + EXPECT_EQ(event.set(), 5u); co_await waiter_count_accessor::wait_for_waiter_count(event, 0); co_await aa; EXPECT_EQ(aa.load(), 5); @@ -148,7 +185,7 @@ TEST_F(CATEGORY, multi_waiter) { auto t = tmc::spawn_many(tasks).fork(); co_await waiter_count_accessor::wait_for_waiter_count(event, 5); EXPECT_EQ(aa.load(), 0); - co_await event.co_set(); + EXPECT_EQ(co_await event.co_set(), 5u); co_await waiter_count_accessor::wait_for_waiter_count(event, 0); co_await aa; EXPECT_EQ(aa.load(), 5); @@ -188,11 +225,11 @@ TEST_F(CATEGORY, co_set) { tmc::manual_reset_event event; { EXPECT_EQ(event.is_set(), false); - co_await event.co_set(); + EXPECT_EQ(co_await event.co_set(), 0u); EXPECT_EQ(event.is_set(), true); event.reset(); EXPECT_EQ(event.is_set(), false); - co_await event.co_set(); + EXPECT_EQ(co_await event.co_set(), 0u); EXPECT_EQ(event.is_set(), true); } event.reset(); @@ -210,7 +247,7 @@ TEST_F(CATEGORY, co_set) { co_await waiter_count_accessor::wait_for_waiter_count(event, 1); EXPECT_EQ(event.is_set(), false); EXPECT_EQ(aa.load(), 0); - co_await event.co_set(); + EXPECT_EQ(co_await event.co_set(), 1u); co_await aa; co_await std::move(t); } @@ -249,4 +286,110 @@ TEST_F(CATEGORY, co_set_no_symmetric) { }()); } +// Wake more than one batch worth of waiters (the batch size is 64). This +// exercises posting a full batch and then a partial trailing batch. +TEST_F(CATEGORY, many_waiters) { + test_async_main(ex(), []() -> tmc::task { + // 200 spans three full batches (192) plus a partial trailing batch. + static constexpr size_t COUNT = 200; + tmc::manual_reset_event event; + EXPECT_EQ(event.is_set(), false); + { + atomic_awaitable aa(COUNT); + std::vector> tasks(COUNT); + for (size_t i = 0; i < COUNT; ++i) { + tasks[i] = wait_and_inc(event, aa); + } + auto t = tmc::spawn_many(tasks.data(), COUNT).fork(); + co_await waiter_count_accessor::wait_for_waiter_count(event, COUNT); + EXPECT_EQ(aa.load(), 0); + EXPECT_EQ(event.set(), COUNT); + co_await aa; + EXPECT_EQ(aa.load(), static_cast(COUNT)); + co_await std::move(t); + } + event.reset(); + // Repeat via co_set(), which wakes the most-recently-added waiter by + // symmetric transfer and posts the remainder in batches. + { + atomic_awaitable aa(COUNT); + std::vector> tasks(COUNT); + for (size_t i = 0; i < COUNT; ++i) { + tasks[i] = wait_and_inc(event, aa); + } + auto t = tmc::spawn_many(tasks.data(), COUNT).fork(); + co_await waiter_count_accessor::wait_for_waiter_count(event, COUNT); + EXPECT_EQ(aa.load(), 0); + EXPECT_EQ(co_await event.co_set(), COUNT); + co_await aa; + EXPECT_EQ(aa.load(), static_cast(COUNT)); + co_await std::move(t); + } + }()); +} + +// Wake waiters that resume on different executors. A batch can only be posted +// to a single executor, so a new batch must be started whenever the executor +// changes. +TEST_F(CATEGORY, different_executors) { + test_async_main(ex(), []() -> tmc::task { + // Enough waiters per executor that the interleaving also crosses a batch + // boundary (64). + static constexpr size_t PER = 50; + tmc::manual_reset_event event; + EXPECT_EQ(event.is_set(), false); + { + atomic_awaitable aa(2 * PER); + std::vector> tasksA(PER); + std::vector> tasksB(PER); + for (size_t i = 0; i < PER; ++i) { + tasksA[i] = + wait_check_exec_and_inc(event, aa, tmc::cpu_executor().type_erased()); + tasksB[i] = wait_check_exec_and_inc(event, aa, otherExec.type_erased()); + } + auto ta = + tmc::spawn_many(tasksA.data(), PER).run_on(tmc::cpu_executor()).fork(); + auto tb = tmc::spawn_many(tasksB.data(), PER).run_on(otherExec).fork(); + co_await waiter_count_accessor::wait_for_waiter_count(event, 2 * PER); + EXPECT_EQ(aa.load(), 0); + EXPECT_EQ(event.set(), 2 * PER); + co_await aa; + EXPECT_EQ(aa.load(), static_cast(2 * PER)); + co_await std::move(ta); + co_await std::move(tb); + } + }()); +} + +// Wake waiters that resume at different priorities. A batch can only be posted +// at a single priority, so a new batch must be started whenever the priority +// changes. +TEST_F(CATEGORY, different_priorities) { + test_async_main(ex(), []() -> tmc::task { + // Enough waiters per priority that the interleaving also crosses a batch + // boundary (64). + static constexpr size_t PER = 50; + tmc::manual_reset_event event; + EXPECT_EQ(event.is_set(), false); + { + atomic_awaitable aa(2 * PER); + std::vector> tasks0(PER); + std::vector> tasks1(PER); + for (size_t i = 0; i < PER; ++i) { + tasks0[i] = wait_check_prio_and_inc(event, aa, 0); + tasks1[i] = wait_check_prio_and_inc(event, aa, 1); + } + auto t0 = tmc::spawn_many(tasks0.data(), PER).with_priority(0).fork(); + auto t1 = tmc::spawn_many(tasks1.data(), PER).with_priority(1).fork(); + co_await waiter_count_accessor::wait_for_waiter_count(event, 2 * PER); + EXPECT_EQ(aa.load(), 0); + EXPECT_EQ(event.set(), 2 * PER); + co_await aa; + EXPECT_EQ(aa.load(), static_cast(2 * PER)); + co_await std::move(t0); + co_await std::move(t1); + } + }()); +} + #undef CATEGORY diff --git a/tests/test_spawn_func_many_common.hpp b/tests/test_spawn_func_many_common.hpp index c23aebb..e57ac23 100644 --- a/tests/test_spawn_func_many_common.hpp +++ b/tests/test_spawn_func_many_common.hpp @@ -20,6 +20,8 @@ template auto func_iter_of_static_size() { // This iterator produces a dynamic number of tasks, // which can be calculated by the caller in O(1) time by // `return.end() - return.begin()` +// (it's the same set of tasks as the unknown iterator, but collected into a vector, which +// allows for size calculation) template auto func_iter_of_dynamic_known_size() { auto iter = std::ranges::views::iota(0, N) | std::ranges::views::filter(unpredictable_filter) | diff --git a/tests/test_spawn_many_common.hpp b/tests/test_spawn_many_common.hpp index 9096920..292c6e3 100644 --- a/tests/test_spawn_many_common.hpp +++ b/tests/test_spawn_many_common.hpp @@ -17,6 +17,8 @@ template auto iter_of_static_size() { // This iterator produces a dynamic number of tasks, // which can be calculated by the caller in O(1) time by // `return.end() - return.begin()` +// (it's the same set of tasks as the unknown iterator, but collected into a vector, which +// allows for size calculation) template auto iter_of_dynamic_known_size() { auto iter = std::ranges::views::iota(0, N) | std::ranges::views::filter(unpredictable_filter) |