future: make cfuture co_await loop-affine too (stacked on #54)

README.md

@@ -1267,10 +1267,10 @@ Composable futures with C++20 coroutine support. Uses `rpp/thread_pool.h` for ba
 |------|-------------|
 | [`cfuture<T>`](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<T>` |
-| [`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<T>` 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<T>` 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<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`) |
+| [`done()`](src/rpp/task.h#L196) | True once resolved; lets a driver poll completion without awaiting (no `wait()`) |
+| [`deferred<T>::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<T>&)`](src/rpp/event_loop.h#L325) or [`run_until_done(deferred<T>&)`](src/rpp/event_loop.h#L354).
 

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

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
             });
         }
 

src/rpp/future_types.h

@@ -46,6 +46,22 @@ namespace rpp
     using coro_handle = RPP_CORO_STD::coroutine_handle<T>;
     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
 
 

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<void*>(uintptr_t{1}); }

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<T>&)/(deferred<T>&) 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<rpp::uint64> 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<int> 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<T>&)/(deferred<T>&) 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<rpp::uint64> body_tid{0};
 
-        auto pool_completing_task = [&]() -> rpp::task<int>
+        auto loop_affine_task = [&]() -> rpp::task<int>
         {
             rpp::cfuture<int> 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<int> 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<int> 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<T> variant: the wrapper's co_await drives the lazy launch (no explicit start()).