diff --git a/README.md b/README.md index 30a21a9..09dd0ab 100644 --- a/README.md +++ b/README.md @@ -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 @@ -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 @@ -1417,12 +1417,12 @@ Neither is a future: there is no `get()`/`wait()`/`.then()` — drive by `co_awa | Type | Description | |------|-------------| -| [`task`](src/rpp/task.h#L62) | Eager coroutine; body runs at construction. `co_await` resumes the caller on its loop | -| [`deferred`](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::start()`](src/rpp/task.h#L197) | Launch a not-yet-awaited deferred (used by `run_until_done`) | +| [`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`) | -Drive a top-level task to completion with [`event_loop::run_until_done(task&)`](src/rpp/event_loop.h#L319) or [`run_until_done(deferred&)`](src/rpp/event_loop.h#L330). +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). Example: [tests/test_task.cpp](tests/test_task.cpp) @@ -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& task)`](src/rpp/event_loop.h#L319) | Pump the loop until the eager `rpp::task` completes; returns its value (or rethrows) | -| [`pump_until_ready(cfuture&, 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&, 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&, Duration)`](src/rpp/event_loop.h#L730) | Pop from queue, resume on loop thread | -| [`await_pop(concurrent_queue&, Duration)`](src/rpp/event_loop.h#L744) | Pop from queue returning `optional`, resume on loop thread | -| [`post(delegate 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& task)`](src/rpp/event_loop.h#L343) | Pump the loop until the eager `rpp::task` completes; returns its value (or rethrows) | +| [`pump_until_ready(cfuture&, 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&, 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&, Duration)`](src/rpp/event_loop.h#L753) | Pop from queue, resume on loop thread | +| [`await_pop(concurrent_queue&, Duration)`](src/rpp/event_loop.h#L767) | Pop from queue returning `optional`, resume on loop thread | +| [`post(delegate 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 @@ -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 diff --git a/src/rpp/event_loop.cpp b/src/rpp/event_loop.cpp index 7e647b2..d73fb39 100644 --- a/src/rpp/event_loop.cpp +++ b/src/rpp/event_loop.cpp @@ -9,6 +9,11 @@ namespace rpp { + static void loop_post_resume(void* ctx, rpp::coro_handle<> h) noexcept + { + static_cast(ctx)->post_resume(h); + } + event_loop::event_loop(rpp::uint64 main_thr_id, rpp::thread_pool* background_task_pool, rpp::AtomicTimeSource* warpable_clock) noexcept @@ -16,6 +21,7 @@ namespace rpp , 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 @@ -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 @@ -265,7 +274,7 @@ namespace rpp else if (event.callback) { try - { + { event.callback(); } catch (const std::exception& ex) @@ -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 diff --git a/src/rpp/event_loop.h b/src/rpp/event_loop.h index be77a90..783c4ec 100644 --- a/src/rpp/event_loop.h +++ b/src/rpp/event_loop.h @@ -342,19 +342,18 @@ namespace rpp template T run_until_done(rpp::task& task) { - return drive_to_done(task); // eager: already in flight, just pump + return drive_top_level(task); // route through an event_task so completion lands on the loop } /** * @brief Drives the loop until the given (lazy) `rpp::deferred` 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 T run_until_done(rpp::deferred& task) { - task.start(); // lazy: launch the body before pumping - return drive_to_done(task); + return drive_top_level(task); // co_await drives the lazy launch; completion lands on the loop } /** @@ -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/deferred 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 + static event_task await_into(Awaitable& task, std::optional& out, std::exception_ptr& err) + { + try { out.emplace(co_await task); } + catch (...) { err = std::current_exception(); } + } template - 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/deferred 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 + 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) + { + event_task w = await_into_void(task, err); + run_until_done(w); + if (err) std::rethrow_exception(err); + } + else + { + std::optional 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&& generic_task) noexcept diff --git a/src/rpp/task.h b/src/rpp/task.h index e21f110..1fb5273 100644 --- a/src/rpp/task.h +++ b/src/rpp/task.h @@ -13,6 +13,7 @@ #if RPP_HAS_COROUTINES // NOTE: future_types.h already includes and exposes rpp::coro_handle / rpp::suspend_* / RPP_CORO_STD +#include #include // std::exception_ptr #include // std::conditional_t, std::is_void_v #include // std::exchange, std::move @@ -66,17 +67,37 @@ namespace rpp namespace detail { - // Final-suspend awaiter shared by every task/deferred promise: hands control to the awaiting - // coroutine via symmetric transfer (no thread hop, no spawn). Works on any promise that - // exposes a `.continuation` handle. + // 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; + + // 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}); } + struct task_final_awaiter { bool await_ready() const noexcept { return false; } template rpp::coro_handle<> await_suspend(rpp::coro_handle h) const noexcept { - rpp::coro_handle<> cont = h.promise().continuation; - return cont ? cont : RPP_CORO_STD::noop_coroutine(); + auto& p = h.promise(); + // release: result is visible to whoever acquires. acquire on failure: we see the + // captured loop context published before await_suspend's release store/CAS. + void* expected = nullptr; + if (!p.continuation.compare_exchange_strong(expected, task_done_sentinel(), + std::memory_order_release, std::memory_order_acquire)) + { + rpp::coro_handle<> cont = rpp::coro_handle<>::from_address(expected); + if (auto fn = p.resume_loop.fn) + { + fn(p.resume_loop.ctx, cont); + return RPP_CORO_STD::noop_coroutine(); + } + return cont; + } + return RPP_CORO_STD::noop_coroutine(); } void await_resume() const noexcept {} }; @@ -105,7 +126,8 @@ namespace rpp struct task_promise { using value_type = T; - rpp::coro_handle<> continuation{}; // awaiter to resume when we finish + std::atomic continuation{nullptr}; + loop_ctx resume_loop{}; // plain payload, published via the continuation release/acquire handshake std::variant result{}; Owner get_return_object() noexcept { return Owner{rpp::coro_handle::from_promise(*this)}; } @@ -119,7 +141,8 @@ namespace rpp struct task_promise { using value_type = void; - rpp::coro_handle<> continuation{}; + std::atomic continuation{nullptr}; + loop_ctx resume_loop{}; // plain payload, published via the continuation release/acquire handshake std::exception_ptr error{}; Owner get_return_object() noexcept { return Owner{rpp::coro_handle::from_promise(*this)}; } @@ -175,26 +198,38 @@ namespace rpp /// lets drivers like event_loop::run_until_done check completion without awaiting. bool done() const noexcept { return !handle || handle.done(); } - bool await_ready() const noexcept { return !handle || handle.done(); } + // Always false: the CAS in await_suspend must run to coordinate with task_final_awaiter. + bool await_ready() const noexcept { return false; } - // Register the awaiter, then hand off control: - // - Eager: the body already started at construction -> just suspend (noop_coroutine). - // - Deferred, not yet started: launch it via symmetric transfer (return its handle). - // - Deferred, already started (e.g. start() then co_await): just suspend; resuming the - // handle again would double-resume a coroutine already in flight. await_ready() - // short-circuits when it is already done, so this only handles the in-flight case. rpp::coro_handle<> await_suspend(rpp::coro_handle<> awaiting) noexcept { - handle.promise().continuation = awaiting; + auto& p = handle.promise(); + // Capture loop context; the release store/CAS below publishes it, pairing with + // task_final_awaiter's acquire-on-failure. + p.resume_loop = detail::tl_loop; // plain store, sequenced-before the release store/CAS below + if constexpr (Lazy) { if (!started) { started = true; - return handle; + p.continuation.store(awaiting.address(), std::memory_order_release); + return handle; // symmetric transfer launches the body } } - return RPP_CORO_STD::noop_coroutine(); + + void* expected = nullptr; + if (p.continuation.compare_exchange_strong(expected, awaiting.address(), + std::memory_order_release, std::memory_order_acquire)) + return RPP_CORO_STD::noop_coroutine(); + + // Task already done — resume using captured loop context. + if (detail::tl_loop.fn) + { + detail::tl_loop.fn(detail::tl_loop.ctx, awaiting); + return RPP_CORO_STD::noop_coroutine(); + } + return awaiting; } typename Promise::value_type await_resume() { return detail::take_result(handle); } diff --git a/tests/test_event_loop.cpp b/tests/test_event_loop.cpp index 9880746..23d71f7 100644 --- a/tests/test_event_loop.cpp +++ b/tests/test_event_loop.cpp @@ -1313,6 +1313,117 @@ TestImpl(test_event_loop) AssertThat(off_thread, false); // ran on a different thread (one expected error log above) } + // Regression / TSAN repro: rpp::task that completes on a pool thread + // (via a direct co_await cfuture, not via loop->run_async) must NOT cause + // the awaiting event_task to drift to the pool thread. + // + // Before the fix: task_final_awaiter did symmetric transfer to the event_task + // continuation on the completing pool thread; the event_task body then ran off + // the loop thread, racing the loop thread's event_task.done() check in + // run_until_done() — a genuine TSAN data race on the coroutine frame. + // + // After the fix: task_final_awaiter posts the continuation to the owner loop, + // so the event_task body always resumes on the loop thread. + TestCase(tsan_task_await_resumes_on_loop_thread) + { + std::atomic continuation_tid{0}; + const rpp::uint64 loop_tid = rpp::get_thread_id(); + + // A task whose body is resumed (and co_returns) on a pool thread: + // cfuture::await_suspend spawns a pool thread that calls cont.resume() inline, + // so the task body after co_await executes on that pool thread. + auto pool_completing_task = [&]() -> rpp::task + { + // Sleep 1 ms so the future isn't ready before task_base::await_suspend + // writes continuation/resume_loop, keeping the test race-free under TSAN. + rpp::cfuture fut = rpp::async_task([] { rpp::sleep_ms(1); return 42; }); + int r = co_await fut; // cfuture resumes this task body on pool thread + co_return r; // co_return fires on pool thread -> task_final_awaiter on pool thread + }; + + rpp::event_task et = [&]() -> rpp::event_task + { + int r = co_await pool_completing_task(); + // Before fix: event_task body runs on pool thread -> races loop thread's done() check. + // After fix: always resumes on the loop thread. + 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&) 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. + 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 + { + 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 + body_tid = rpp::get_thread_id(); + co_return r; // task_final_awaiter runs on the pool 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 + + AssertThat(got, 42); + AssertNotEqual(body_tid.load(), loop_tid); // sanity: the body really completed off-loop + } + + // deferred variant: the wrapper's co_await drives the lazy launch (no explicit start()). + TestCase(tsan_run_until_done_deferred_resumes_on_loop_thread) + { + auto make = [&]() -> rpp::deferred + { + rpp::cfuture fut = rpp::async_task([] { rpp::sleep_ms(1); return 7; }); + int r = co_await fut; + co_return r; + }; + + rpp::deferred d = make(); // lazy: not started until awaited + int got = loop->run_until_done(d); + AssertThat(got, 7); + } + + // void-result variant: exercises the if-constexpr void branch of the wrapper driver. + TestCase(tsan_run_until_done_task_void) + { + std::atomic ran{false}; + auto make = [&]() -> rpp::task + { + co_await rpp::async_task([] { rpp::sleep_ms(1); }); + ran = true; // completes on a pool thread + co_return; + }; + + rpp::task t = make(); + loop->run_until_done(t); + AssertThat(ran.load(), true); + } + + // An exception thrown after an off-loop completion must propagate out of run_until_done + // (covers the wrapper's exception_ptr round-trip). + TestCase(run_until_done_task_propagates_exception) + { + auto make = [&]() -> rpp::task + { + co_await rpp::async_task([] { rpp::sleep_ms(1); }); + throw std::runtime_error("boom"); // thrown on a pool thread + }; + + rpp::task t = make(); + AssertThrows(loop->run_until_done(t), std::runtime_error); + } + // NOLINTEND(cppcoreguidelines-avoid-capturing-lambda-coroutines) #endif // RPP_HAS_COROUTINES