From 0bfa8d23a22c97fb464e6498ccbc12988a9751c9 Mon Sep 17 00:00:00 2001 From: Mark Cowan Date: Sun, 21 Jun 2026 04:14:17 +0300 Subject: [PATCH] future/event_loop: make cfuture co_await loop-affine (no pool-thread drift) #54 made task co_await loop-affine (task_final_awaiter posts the continuation back to the owner loop). cfuture coroutines (initial/final_suspend = never, no task_final_awaiter) were left out: cfuture::await_suspend spawns a pool worker that waits on the future and called cont.resume() on that worker, so an event-loop-bound coroutine that co_awaits a cfuture drifts onto a pool thread and then races the loop thread on its own coroutine frame. A chain of cfuture coroutines (each co_awaiting the next) drifts across pool threads with no happens-before -> TSAN data race on the frame. Fix (mirror #54's loop-affine-by-default): - future_types.h: move the loop-context TLS (loop_post_fn / loop_ctx / tl_loop) here from task.h so both the task and the cfuture awaiters can capture it. - future.h: cfuture::await_suspend and cfuture::await_suspend capture tl_loop and, on the pool worker, post the continuation back to the owner loop (falling back to cont.resume() for standalone, no-loop use). - event_loop.cpp: ~event_loop clears tl_loop if it still points at this loop. The constructor advertises the loop on the owner thread's TLS; without the matching clear a later task/cfuture await on that thread would capture a pointer to the destroyed loop (use-after-free) - a pre-existing latent bug the cfuture change above would otherwise widen. Test: tsan_cfuture_await_resumes_on_loop_thread - an event_task co_awaits a pool-completing cfuture directly and asserts the continuation runs on the loop thread (RED before the fix, GREEN after). The existing run_until_done task test is updated: with the loop-affine cfuture await the body now stays on the loop. Co-Authored-By: Claude Opus 4.8 (1M context) --- README.md | 16 +++++++------- src/rpp/event_loop.cpp | 6 ++++++ src/rpp/future.h | 16 ++++++++++---- src/rpp/future_types.h | 16 ++++++++++++++ src/rpp/task.h | 7 ++---- tests/test_event_loop.cpp | 45 ++++++++++++++++++++++++++++++--------- 6 files changed, 79 insertions(+), 27 deletions(-) diff --git a/README.md b/README.md index 09dd0ab8..7bb928e0 100644 --- a/README.md +++ b/README.md @@ -1267,10 +1267,10 @@ Composable futures with C++20 coroutine support. Uses `rpp/thread_pool.h` for ba |------|-------------| | [`cfuture`](src/rpp/future.h#L126) | Extended `std::future` with composition and coroutine support | | [`async_task(task)`](src/rpp/future.h#L32) | Launch a task on the thread pool, returns `cfuture` | -| [`make_ready_future(value)`](src/rpp/future.h#L951) | Create an already-completed future | -| [`make_exceptional_future(e)`](src/rpp/future.h#L968) | Create an already-errored future | -| [`wait_all(futures)`](src/rpp/future.h#L1036) | Block until all futures complete | -| [`get_all(futures)`](src/rpp/future.h#L990) | Block and gather results from all futures | +| [`make_ready_future(value)`](src/rpp/future.h#L959) | Create an already-completed future | +| [`make_exceptional_future(e)`](src/rpp/future.h#L976) | Create an already-errored future | +| [`wait_all(futures)`](src/rpp/future.h#L1044) | Block until all futures complete | +| [`get_all(futures)`](src/rpp/future.h#L998) | Block and gather results from all futures | ### cfuture Methods @@ -1290,8 +1290,8 @@ Composable futures with C++20 coroutine support. Uses `rpp/thread_pool.h` for ba | [`collect_ready(T* result)`](src/rpp/future.h#L433) | If already finished, collects the result into `*result` (non-blocking). Returns `true` if collected | | [`collect_wait(T* result)`](src/rpp/future.h#L451) | If valid, blocks until finished and collects the result into `*result`. Returns `true` if collected | | [`await_suspend(coro_handle<>)`](src/rpp/future.h#L463) | C++20 coroutine suspension point — waits on background thread, then resumes | -| [`await_resume()`](src/rpp/future.h#L475) | C++20 coroutine resume — returns the result, rethrows exceptions | -| [`promise_type`](src/rpp/future.h#L502) | C++20 coroutine promise enabling `rpp::cfuture` as a coroutine return type | +| [`await_resume()`](src/rpp/future.h#L486) | C++20 coroutine resume — returns the result, rethrows exceptions | +| [`promise_type`](src/rpp/future.h#L506) | C++20 coroutine promise enabling `rpp::cfuture` as a coroutine return type | | [`RPP_HAS_COROUTINES`](src/rpp/future_types.h#L12) | Detects whether C++20 coroutine headers are available | | [`RPP_CORO_STD`](src/rpp/future_types.h#L13) | Namespace alias for coroutine types (std or std::experimental) | @@ -1419,8 +1419,8 @@ Neither is a future: there is no `get()`/`wait()`/`.then()` — drive by `co_awa |------|-------------| | [`task`](src/rpp/task.h#L64) | Eager coroutine; body runs at construction. `co_await` resumes the caller on its loop | | [`deferred`](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::start()`](src/rpp/task.h#L265) | Launch a not-yet-awaited deferred (used by `run_until_done`) | +| [`done()`](src/rpp/task.h#L196) | True once resolved; lets a driver poll completion without awaiting (no `wait()`) | +| [`deferred::start()`](src/rpp/task.h#L262) | 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&)`](src/rpp/event_loop.h#L325) or [`run_until_done(deferred&)`](src/rpp/event_loop.h#L354). diff --git a/src/rpp/event_loop.cpp b/src/rpp/event_loop.cpp index d73fb39b..aa61df69 100644 --- a/src/rpp/event_loop.cpp +++ b/src/rpp/event_loop.cpp @@ -53,6 +53,12 @@ namespace rpp // destroy all fork coroutine frames (both completed and stale) fork_tasks.clear(); + + // Stop advertising this loop on the owner thread's TLS. The constructor registered it so + // task/cfuture awaiters can post their continuation back here; once we are gone a later + // await on this thread must not capture a pointer to this destroyed loop (use-after-free). + if (detail::tl_loop.ctx == this) + detail::tl_loop = {}; } void event_loop::stop() noexcept diff --git a/src/rpp/future.h b/src/rpp/future.h index 36d00634..a1ba9fed 100644 --- a/src/rpp/future.h +++ b/src/rpp/future.h @@ -468,11 +468,15 @@ namespace rpp cont.resume(); return; } - rpp::parallel_task_detached([this, cont]() /*clang-12 compat:*/mutable + // Capture the owner loop (if the awaiting coroutine runs on one) so the continuation + // resumes on the loop thread instead of the pool thread that waited on the future. + // Without this, an event-loop-bound coroutine drifts onto a pool thread after co_await. + const rpp::detail::loop_ctx loop = rpp::detail::tl_loop; + rpp::parallel_task_detached([this, cont, loop]() /*clang-12 compat:*/mutable { if (this->valid()) this->wait(); - cont.resume(); // call await_resume() and continue on this background thread + loop.resume(cont); // resume on the owner loop if any, else inline on this worker }); } @@ -878,11 +882,15 @@ namespace rpp cont.resume(); return; } - rpp::parallel_task_detached([this, cont]() /*clang-12 compat:*/mutable + // Capture the owner loop (if the awaiting coroutine runs on one) so the continuation + // resumes on the loop thread instead of the pool thread that waited on the future. + // Without this, an event-loop-bound coroutine drifts onto a pool thread after co_await. + const rpp::detail::loop_ctx loop = rpp::detail::tl_loop; + rpp::parallel_task_detached([this, cont, loop]() /*clang-12 compat:*/mutable { if (this->valid()) this->wait(); - cont.resume(); // call await_resume() and continue on this background thread + loop.resume(cont); // resume on the owner loop if any, else inline on this worker }); } diff --git a/src/rpp/future_types.h b/src/rpp/future_types.h index 85465ab4..6d6d7530 100644 --- a/src/rpp/future_types.h +++ b/src/rpp/future_types.h @@ -46,6 +46,22 @@ namespace rpp using coro_handle = RPP_CORO_STD::coroutine_handle; using suspend_never = RPP_CORO_STD::suspend_never; using suspend_always = RPP_CORO_STD::suspend_always; + + namespace detail + { + // Set by event_loop on its owner thread; captured by the awaiter of a task/cfuture so the + // continuation is posted back to the owner loop instead of resuming on whatever pool thread + // completed the awaited work. Keeps event-loop-bound coroutines loop-affine across co_await. + using loop_post_fn = void (*)(void* ctx, rpp::coro_handle<>) noexcept; + struct loop_ctx + { + loop_post_fn fn = nullptr; + void* ctx = nullptr; + // Resume the continuation on the owner loop if there is one, else inline on this thread. + void resume(rpp::coro_handle<> cont) const noexcept { fn ? fn(ctx, cont) : cont.resume(); } + }; + inline thread_local loop_ctx tl_loop; + } #endif // RPP_HAS_COROUTINES diff --git a/src/rpp/task.h b/src/rpp/task.h index 1fb52738..ac7533b4 100644 --- a/src/rpp/task.h +++ b/src/rpp/task.h @@ -67,11 +67,8 @@ namespace rpp namespace detail { - // Set by event_loop on the owner thread; captured by task_base::await_suspend so - // task_final_awaiter posts the continuation back instead of doing an inline transfer. - using loop_post_fn = void (*)(void* ctx, rpp::coro_handle<>) noexcept; - struct loop_ctx { loop_post_fn fn = nullptr; void* ctx = nullptr; }; - inline thread_local loop_ctx tl_loop; + // loop_post_fn / loop_ctx / tl_loop live in future_types.h so both the task and the cfuture + // awaiters can capture the owner loop and keep event-loop-bound coroutines loop-affine. // Sentinel stored by task_final_awaiter to signal "task completed before await_suspend ran". inline void* task_done_sentinel() noexcept { return reinterpret_cast(uintptr_t{1}); } diff --git a/tests/test_event_loop.cpp b/tests/test_event_loop.cpp index 23d71f76..3119e432 100644 --- a/tests/test_event_loop.cpp +++ b/tests/test_event_loop.cpp @@ -1355,28 +1355,53 @@ TestImpl(test_event_loop) AssertThat(continuation_tid.load(), loop_tid); } - // Follow-up: run_until_done(task&)/(deferred&) must also drive a TOP-LEVEL task to - // completion on the loop thread when the task body completes on a pool thread (it co_awaited a - // cfuture directly). Before the fix the driver polled handle.done() cross-thread (TSAN frame - // race); now it routes completion through an internal event_task so the result is read on loop. + // A direct `co_await someCfuture` must also resume the awaiting coroutine on the owner loop + // thread, not on the pool worker that waited on the future. cfuture::await_suspend spawns a + // pool worker to wait; before the loop-affine fix it called cont.resume() on that worker, so an + // event-loop-bound coroutine drifted onto a pool thread after the await (TSAN frame race vs the + // loop thread, plus a silent thread-affinity violation). Unlike the task case above this never + // went through task_final_awaiter, so #54's task fix did not cover it. + TestCase(tsan_cfuture_await_resumes_on_loop_thread) + { + std::atomic continuation_tid{0}; + const rpp::uint64 loop_tid = rpp::get_thread_id(); + + rpp::event_task et = [&]() -> rpp::event_task + { + // Sleep 1 ms so the future isn't ready before await_suspend captures the loop context. + rpp::cfuture fut = rpp::async_task([] { rpp::sleep_ms(1); return 42; }); + int r = co_await fut; // direct cfuture await: must post the continuation back to the loop + continuation_tid = rpp::get_thread_id(); + AssertThat(r, 42); + }(); + + loop->run_until_done(et); + + AssertThat(continuation_tid.load(), loop_tid); + } + + // Follow-up: run_until_done(task&)/(deferred&) drives a TOP-LEVEL eager task to completion + // and returns its result on the loop thread. The task co_awaits a cfuture directly; with the + // loop-affine cfuture await the body resumes back on the loop (no pool-thread drift), and + // run_until_done routes completion through an internal event_task so the result is read on loop. TestCase(tsan_run_until_done_task_resumes_on_loop_thread) { const rpp::uint64 loop_tid = rpp::get_thread_id(); std::atomic body_tid{0}; - auto pool_completing_task = [&]() -> rpp::task + auto loop_affine_task = [&]() -> rpp::task { rpp::cfuture fut = rpp::async_task([] { rpp::sleep_ms(1); return 42; }); - int r = co_await fut; // resumes this task body on a pool thread + int r = co_await fut; // loop-affine: the body resumes back on the loop thread body_tid = rpp::get_thread_id(); - co_return r; // task_final_awaiter runs on the pool thread + co_return r; // task_final_awaiter runs on the loop thread }; - rpp::task t = pool_completing_task(); // eager: already in flight - int got = loop->run_until_done(t); // drives + reads the result on the loop thread + rpp::task t = loop_affine_task(); // eager: already in flight + int got = loop->run_until_done(t); // drives + reads the result on the loop thread AssertThat(got, 42); - AssertNotEqual(body_tid.load(), loop_tid); // sanity: the body really completed off-loop + AssertThat(body_tid.load(), loop_tid); // body stayed on the loop (cfuture await is loop-affine) } // deferred variant: the wrapper's co_await drives the lazy launch (no explicit start()).