Skip to content

Fix cross-thread data races where event-loop coroutines complete off the loop thread #54

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

Open
battlesnake wants to merge 6 commits into
base: master
Choose a base branch
from
Open

Fix cross-thread data races where event-loop coroutines complete off the loop thread #54

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
206c3fb
task/event_loop: fix TSAN race where event_task drifts to pool thread…
battlesnake Jun 20, 2026
198fcb9
task: fix memory ordering in task_final_awaiter CAS failure path
battlesnake Jun 20, 2026
41bbd39
task/event_loop: simplify CAS handshake after review
battlesnake Jun 20, 2026
be3dc49
task/event_loop: demote resume-loop payload to a plain member
battlesnake Jun 20, 2026
76282cf
docs: refresh README line refs after task/event_loop changes
battlesnake Jun 20, 2026
0de5f37
event_loop: fix run_until_done(task/deferred) completing off the loop…
battlesnake Jun 20, 2026
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
126 changes: 63 additions & 63 deletions README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -663,17 +663,17 @@ Fast string building and type-safe formatting. `string_buffer` is an always-null

| Method | Description |
|--------|-------------|
| [`write(const T& v)`](src/rpp/sprint.h#L124) | Write a value (auto-converts most types) |
| [`writeln(const Args&... args)`](src/rpp/sprint.h#L339) | Write values followed by newline |
| [`writef(const char* format, ...)`](src/rpp/sprint.h#L122) | Printf-style formatted write |
| [`write_hex(const void* data, int numBytes)`](src/rpp/sprint.h#L285) | Write data as hex string |
| [`write_cont(const Container& c)`](src/rpp/sprint.h#L264) | Write container contents |
| [`prettyprint(const T& value)`](src/rpp/sprint.h#L347) | Pretty-print a value |
| [`clear()`](src/rpp/sprint.h#L113) | Clear the buffer |
| [`reserve(int capacity)`](src/rpp/sprint.h#L114) | Reserve capacity |
| [`resize(int size)`](src/rpp/sprint.h#L115) | Resize buffer |
| [`append(const char* data, int len)`](src/rpp/sprint.h#L118) | Append raw data |
| [`emplace_buffer(int n)`](src/rpp/sprint.h#L121) | Get writable buffer of N bytes |
| [`write(const T& v)`](src/rpp/sprint.h#L125) | Write a value (auto-converts most types) |
| [`writeln(const Args&... args)`](src/rpp/sprint.h#L340) | Write values followed by newline |
| [`writef(const char* format, ...)`](src/rpp/sprint.h#L123) | Printf-style formatted write |
| [`write_hex(const void* data, int numBytes)`](src/rpp/sprint.h#L286) | Write data as hex string |
| [`write_cont(const Container& c)`](src/rpp/sprint.h#L265) | Write container contents |
| [`prettyprint(const T& value)`](src/rpp/sprint.h#L348) | Pretty-print a value |
| [`clear()`](src/rpp/sprint.h#L114) | Clear the buffer |
| [`reserve(int capacity)`](src/rpp/sprint.h#L115) | Reserve capacity |
| [`resize(int size)`](src/rpp/sprint.h#L116) | Resize buffer |
| [`append(const char* data, int len)`](src/rpp/sprint.h#L119) | Append raw data |
| [`emplace_buffer(int n)`](src/rpp/sprint.h#L122) | Get writable buffer of N bytes |

### Free Functions

Expand All @@ -684,9 +684,9 @@ Fast string building and type-safe formatting. `string_buffer` is an always-null
| [`to_string(float)`](src/rpp/sprint.h#L51) | Locale-agnostic float to string |
| [`to_string(double)`](src/rpp/sprint.h#L52) | Locale-agnostic double to string |
| [`to_string(bool)`](src/rpp/sprint.h#L55) | Bool to `"true"` or `"false"` |
| [`print(args...)`](src/rpp/sprint.h#L461) | Print to stdout |
| [`println(args...)`](src/rpp/sprint.h#L481) | Print to stdout with newline |
| [`to_hex_string(s, opt)`](src/rpp/sprint.h#L407) | Converts string bytes to hexadecimal representation |
| [`print(args...)`](src/rpp/sprint.h#L462) | Print to stdout |
| [`println(args...)`](src/rpp/sprint.h#L482) | Print to stdout with newline |
| [`to_hex_string(s, opt)`](src/rpp/sprint.h#L408) | Converts string bytes to hexadecimal representation |

### Example: Basic String Building

Expand Down Expand Up @@ -1417,12 +1417,12 @@ Neither is a future: there is no `get()`/`wait()`/`.then()` — drive by `co_awa

| Type | Description |
|------|-------------|
| [`task<T>`](src/rpp/task.h#L62) | Eager coroutine; body runs at construction. `co_await` resumes the caller on its loop |
| [`deferred<T>`](src/rpp/task.h#L63) | Lazy coroutine; body runs only when first awaited / `start()`ed. Same API as `task` |
| [`done()`](src/rpp/task.h#L154) | True once resolved; lets a driver poll completion without awaiting (no `wait()`) |
| [`deferred<T>::start()`](src/rpp/task.h#L197) | Launch a not-yet-awaited deferred (used by `run_until_done`) |
| [`task<T>`](src/rpp/task.h#L64) | Eager coroutine; body runs at construction. `co_await` resumes the caller on its loop |
| [`deferred<T>`](src/rpp/task.h#L66) | Lazy coroutine; body runs only when first awaited / `start()`ed. Same API as `task` |
| [`done()`](src/rpp/task.h#L199) | True once resolved; lets a driver poll completion without awaiting (no `wait()`) |
| [`deferred<T>::start()`](src/rpp/task.h#L265) | Launch a not-yet-awaited deferred (used by `run_until_done`) |

Drive a top-level task to completion with [`event_loop::run_until_done(task<T>&)`](src/rpp/event_loop.h#L319) or [`run_until_done(deferred<T>&)`](src/rpp/event_loop.h#L330).
Drive a top-level task to completion with [`event_loop::run_until_done(task<T>&)`](src/rpp/event_loop.h#L325) or [`run_until_done(deferred<T>&)`](src/rpp/event_loop.h#L354).

Example: [tests/test_task.cpp](tests/test_task.cpp)

Expand Down Expand Up @@ -1473,34 +1473,34 @@ Single-threaded event loop that serializes coroutine completions. Unlike `thread

| Method | Description |
|--------|-------------|
| [`run_loop()`](src/rpp/event_loop.h#L264) | Run the loop until `stop()` is called, then drain remaining work |
| [`run_once(Duration timeout)`](src/rpp/event_loop.h#L273) | Process at most one pending resume event; `Duration::zero()` for non-blocking poll |
| [`run_until_idle()`](src/rpp/event_loop.h#L289) | Run until no background tasks and no pending resume events remain |
| [`run_until_done(event_task& task)`](src/rpp/event_loop.h#L301) | Drive the loop until the given `event_task` completes, then rethrow on failure |
| [`run_until_done(task<T>& task)`](src/rpp/event_loop.h#L319) | Pump the loop until the eager `rpp::task<T>` completes; returns its value (or rethrows) |
| [`pump_until_ready(cfuture<T>&, timeout)`](src/rpp/event_loop.h#L349) | Pump on the owner thread until that one future is ready; `bool`, never blocks past timeout |
| [`run_until_ready(cfuture<T>&, timeout)`](src/rpp/event_loop.h#L367) | Pump until that future is ready, then return its value; throws on timeout |
| [`ensure_on_owner_thread(source_location)`](src/rpp/event_loop.h#L380) | Debug check: true if on the loop's owner thread, else logs an error at the call site |
| [`run_async(Func&& fut_or_cb)`](src/rpp/event_loop.h#L685) | Dispatch future or lambda to thread pool, resume coroutine on the loop thread |
| [`fork(Func&& coro_factory)`](src/rpp/event_loop.h#L409) | Fork a concurrent coroutine path (fire-and-forget, tracked internally) |
| [`join_forks(Duration timeout)`](src/rpp/event_loop.h#L884) | Event-driven join: suspend until all forks complete or timeout expires |
| [`num_forks()`](src/rpp/event_loop.h#L453) | Number of active forked coroutines |
| [`drain_forks()`](src/rpp/event_loop.h#L461) | Check completed forks for exceptions and clear them |
| [`await(semaphore&, Duration)`](src/rpp/event_loop.h#L716) | Wait for semaphore signal, resume on loop thread |
| [`await(concurrent_queue<T>&, T&, Duration)`](src/rpp/event_loop.h#L730) | Pop from queue, resume on loop thread |
| [`await_pop(concurrent_queue<T>&, Duration)`](src/rpp/event_loop.h#L744) | Pop from queue returning `optional<T>`, resume on loop thread |
| [`post(delegate<void()> callback)`](src/rpp/event_loop.h#L510) | Post a callback to execute on the loop thread (like `run_on_main_thread`) |
| [`post_resume(coro_handle<> handle)`](src/rpp/event_loop.h#L502) | Post a raw coroutine handle resume to the loop thread |
| [`resume_on_loop()`](src/rpp/event_loop.h#L899) | `co_await` to unconditionally reschedule the current coroutine onto the loop thread |
| [`delay(Duration duration)`](src/rpp/event_loop.h#L793) | Sleep on a background thread, resume on the loop thread |
| [`delay_until(TimePoint until)`](src/rpp/event_loop.h#L797) | Sleep until a time point, resume on the loop thread |
| [`stop()`](src/rpp/event_loop.h#L234) | Signal the loop to stop and finalize pending tasks |
| [`wait_on_all(Duration timeout)`](src/rpp/event_loop.h#L241) | Block until all pending work drains, with timeout |
| [`set_except_handler(handler)`](src/rpp/event_loop.h#L247) | Set custom exception handler for unhandled background errors |
| [`has_pending_work()`](src/rpp/event_loop.h#L226) | True if any background tasks or resume events are pending |
| [`background_tasks()`](src/rpp/event_loop.h#L217) | Number of tasks currently suspended in background work |
| [`pending_completions()`](src/rpp/event_loop.h#L223) | Number of pending resume events queued for the loop thread |
| [`main_thread_id()`](src/rpp/event_loop.h#L229) | Thread ID of the loop's owner thread |
| [`run_loop()`](src/rpp/event_loop.h#L287) | Run the loop until `stop()` is called, then drain remaining work |
| [`run_once(Duration timeout)`](src/rpp/event_loop.h#L296) | Process at most one pending resume event; `Duration::zero()` for non-blocking poll |
| [`run_until_idle()`](src/rpp/event_loop.h#L313) | Run until no background tasks and no pending resume events remain |
| [`run_until_done(event_task& task)`](src/rpp/event_loop.h#L325) | Drive the loop until the given `event_task` completes, then rethrow on failure |
| [`run_until_done(task<T>& task)`](src/rpp/event_loop.h#L343) | Pump the loop until the eager `rpp::task<T>` completes; returns its value (or rethrows) |
| [`pump_until_ready(cfuture<T>&, timeout)`](src/rpp/event_loop.h#L372) | Pump on the owner thread until that one future is ready; `bool`, never blocks past timeout |
| [`run_until_ready(cfuture<T>&, timeout)`](src/rpp/event_loop.h#L390) | Pump until that future is ready, then return its value; throws on timeout |
| [`ensure_on_owner_thread(source_location)`](src/rpp/event_loop.h#L403) | Debug check: true if on the loop's owner thread, else logs an error at the call site |
| [`run_async(Func&& fut_or_cb)`](src/rpp/event_loop.h#L708) | Dispatch future or lambda to thread pool, resume coroutine on the loop thread |
| [`fork(Func&& coro_factory)`](src/rpp/event_loop.h#L432) | Fork a concurrent coroutine path (fire-and-forget, tracked internally) |
| [`join_forks(Duration timeout)`](src/rpp/event_loop.h#L907) | Event-driven join: suspend until all forks complete or timeout expires |
| [`num_forks()`](src/rpp/event_loop.h#L476) | Number of active forked coroutines |
| [`drain_forks()`](src/rpp/event_loop.h#L484) | Check completed forks for exceptions and clear them |
| [`await(semaphore&, Duration)`](src/rpp/event_loop.h#L739) | Wait for semaphore signal, resume on loop thread |
| [`await(concurrent_queue<T>&, T&, Duration)`](src/rpp/event_loop.h#L753) | Pop from queue, resume on loop thread |
| [`await_pop(concurrent_queue<T>&, Duration)`](src/rpp/event_loop.h#L767) | Pop from queue returning `optional<T>`, resume on loop thread |
| [`post(delegate<void()> callback)`](src/rpp/event_loop.h#L533) | Post a callback to execute on the loop thread (like `run_on_main_thread`) |
| [`post_resume(coro_handle<> handle)`](src/rpp/event_loop.h#L525) | Post a raw coroutine handle resume to the loop thread |
| [`resume_on_loop()`](src/rpp/event_loop.h#L922) | `co_await` to unconditionally reschedule the current coroutine onto the loop thread |
| [`delay(Duration duration)`](src/rpp/event_loop.h#L816) | Sleep on a background thread, resume on the loop thread |
| [`delay_until(TimePoint until)`](src/rpp/event_loop.h#L820) | Sleep until a time point, resume on the loop thread |
| [`stop()`](src/rpp/event_loop.h#L239) | Signal the loop to stop and finalize pending tasks |
| [`wait_on_all(Duration timeout)`](src/rpp/event_loop.h#L246) | Block until all pending work drains, with timeout |
| [`set_except_handler(handler)`](src/rpp/event_loop.h#L252) | Set custom exception handler for unhandled background errors |
| [`has_pending_work()`](src/rpp/event_loop.h#L231) | True if any background tasks or resume events are pending |
| [`background_tasks()`](src/rpp/event_loop.h#L222) | Number of tasks currently suspended in background work |
| [`pending_completions()`](src/rpp/event_loop.h#L228) | Number of pending resume events queued for the loop thread |
| [`main_thread_id()`](src/rpp/event_loop.h#L234) | Thread ID of the loop's owner thread |

### event_task Methods

Expand Down Expand Up @@ -4343,32 +4343,32 @@ Minimal unit testing framework with test discovery, assertions, and verbose outp

| Item | Description |
|------|-------------|
| [`test`](src/rpp/tests.h#L39) | Base test class with lifecycle hooks |
| [`test_info`](src/rpp/tests.h#L40) | Test registration metadata |
| [`TestVerbosity`](src/rpp/tests.h#L67) | `None`, `Summary`, `TestLabels`, `AllMessages` |
| [`test`](src/rpp/tests.h#L29) | Base test class with lifecycle hooks |
| [`test_info`](src/rpp/tests.h#L35) | Test registration metadata |
| [`TestVerbosity`](src/rpp/tests.h#L47) | `None`, `Summary`, `TestLabels`, `AllMessages` |

### Key Macros

| Macro | Description |
|-------|-------------|
| [`TestImpl(ClassName)`](src/rpp/tests.h#L713) | Register a test class |
| [`TestInit(...)`](src/rpp/tests.h#L728) | Test initialization method |
| [`TestCase(name)`](src/rpp/tests.h#L745) | Define a test case |
| [`AssertThat(expr, expected)`](src/rpp/tests.h#L585) | Assert equality |
| [`AssertEqual(a, b)`](src/rpp/tests.h#L601) | Assert exact equality |
| [`AssertNotEqual(a, b)`](src/rpp/tests.h#L637) | Assert inequality |
| [`AssertTrue(expr)`](src/rpp/tests.h#L706) | Assert expression is true |
| [`AssertFalse(expr)`](src/rpp/tests.h#L576) | Assert expression is false |
| [`AssertThrows(expr)`](src/rpp/tests.h#L610) | Assert expression throws |
| [`TestImpl(ClassName)`](src/rpp/tests.h#L693) | Register a test class |
| [`TestInit(...)`](src/rpp/tests.h#L708) | Test initialization method |
| [`TestCase(name)`](src/rpp/tests.h#L725) | Define a test case |
| [`AssertThat(expr, expected)`](src/rpp/tests.h#L565) | Assert equality |
| [`AssertEqual(a, b)`](src/rpp/tests.h#L581) | Assert exact equality |
| [`AssertNotEqual(a, b)`](src/rpp/tests.h#L617) | Assert inequality |
| [`AssertTrue(expr)`](src/rpp/tests.h#L686) | Assert expression is true |
| [`AssertFalse(expr)`](src/rpp/tests.h#L556) | Assert expression is false |
| [`AssertThrows(expr)`](src/rpp/tests.h#L590) | Assert expression throws |

### Running Tests

| Method | Description |
|--------|-------------|
| [`test::run_tests(patterns)`](src/rpp/tests.h#L283) | Run tests matching patterns |
| [`test::run_tests(argc, argv)`](src/rpp/tests.h#L294) | Run tests from command line args |
| [`test::run_tests()`](src/rpp/tests.h#L283) | Run all registered tests |
| [`register_test(name, factory, autorun)`](src/rpp/tests.h#L65) | Registers a unit test with given name, factory and autorun flag |
| [`test::run_tests(patterns)`](src/rpp/tests.h#L279) | Run tests matching patterns |
| [`test::run_tests(argc, argv)`](src/rpp/tests.h#L274) | Run tests from command line args |
| [`test::run_tests()`](src/rpp/tests.h#L279) | Run all registered tests |
| [`register_test(name, factory, autorun)`](src/rpp/tests.h#L45) | Registers a unit test with given name, factory and autorun flag |

### Example: Defining a Test Class with TestCase

Expand Down
13 changes: 12 additions & 1 deletion src/rpp/event_loop.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,13 +9,19 @@

namespace rpp
{
static void loop_post_resume(void* ctx, rpp::coro_handle<> h) noexcept
{
static_cast<event_loop*>(ctx)->post_resume(h);
}

event_loop::event_loop(rpp::uint64 main_thr_id,
rpp::thread_pool* background_task_pool,
rpp::AtomicTimeSource* warpable_clock) noexcept
: owner_thread_id{main_thr_id ? main_thr_id : rpp::get_thread_id()}
, background_pool{background_task_pool ? *background_task_pool : rpp::thread_pool::global()}
, time_source{warpable_clock}
{
detail::tl_loop = { &loop_post_resume, this };
}

event_loop::~event_loop() noexcept
Expand Down Expand Up @@ -244,6 +250,9 @@ namespace rpp

void event_loop::process_event(resume_event& event) noexcept
{
auto prev = detail::tl_loop;
detail::tl_loop = { &loop_post_resume, this };

if (event.handle)
{
try
Expand All @@ -265,7 +274,7 @@ namespace rpp
else if (event.callback)
{
try
{
{
event.callback();
}
catch (const std::exception& ex)
Expand All @@ -280,6 +289,8 @@ namespace rpp
else LogError("event_loop unhandled exception from delegate");
}
}

detail::tl_loop = prev;
}

void event_loop::invoke_loop_hook() noexcept
Expand Down
53 changes: 41 additions & 12 deletions src/rpp/event_loop.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -342,19 +342,18 @@ namespace rpp
template <typename T>
T run_until_done(rpp::task<T>& task)
{
return drive_to_done(task); // eager: already in flight, just pump
return drive_top_level<T>(task); // route through an event_task so completion lands on the loop
}

/**
* @brief Drives the loop until the given (lazy) `rpp::deferred<T>` completes, then returns
* its value (or rethrows). Unlike `task`, a deferred has not started yet — this
* starts it, then pumps the loop to completion. @warning Owner-thread only.
* its value (or rethrows). Unlike `task`, a deferred has not started yet — awaiting
* it launches the body, then the loop is pumped to completion. @warning Owner-thread only.
*/
template <typename T>
T run_until_done(rpp::deferred<T>& task)
{
task.start(); // lazy: launch the body before pumping
return drive_to_done(task);
return drive_top_level<T>(task); // co_await drives the lazy launch; completion lands on the loop
}

/**
Expand Down Expand Up @@ -924,15 +923,45 @@ namespace rpp

private:

// Shared pump used by both run_until_done(task)/(deferred): drive the loop until the
// already-started task completes, drain the trailing resumes, then return its result.
// Internal event_task wrappers that co_await a top-level task<T>/deferred<T> so its
// task_final_awaiter posts the continuation back to this loop (task.h): the result is then
// produced — and the user task's frame destroyed — on the loop thread, even when the task
// body completed on a pool thread (e.g. it co_awaited a cfuture directly). Captures are
// reference PARAMETERS, never a capturing lambda, so the coroutine frame can't dangle.
template <class Awaitable, class T>
static event_task await_into(Awaitable& task, std::optional<T>& out, std::exception_ptr& err)
{
try { out.emplace(co_await task); }
catch (...) { err = std::current_exception(); }
}
template <class Awaitable>
auto drive_to_done(Awaitable& task) -> decltype(task.await_resume())
static event_task await_into_void(Awaitable& task, std::exception_ptr& err)
{
try { co_await task; }
catch (...) { err = std::current_exception(); }
}

// Drive a top-level task<T>/deferred<T> to completion through the (already loop-affine)
// run_until_done(event_task&) driver, then hand back its value or rethrow. Replaces the old
// cross-thread done() poll, which raced a pool thread completing the task frame off-loop.
template <typename T, class Awaitable>
T drive_top_level(Awaitable& task)
{
while (!task.done())
run_once(rpp::millis(5));
run_all_ready(); // drain callbacks posted during the final resume (e.g. a trailing post())
return task.await_resume(); // done: returns the value or rethrows (non-blocking)
std::exception_ptr err;
if constexpr (std::is_void_v<T>)
{
event_task w = await_into_void(task, err);
run_until_done(w);
if (err) std::rethrow_exception(err);
}
else
{
std::optional<T> out;
event_task w = await_into(task, out, err);
run_until_done(w);
if (err) std::rethrow_exception(err);
return std::move(*out);
}
}

void start_in_background(task_delegate<void()>&& generic_task) noexcept
Expand Down
Loading