#define DOCTEST_CONFIG_IMPLEMENT_WITH_MAIN

#include <doctest.h>

#include <taskflow/taskflow.hpp>

#include <atomic>

#include <thread>

#include <vector>

#include <random>

#include <chrono>

#include <algorithm>

// ============================================================================

// Shared helpers

// ============================================================================

static inline void tiny_jitter(std::mt19937& rng) {

std::uniform_int_distribution<int> pick(0, 9);

int x = pick(rng);

if (x < 4) {

std::this_thread::yield();

} else if (x < 8) {

volatile int sink = 0;

int spins = 20 + (pick(rng) * 30);

for (int i = 0; i < spins; ++i) sink += i;

(void)sink;

} else {

std::this_thread::sleep_for(std::chrono::nanoseconds(200));

}

}

enum class NotificationType { ONE, N, ALL };

// ============================================================================

// no_missing_notify_all

// Each round: all N threads prepare_wait(), pause in the window between

// prepare and commit until all threads are ready, then commit_wait().

// Main calls notify_all() in that window. All commit_wait() calls must

// return without blocking (no lost wakeup).

// End check: completed == R*N, num_waiters() == 0.

// ============================================================================

template <typename T>

void no_missing_notify_all(size_t N) {

T notifier(N);

REQUIRE(notifier.size() == N);

size_t R = 20 * (N + 1);

if (N >= 31) R = 1 * (N + 1);

std::atomic<size_t> prepared(0);

std::atomic<size_t> completed(0);

std::atomic<size_t> round(0);

std::atomic<bool> stop(false);

std::vector<std::thread> threads;

threads.reserve(N);

for (size_t i = 0; i < N; ++i) {

threads.emplace_back([&, i]() {

size_t local_round = 0;

while (!stop.load(std::memory_order_relaxed)) {

while (round.load(std::memory_order_acquire) == local_round &&

!stop.load(std::memory_order_relaxed)) {

std::this_thread::yield();

}

if (stop.load(std::memory_order_relaxed)) break;

notifier.prepare_wait(i);

prepared.fetch_add(1, std::memory_order_relaxed);

while (prepared.load(std::memory_order_acquire) < (local_round + 1) * N &&

!stop.load(std::memory_order_relaxed)) {

std::this_thread::yield();

}

notifier.commit_wait(i);

completed.fetch_add(1, std::memory_order_relaxed);

local_round++;

}

});

}

for (size_t r = 0; r < R; ++r) {

round.fetch_add(1, std::memory_order_release);

while (prepared.load(std::memory_order_acquire) != (r + 1) * N) {

std::this_thread::yield();

}

notifier.notify_all();

}

stop.store(true, std::memory_order_release);

notifier.notify_all();

for (auto& t : threads) t.join();

REQUIRE(completed.load() == R * N);

REQUIRE(notifier.num_waiters() == 0);

}

// ============================================================================

// no_missing_notify_one

// Each round: all N threads prepare_wait() then commit_wait().

// Main calls notify_one() and verifies at least 1 thread unblocks,

// then notify_all() drains the rest. num_waiters() == 0 at end.

// ============================================================================

template <typename T>

void no_missing_notify_one(size_t N) {

T notifier(N);

REQUIRE(notifier.size() == N);

size_t R = 20 * (N + 1);

if (N >= 31) R = 1 * (N + 1);

std::atomic<size_t> round(0);

std::atomic<size_t> prepared(0);

std::atomic<size_t> committed(0);

std::atomic<bool> stop(false);

std::vector<std::thread> threads;

threads.reserve(N);

for (size_t i = 0; i < N; ++i) {

threads.emplace_back([&, i]() {

size_t local_round = 0;

while (!stop.load(std::memory_order_relaxed)) {

while (round.load(std::memory_order_acquire) <= local_round &&

!stop.load(std::memory_order_relaxed)) {

std::this_thread::yield();

}

if (stop.load(std::memory_order_relaxed)) break;

notifier.prepare_wait(i);

prepared.fetch_add(1, std::memory_order_release);

notifier.commit_wait(i);

committed.fetch_add(1, std::memory_order_release);

local_round++;

}

});

}

for (size_t r = 0; r < R; ++r) {

prepared.store(0, std::memory_order_relaxed);

committed.store(0, std::memory_order_relaxed);

round.store(r + 1, std::memory_order_release);

while (prepared.load(std::memory_order_acquire) != N) {

std::this_thread::yield();

}

notifier.notify_one();

// At least one thread must unblock from commit_wait.

while (committed.load() == 0) std::this_thread::yield();

// Drain any remaining waiters.

notifier.notify_all();

while (committed.load(std::memory_order_acquire) != N) {

std::this_thread::yield();

}

}

stop.store(true, std::memory_order_release);

notifier.notify_all();

for (auto& t : threads) t.join();

REQUIRE(notifier.num_waiters() == 0);

}

// ============================================================================

// no_missing_notify_n

// Each round: all N threads prepare then commit_wait. Main calls notify_n(k)

// where k is random in [0, N], verifies at least k unblock, then notify_all

// drains the rest. num_waiters() == 0 at end.

// ============================================================================

template <typename T>

void no_missing_notify_n(size_t N, uint32_t seed = 12345) {

T notifier(N);

REQUIRE(notifier.size() == N);

size_t R = 20 * (N + 1);

if (N >= 31) R = 1 * (N + 1);

std::atomic<size_t> round(0);

std::atomic<size_t> prepared(0);

std::atomic<size_t> committed(0);

std::atomic<bool> stop(false);

std::vector<std::thread> threads;

threads.reserve(N);

for (size_t i = 0; i < N; ++i) {

threads.emplace_back([&, i]() {

size_t local_round = 0;

while (!stop.load(std::memory_order_relaxed)) {

while (round.load(std::memory_order_acquire) <= local_round &&

!stop.load(std::memory_order_relaxed)) {

std::this_thread::yield();

}

if (stop.load(std::memory_order_relaxed)) break;

notifier.prepare_wait(i);

prepared.fetch_add(1, std::memory_order_release);

notifier.commit_wait(i);

committed.fetch_add(1, std::memory_order_release);

local_round++;

}

});

}

std::mt19937 rng(seed);

std::uniform_int_distribution<size_t> dist(0, N);

for (size_t r = 0; r < R; ++r) {

prepared.store(0, std::memory_order_relaxed);

committed.store(0, std::memory_order_relaxed);

round.store(r + 1, std::memory_order_release);

while (prepared.load(std::memory_order_acquire) != N) {

std::this_thread::yield();

}

size_t k = dist(rng);

notifier.notify_n(k);

while (committed.load(std::memory_order_acquire) < k) {

std::this_thread::yield();

}

notifier.notify_all();

while (committed.load(std::memory_order_acquire) != N) {

std::this_thread::yield();

}

}

stop.store(true, std::memory_order_release);

notifier.notify_all();

for (auto& t : threads) t.join();

REQUIRE(notifier.num_waiters() == 0);

}

// ============================================================================

// no_missing_notify_x (mixed: notify_one / notify_n / notify_all)

// Same structure as no_missing_notify_n but randomly picks notify variant

// each round. Verifies all variants can coexist without losing wakeups.

// ============================================================================

template <typename T>

void no_missing_notify_x(size_t N, uint32_t seed = 12345) {

T notifier(N);

REQUIRE(notifier.size() == N);

size_t R = 20 * (N + 1);

if (N >= 31) R = 1 * (N + 1);

std::atomic<size_t> round(0);

std::atomic<size_t> prepared(0);

std::atomic<size_t> committed(0);

std::atomic<bool> stop(false);

std::vector<std::thread> threads;

threads.reserve(N);

for (size_t i = 0; i < N; ++i) {

threads.emplace_back([&, i]() {

size_t local_round = 0;

while (!stop.load(std::memory_order_relaxed)) {

while (round.load(std::memory_order_acquire) <= local_round &&

!stop.load(std::memory_order_relaxed)) {

std::this_thread::yield();

}

if (stop.load(std::memory_order_relaxed)) break;

notifier.prepare_wait(i);

prepared.fetch_add(1, std::memory_order_release);

notifier.commit_wait(i);

committed.fetch_add(1, std::memory_order_release);

local_round++;

}

});

}

std::mt19937 rng(seed);

std::uniform_int_distribution<size_t> dist(0, N);

for (size_t r = 0; r < R; ++r) {

prepared.store(0, std::memory_order_relaxed);

committed.store(0, std::memory_order_relaxed);

round.store(r + 1, std::memory_order_release);

while (prepared.load(std::memory_order_acquire) != N) {

std::this_thread::yield();

}

size_t k = dist(rng);

if (k == 1) {

notifier.notify_one();

} else if (k == N) {

notifier.notify_all();

} else {

notifier.notify_n(k);

}

while (committed.load(std::memory_order_acquire) < k) {

std::this_thread::yield();

}

notifier.notify_all();

while (committed.load(std::memory_order_acquire) != N) {

std::this_thread::yield();

}

}

stop.store(true, std::memory_order_release);

notifier.notify_all();

for (auto& t : threads) t.join();

REQUIRE(notifier.num_waiters() == 0);

}

// ============================================================================

// no_missing_cancel_wait

// Each round: every thread calls prepare_wait(). Main randomly assigns

// per-thread has_work flag. Threads with has_work cancel, rest commit.

// Verifies cancel_wait cleans up state (no ghost waiters) and that

// commit_wait + notifications complete the round. num_waiters() == 0

// after every round and at end.

// ============================================================================

template <typename T>

void no_missing_cancel_wait(size_t N, uint32_t seed = 12345) {

T notifier(N);

REQUIRE(notifier.size() == N);

size_t R = 20 * (N + 1);

if (N >= 31) R = 1 * (N + 1);

std::atomic<size_t> round(0);

std::atomic<size_t> prepared(0);

std::atomic<size_t> canceled(0);

std::atomic<size_t> committed(0);

std::atomic<bool> stop(false);

std::atomic<bool> go(false);

std::vector<std::atomic<bool>> has_work(N);

for (size_t i = 0; i < N; ++i) has_work[i].store(false, std::memory_order_relaxed);

std::vector<std::thread> threads;

threads.reserve(N);

for (size_t i = 0; i < N; ++i) {

threads.emplace_back([&, i]() {

size_t local_round = 0;

while (!stop.load(std::memory_order_relaxed)) {

while (round.load(std::memory_order_acquire) <= local_round &&

!stop.load(std::memory_order_relaxed)) {

std::this_thread::yield();

}

if (stop.load(std::memory_order_relaxed)) break;

notifier.prepare_wait(i);

prepared.fetch_add(1, std::memory_order_release);

while (!go.load(std::memory_order_acquire) &&

!stop.load(std::memory_order_relaxed)) {

std::this_thread::yield();

}

if (stop.load(std::memory_order_relaxed)) break;

if (has_work[i].load(std::memory_order_acquire)) {

notifier.cancel_wait(i);

canceled.fetch_add(1, std::memory_order_release);

} else {

notifier.commit_wait(i);

committed.fetch_add(1, std::memory_order_release);

}

local_round++;

}

});

}

std::mt19937 rng(seed);

std::uniform_int_distribution<int> dice(0, 1);

for (size_t r = 0; r < R; ++r) {

prepared.store(0, std::memory_order_relaxed);

canceled.store(0, std::memory_order_relaxed);

committed.store(0, std::memory_order_relaxed);

go.store(false, std::memory_order_release);

for (size_t i = 0; i < N; ++i) has_work[i].store(false, std::memory_order_relaxed);

round.store(r + 1, std::memory_order_release);

while (prepared.load(std::memory_order_acquire) != N) std::this_thread::yield();

size_t expected_cancels = 0;

for (size_t i = 0; i < N; ++i) {

if (dice(rng)) {

has_work[i].store(true, std::memory_order_release);

++expected_cancels;

}

}

size_t expected_commits = N - expected_cancels;

go.store(true, std::memory_order_release);

while (canceled.load(std::memory_order_acquire) != expected_cancels) std::this_thread::yield();

while (notifier.num_waiters() != expected_commits) std::this_thread::yield();

if (expected_commits > 0) {

std::uniform_int_distribution<size_t> pickX(0, expected_commits);

size_t X = pickX(rng);

size_t first = expected_commits - X;

notifier.notify_n(first);

for (size_t j = 0; j < X; ++j) notifier.notify_one();

}

// Ensure the round always completes.

notifier.notify_all();

while ((canceled.load(std::memory_order_acquire) +

committed.load(std::memory_order_acquire)) != N) {

std::this_thread::yield();

}

REQUIRE(notifier.num_waiters() == 0);

}

stop.store(true, std::memory_order_release);

go.store(true, std::memory_order_release);

notifier.notify_all();

for (auto& t : threads) t.join();

REQUIRE(notifier.num_waiters() == 0);

}

// ============================================================================

// no_missing_notifications (compile-time switch: ONE / N / ALL)

// M concurrent notifier threads continuously hammer one notify variant.

// N worker threads follow prepare -> cancel_or_commit protocol.

// Round ends when every worker has either canceled or returned from commit.

// num_waiters() == 0 after every round.

// ============================================================================

template <typename T, NotificationType NT>

void no_missing_notifications(size_t N, size_t M = 4, uint32_t seed = 12345) {

T notifier(N);

REQUIRE(notifier.size() == N);

size_t R = 20 * (N + 1);

if (N >= 31) R = 1 * (N + 1);

std::atomic<size_t> round(0);

std::atomic<size_t> prepared(0);

std::atomic<size_t> canceled(0);

std::atomic<size_t> committed(0);

std::atomic<bool> stop(false);

std::atomic<bool> go(false);

std::vector<std::atomic<bool>> has_work(N);

for (size_t i = 0; i < N; ++i) has_work[i].store(false, std::memory_order_relaxed);

std::vector<std::thread> workers;

workers.reserve(N);

for (size_t i = 0; i < N; ++i) {

workers.emplace_back([&, i]() {

size_t local_round = 0;

while (!stop.load(std::memory_order_relaxed)) {

while (round.load(std::memory_order_acquire) <= local_round &&

!stop.load(std::memory_order_relaxed)) {

std::this_thread::yield();

}

if (stop.load(std::memory_order_relaxed)) break;

notifier.prepare_wait(i);

prepared.fetch_add(1, std::memory_order_release);

while (!go.load(std::memory_order_acquire) &&

!stop.load(std::memory_order_relaxed)) {

std::this_thread::yield();

}

if (stop.load(std::memory_order_relaxed)) break;

if (has_work[i].load(std::memory_order_acquire)) {

notifier.cancel_wait(i);

canceled.fetch_add(1, std::memory_order_release);

} else {

notifier.commit_wait(i);

committed.fetch_add(1, std::memory_order_release);

}

local_round++;

}

});

}

std::vector<std::thread> notifiers;

notifiers.reserve(M);

if constexpr (NT == NotificationType::ONE) {

for (size_t t = 0; t < M; ++t) {

notifiers.emplace_back([&]() {

while (!stop.load(std::memory_order_relaxed)) {

notifier.notify_one();

std::this_thread::yield();

}

});

}

} else if constexpr (NT == NotificationType::N) {

for (size_t t = 0; t < M; ++t) {

notifiers.emplace_back([&, t]() {

std::mt19937 trng(seed + static_cast<uint32_t>(t + 1));

std::uniform_int_distribution<size_t> dist(0, N);

while (!stop.load(std::memory_order_relaxed)) {

notifier.notify_n(dist(trng));

std::this_thread::yield();

}

});

}

} else if constexpr (NT == NotificationType::ALL) {

for (size_t t = 0; t < M; ++t) {

notifiers.emplace_back([&, t]() {

std::mt19937 trng(seed + static_cast<uint32_t>(777 + t));

std::uniform_int_distribution<int> burst(1, 8);

while (!stop.load(std::memory_order_relaxed)) {

int b = burst(trng);

for (int i = 0; i < b; ++i) notifier.notify_all();

std::this_thread::yield();

}

});

}

}

std::mt19937 rng(seed);

std::uniform_int_distribution<int> dice(0, 1);

for (size_t r = 0; r < R; ++r) {

prepared.store(0, std::memory_order_relaxed);

canceled.store(0, std::memory_order_relaxed);

committed.store(0, std::memory_order_relaxed);

go.store(false, std::memory_order_release);

for (size_t i = 0; i < N; ++i) has_work[i].store(false, std::memory_order_relaxed);

round.store(r + 1, std::memory_order_release);

while (prepared.load(std::memory_order_acquire) != N) std::this_thread::yield();

for (size_t i = 0; i < N; ++i) {

if (dice(rng)) has_work[i].store(true, std::memory_order_release);

}

go.store(true, std::memory_order_release);

while ((canceled.load(std::memory_order_acquire) +

committed.load(std::memory_order_acquire)) != N) {

std::this_thread::yield();

}

REQUIRE(notifier.num_waiters() == 0);

}

stop.store(true, std::memory_order_release);

go.store(true, std::memory_order_release);

notifier.notify_all();

for (auto& w : workers) w.join();

for (auto& n : notifiers) n.join();

REQUIRE(notifier.num_waiters() == 0);

}

// ============================================================================

// fuzz_stress_notifier

// Condition-variable style predicate (signal_epoch). Workers follow the

// full two-phase wait protocol with randomized delays and partial

// participation. M notifier threads continuously advance the epoch and

// mix all three notify variants. Validates:

// - Every commit_wait returns (no stuck thread).

// - num_waiters() == 0 after shutdown.

// - prepares > 0 (slow path exercised).

// - fast_path > 0 (predicate check before commit works).

// ============================================================================

template <typename T>

void fuzz_stress_notifier(size_t N, size_t M_notifiers, size_t rounds, uint32_t seed) {

T notifier(N);

REQUIRE(notifier.size() == N);

std::atomic<bool> stop{false};

std::atomic<uint64_t> signal_epoch{0};

std::atomic<uint64_t> prepares{0};

std::atomic<uint64_t> cancels{0};

std::atomic<uint64_t> commits_entered{0};

std::atomic<uint64_t> commits_returned{0};

std::atomic<uint64_t> fast_path{0};

std::vector<std::thread> workers;

workers.reserve(N);

for (size_t i = 0; i < N; ++i) {

workers.emplace_back([&, i] {

std::mt19937 rng(seed ^ (0x9e3779b9u + (uint32_t)i * 101u));

uint64_t local = signal_epoch.load(std::memory_order_relaxed);

std::uniform_int_distribution<int> coin(0, 99);

for (size_t it = 0; it < rounds && !stop.load(std::memory_order_relaxed); ++it) {

// Partial participation: sometimes do "work" without waiting.

if (coin(rng) < 15) { tiny_jitter(rng); continue; }

uint64_t cur = signal_epoch.load(std::memory_order_acquire);

if (cur != local) {

local = cur;

fast_path.fetch_add(1, std::memory_order_relaxed);

tiny_jitter(rng);

continue;

}

// Two-phase wait protocol.

tiny_jitter(rng);

notifier.prepare_wait(i);

prepares.fetch_add(1, std::memory_order_relaxed);

tiny_jitter(rng);

cur = signal_epoch.load(std::memory_order_acquire);

if (cur != local) {

notifier.cancel_wait(i);

cancels.fetch_add(1, std::memory_order_relaxed);

local = cur;

tiny_jitter(rng);

continue;

}

commits_entered.fetch_add(1, std::memory_order_relaxed);

notifier.commit_wait(i);

commits_returned.fetch_add(1, std::memory_order_relaxed);

local = signal_epoch.load(std::memory_order_acquire);

tiny_jitter(rng);

}

});

}

std::vector<std::thread> notifiers;

notifiers.reserve(M_notifiers);

for (size_t t = 0; t < M_notifiers; ++t) {

notifiers.emplace_back([&, t] {

std::mt19937 rng(seed + (uint32_t)(777u + t * 17u));

std::uniform_int_distribution<int> which(0, 99);

while (!stop.load(std::memory_order_relaxed)) {

tiny_jitter(rng);

int w = which(rng);

// Occasionally send "empty" notifies (no predicate change) — must be harmless.

if (w < 15) {

int kind = which(rng) % 3;

if (kind == 0) notifier.notify_one();

else if (kind == 1) notifier.notify_n((size_t)(which(rng) % (int)(N + 1)));

else notifier.notify_all();

continue;

}

// Normal: advance predicate then notify (condition-variable style).

signal_epoch.fetch_add(1, std::memory_order_release);

if (w < 60) notifier.notify_one();

else if (w < 85) notifier.notify_n((size_t)(which(rng) % (int)(N + 1)));

else notifier.notify_all();

}

});

}

for (auto& w : workers) w.join();

stop.store(true, std::memory_order_release);

notifier.notify_all();

for (auto& n : notifiers) n.join();

REQUIRE(notifier.num_waiters() == 0);

REQUIRE(commits_returned.load() == commits_entered.load());

REQUIRE(prepares.load() > 0);

REQUIRE(fast_path.load() > 0);

}

// ============================================================================

// notify_n_releases_committed

// Forces ALL N threads to become committed waiters (num_waiters() == N),

// then verifies notify_n(k) alone releases at least min(k, N) of them.

// notify_all() drains the rest so the round completes.

// This is a targeted regression for notify_n boundary semantics and for

// spurious early-exit from commit_wait due to an epoch field bug.

// ============================================================================

template <typename T>

void notify_n_releases_committed(size_t N, size_t k, size_t rounds, uint32_t seed) {

(void)seed;

T notifier(N);

REQUIRE(notifier.size() == N);

std::atomic<size_t> round(0);

std::atomic<size_t> prepared(0);

std::atomic<size_t> committed_done(0);

std::atomic<bool> stop(false);

std::vector<std::thread> workers;

workers.reserve(N);

for (size_t i = 0; i < N; ++i) {

workers.emplace_back([&, i] {

size_t local_round = 0;

while (!stop.load(std::memory_order_relaxed)) {

while (round.load(std::memory_order_acquire) == local_round &&

!stop.load(std::memory_order_relaxed)) {

std::this_thread::yield();

}

if (stop.load(std::memory_order_relaxed)) break;

// No cancel path: we want all threads to become committed waiters.

notifier.prepare_wait(i);

prepared.fetch_add(1, std::memory_order_release);

notifier.commit_wait(i);

committed_done.fetch_add(1, std::memory_order_release);

local_round++;

}

});

}

for (size_t r = 1; r <= rounds; ++r) {

prepared.store(0, std::memory_order_relaxed);

committed_done.store(0, std::memory_order_relaxed);

round.store(r, std::memory_order_release);

while (prepared.load(std::memory_order_acquire) != N) std::this_thread::yield();

// Wait until ALL threads are truly committed waiters.

while (notifier.num_waiters() != N) std::this_thread::yield();

size_t target = (k < N) ? k : N;

notifier.notify_n(k);

auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(5);

while (committed_done.load(std::memory_order_acquire) < target &&

std::chrono::steady_clock::now() < deadline) {

std::this_thread::yield();

}

REQUIRE(committed_done.load(std::memory_order_acquire) >= target);

// Drain remaining waiters.

notifier.notify_all();

auto deadline2 = std::chrono::steady_clock::now() + std::chrono::seconds(10);

while (committed_done.load(std::memory_order_acquire) != N &&

std::chrono::steady_clock::now() < deadline2) {

notifier.notify_all();

std::this_thread::yield();

}

REQUIRE(committed_done.load(std::memory_order_acquire) == N);

REQUIRE(notifier.num_waiters() == 0);

}

stop.store(true, std::memory_order_release);

notifier.notify_all();

for (auto& t : workers) t.join();

REQUIRE(notifier.num_waiters() == 0);

}

// ============================================================================

// stress_test_notifier

// Validates:

// 1. notify_n(k) wakes at least k threads if k <= N.

// 2. notify_all() clears all remaining waiters.

// 3. num_waiters() correctly reflects the epoch-waiter state.

// 4. The epoch mechanism handles wakeups without spurious immediate re-sleeps.

// ============================================================================

template <typename T>

void stress_test_notifier(size_t N, size_t k, size_t rounds) {

T notifier(N);

REQUIRE(notifier.size() == N);

std::atomic<size_t> round(0);

std::atomic<size_t> prepared_count(0);

std::atomic<size_t> wake_count(0);

std::atomic<bool> stop(false);

std::vector<std::thread> workers;

workers.reserve(N);

for (size_t i = 0; i < N; ++i) {

workers.emplace_back([&, i] {

size_t local_round = 0;

while (!stop.load(std::memory_order_relaxed)) {

// Include !stop in the condition so the inner spin exits cleanly

// when the main sets stop=true (not only when round advances).

while (round.load(std::memory_order_acquire) == local_round &&

!stop.load(std::memory_order_relaxed)) {

std::this_thread::yield();

}

if (stop.load(std::memory_order_relaxed)) return;

notifier.prepare_wait(i);

prepared_count.fetch_add(1, std::memory_order_release);

// Two-phase stop check: prepare_wait's seq_cst fence guarantees

// that stop=true is visible here if main stored it before notify_all().

// Without this, notify_all() could fire before prepare_wait and the

// worker would block in commit_wait with no future wakeup.

if (stop.load(std::memory_order_relaxed)) {

notifier.cancel_wait(i);

return;

}

notifier.commit_wait(i);

wake_count.fetch_add(1, std::memory_order_release);

local_round++;

}

});

}

for (size_t r = 1; r <= rounds; ++r) {

prepared_count.store(0);

wake_count.store(0);

round.store(r, std::memory_order_release);

while (prepared_count.load(std::memory_order_acquire) != N) {

std::this_thread::yield();

}

while (notifier.num_waiters() != N) {

std::this_thread::yield();

}

size_t target = std::min(k, N);

notifier.notify_n(k);

auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(2);

while (wake_count.load(std::memory_order_acquire) < target) {

if (std::chrono::steady_clock::now() > deadline) break;

std::this_thread::yield();

}

REQUIRE(wake_count.load(std::memory_order_acquire) >= target);

notifier.notify_all();

auto drain_deadline = std::chrono::steady_clock::now() + std::chrono::seconds(2);

while (wake_count.load(std::memory_order_acquire) < N) {

if (std::chrono::steady_clock::now() > drain_deadline) {

notifier.notify_all();

}

std::this_thread::yield();

}

REQUIRE(wake_count.load(std::memory_order_acquire) == N);

REQUIRE(notifier.num_waiters() == 0);

}

stop.store(true);

round.fetch_add(1);

notifier.notify_all();

for (auto& t : workers) {

if (t.joinable()) t.join();

}

REQUIRE(notifier.num_waiters() == 0);

}

// ============================================================================

// notify_before_commit

// notify_all() fires BEFORE threads call commit_wait(). Threads must NOT

// block (no lost wakeup). Exercises the window between prepare_wait() and

// commit_wait() where the notification arrives.

// ============================================================================

template <typename T>

void notify_before_commit(size_t N) {

T notifier(N);

std::atomic<size_t> prep_count{0};

std::atomic<size_t> committed{0};

std::atomic<bool> notified{false};

std::vector<std::thread> threads;

threads.reserve(N);

for (size_t i = 0; i < N; ++i) {

threads.emplace_back([&, i]() {

notifier.prepare_wait(i);

prep_count.fetch_add(1, std::memory_order_release);

// Spin until the main thread has already called notify_all().

while (!notified.load(std::memory_order_acquire)) std::this_thread::yield();

notifier.commit_wait(i);

committed.fetch_add(1, std::memory_order_release);

});

}

while (prep_count.load(std::memory_order_acquire) != N) std::this_thread::yield();

// Notify before any thread calls commit_wait.

notifier.notify_all();

notified.store(true, std::memory_order_release);

for (auto& t : threads) t.join();

REQUIRE(committed.load() == N);

REQUIRE(notifier.num_waiters() == 0);

}

// ============================================================================

// notify_n_zero_is_noop

// After all N threads are fully committed (num_waiters() == N),

// notify_n(0) must not wake any of them. Immediately after the call

// num_waiters() must still be N. Then notify_all() drains cleanly.

// Catches implementations that treat 0 as "notify all" or misread the

// loop bound.

// ============================================================================

template <typename T>

void notify_n_zero_is_noop(size_t N) {

T notifier(N);

REQUIRE(notifier.size() == N);

std::atomic<size_t> prepared{0};

std::atomic<size_t> committed{0};