From 093b565b7c061880bec2b161d4451da929518d1c Mon Sep 17 00:00:00 2001 From: Coldwings Date: Fri, 23 Oct 2026 12:08:12 +0800 Subject: [PATCH 1/5] feat(runtime): add worker lifetime leases for external owners --- CHANGELOG.md | 7 + README.md | 7 +- examples/CMakeLists.txt | 10 + examples/worker_lease.cpp | 41 ++++ include/elio/elio.hpp | 1 + include/elio/runtime/scheduler.hpp | 25 +- include/elio/runtime/worker_lease.hpp | 101 +++++++++ include/elio/runtime/worker_thread.hpp | 23 +- tests/CMakeLists.txt | 8 +- tests/unit/test_worker_lease.cpp | 303 +++++++++++++++++++++++++ wiki/API-Contracts.md | 1 + wiki/API-Reference.md | 20 ++ wiki/Home.md | 1 + wiki/Worker-Leases.md | 73 ++++++ 14 files changed, 613 insertions(+), 8 deletions(-) create mode 100644 examples/worker_lease.cpp create mode 100644 include/elio/runtime/worker_lease.hpp create mode 100644 tests/unit/test_worker_lease.cpp create mode 100644 wiki/Worker-Leases.md diff --git a/CHANGELOG.md b/CHANGELOG.md index 1ddda70f..c6640954 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- Move-only worker lifetime leases for external thread-affine owners. Lease + acquisition retains exact-owner continuations and worker drain accounting; + release safely wakes retirement, including from a foreign thread. Retiring + workers service requeued owner tasks in bounded turns (#1301). + ## [0.6.0] - 2026-10-12 This is the first formal 0.6 release. It includes intentional API and diff --git a/README.md b/README.md index 3b2e87b9..f7ce1e0f 100644 --- a/README.md +++ b/README.md @@ -319,7 +319,7 @@ their corresponding feature targets. - **Coroutine/runtime**: `hello_world.cpp`, `chained_coroutines.cpp`, `exception_handling.cpp`, `join_destroyed_async.cpp`, `join_wait.cpp`, `parallel_tasks.cpp`, `dynamic_threads.cpp`, - `autoscaler_example.cpp`, `thread_affinity.cpp`, `signal_handling.cpp`, + `autoscaler_example.cpp`, `thread_affinity.cpp`, `worker_lease.cpp`, `signal_handling.cpp`, `debug_test.cpp` - **Benchmarks**: `benchmark.cpp`, `quick_benchmark.cpp`, `microbench.cpp`, `scheduler_service_benchmark.cpp`, `io_benchmark.cpp`, @@ -428,6 +428,11 @@ size_t executed = sched.total_tasks_executed(); sched.shutdown(); ``` +External thread-affine owners can acquire a +[worker lifetime lease](wiki/Worker-Leases.md) from an Elio coroutine to retain +their worker through normal pool shrink and drain. Affinity alone does not +provide that lifetime guarantee. + Scheduler execution metrics are published from worker-local counters. Optional autoscaler `on_block` detection samples those counters at `tick_interval`, so ordinary coroutine resumes do not read the clock. Block notifications are diff --git a/examples/CMakeLists.txt b/examples/CMakeLists.txt index 84aa1c6f..111a87d6 100644 --- a/examples/CMakeLists.txt +++ b/examples/CMakeLists.txt @@ -73,6 +73,16 @@ add_executable(thread_affinity thread_affinity.cpp) target_link_libraries(thread_affinity PRIVATE elio) target_link_options(thread_affinity PRIVATE ${STATIC_LINK_FLAGS}) +add_executable(worker_lease worker_lease.cpp) +target_link_libraries(worker_lease PRIVATE elio) +target_link_options(worker_lease PRIVATE ${STATIC_LINK_FLAGS}) +if(ELIO_BUILD_TESTS) + add_test(NAME worker_lease_example COMMAND worker_lease) + set_tests_properties(worker_lease_example PROPERTIES TIMEOUT 30 + PASS_REGULAR_EXPRESSION "Owner cleanup on worker 1") + add_dependencies(test_all worker_lease) +endif() + add_executable(benchmark benchmark.cpp) target_link_libraries(benchmark PRIVATE elio) target_link_options(benchmark PRIVATE ${STATIC_LINK_FLAGS}) diff --git a/examples/worker_lease.cpp b/examples/worker_lease.cpp new file mode 100644 index 00000000..2e270c1f --- /dev/null +++ b/examples/worker_lease.cpp @@ -0,0 +1,41 @@ +#include +#include +#include +#include +#include + +int main() { + elio::runtime::scheduler sched(2); + elio::sync::event retire_owner; + std::promise ready; + auto ready_result = ready.get_future(); + std::exception_ptr failure; + sched.start(); + sched.go_to(1, [&]() -> elio::coro::task { + bool published = false; + try { + auto owner = co_await elio::runtime::lease_current_worker(); + const auto original_thread = std::this_thread::get_id(); + ready.set_value(); + published = true; + // Represents an external owner whose lifetime spans a non-I/O wait. + co_await retire_owner.wait(); + if (std::this_thread::get_id() != original_thread) { + throw std::runtime_error("owner migrated during retirement"); + } + std::cout << "Owner cleanup on worker " << owner.worker_id() << '\n'; + // Destroy external thread-affine resources before releasing. + owner.release(); + } catch (...) { + if (!published) ready.set_exception(std::current_exception()); + else failure = std::current_exception(); + } + }); + // Blocking waits are on the controller thread, never a scheduler worker. + ready_result.get(); + sched.set_thread_count(1); + retire_owner.set(); + const bool drained = sched.shutdown(); + if (failure) std::rethrow_exception(failure); + return drained ? 0 : 1; +} diff --git a/include/elio/elio.hpp b/include/elio/elio.hpp index 946ad582..2357f84c 100644 --- a/include/elio/elio.hpp +++ b/include/elio/elio.hpp @@ -34,6 +34,7 @@ #include "runtime/chase_lev_deque.hpp" #include "runtime/async_main.hpp" #include "runtime/affinity.hpp" +#include "runtime/worker_lease.hpp" #include "runtime/serve.hpp" #include "runtime/spawn.hpp" #include "runtime/blocking_pool.hpp" diff --git a/include/elio/runtime/scheduler.hpp b/include/elio/runtime/scheduler.hpp index 83d6fc86..43404601 100644 --- a/include/elio/runtime/scheduler.hpp +++ b/include/elio/runtime/scheduler.hpp @@ -1770,6 +1770,22 @@ inline void worker_thread::start() { thread_ = std::thread(&worker_thread::run, this); } +inline io::detail::io_operation_guard worker_thread::acquire_external_lease( + std::shared_ptr execution_context) { + if (current() != this || !execution_context) { + throw std::logic_error("worker lease requires the current Elio task"); + } + std::lock_guard lifecycle_lock(scheduler_->shutdown_mutex_); + std::lock_guard admission_lock(schedule_mutex_); + if (scheduler_->shutdown_started_ || !scheduler_->is_running() || + !is_running() || is_draining() || + worker_id_ >= scheduler_->num_threads(std::memory_order_acquire)) { + throw std::runtime_error("worker lease admission is closed"); + } + return io::detail::io_operation_guard( + std::move(execution_context), io_context_->operation_identity()); +} + inline void worker_thread::request_stop() noexcept { { std::lock_guard lock(schedule_mutex_); @@ -1851,8 +1867,10 @@ inline void worker_thread::redistribute_tasks(scheduler* sched) noexcept { // First drain external queues to the local deque. drain_inbox(); void* addr; - // Then redistribute all tasks to active workers - while ((addr = queue_->pop()) != nullptr) { + // Owner-local work can requeue itself. Bound this turn so it cannot keep + // retirement inside redistribution forever and starve I/O/maintenance. + for (size_t serviced = 0; serviced < external_service_quantum_ && + (addr = queue_->pop()) != nullptr; ++serviced) { auto handle = std::coroutine_handle<>::from_address(addr); if (handle && !handle.done()) { if (!run_or_redistribute_retiring_task(sched, handle)) { @@ -2039,7 +2057,8 @@ inline void worker_thread::run() { #ifdef ELIO_RUNTIME_TEST_HOOKS record_blocking_poll_for_test(); #endif - io_context_->poll(std::chrono::milliseconds(50)); + io_context_->poll(std::chrono::milliseconds( + queue_->empty() ? 50 : 0)); continue; } diff --git a/include/elio/runtime/worker_lease.hpp b/include/elio/runtime/worker_lease.hpp new file mode 100644 index 00000000..8e48553d --- /dev/null +++ b/include/elio/runtime/worker_lease.hpp @@ -0,0 +1,101 @@ +#pragma once + +#include "scheduler.hpp" +#include +#include +#include +#include +#include +#include +#include + +namespace elio::runtime { + +class lease_current_worker_awaitable; + +/// Retains a worker and the acquiring logical task's exact-owner routing. +/// Release only after external owner-thread obligations have been settled. +/// Token access must be serialized; release may run on a foreign thread. +class external_worker_lease final { +public: + external_worker_lease() noexcept = default; + external_worker_lease(const external_worker_lease&) = delete; + external_worker_lease& operator=(const external_worker_lease&) = delete; + + external_worker_lease(external_worker_lease&& other) noexcept { + move_from(std::move(other)); + } + + external_worker_lease& operator=(external_worker_lease&& other) noexcept { + if (this != &other) { + release(); + move_from(std::move(other)); + } + return *this; + } + + ~external_worker_lease() { release(); } + + [[nodiscard]] bool active() const noexcept { return guard_.active(); } + [[nodiscard]] size_t worker_id() const noexcept { return worker_id_; } + [[nodiscard]] uint64_t context_generation() const noexcept { + return generation_; + } + + void release() noexcept { + if (!active()) return; + // Serialize the last decrement/wake with destruction of the context. + // The shared wake slot survives forced scheduler teardown. + std::lock_guard lock(wakeup_->mutex); + guard_.release(); + if (wakeup_->context) wakeup_->context->notify(); + } + +private: + friend class lease_current_worker_awaitable; + + void move_from(external_worker_lease&& other) noexcept { + guard_ = std::move(other.guard_); + wakeup_ = std::move(other.wakeup_); + worker_id_ = std::exchange(other.worker_id_, coro::NO_AFFINITY); + generation_ = std::exchange(other.generation_, 0); + } + + io::detail::io_operation_guard guard_; + std::shared_ptr wakeup_; + size_t worker_id_ = coro::NO_AFFINITY; + uint64_t generation_ = 0; +}; + +class lease_current_worker_awaitable final { +public: + [[nodiscard]] bool await_ready() const noexcept { return false; } + + template + requires std::is_base_of_v + bool await_suspend(std::coroutine_handle handle) { + auto* worker = worker_thread::current(); + if (!worker) { + throw std::logic_error("worker lease requires an Elio worker"); + } + lease_.guard_ = worker->acquire_external_lease( + handle.promise().execution_context()); + lease_.wakeup_ = worker->external_wakeup_; + lease_.worker_id_ = worker->worker_id(); + lease_.generation_ = worker->io_context().generation(); + return false; + } + + [[nodiscard]] external_worker_lease await_resume() noexcept { + return std::move(lease_); + } + +private: + external_worker_lease lease_; +}; + +[[nodiscard]] inline auto lease_current_worker() noexcept { + return lease_current_worker_awaitable{}; +} + +} // namespace elio::runtime diff --git a/include/elio/runtime/worker_thread.hpp b/include/elio/runtime/worker_thread.hpp index 4e661883..3152a3e7 100644 --- a/include/elio/runtime/worker_thread.hpp +++ b/include/elio/runtime/worker_thread.hpp @@ -19,6 +19,11 @@ namespace elio::runtime { class scheduler; namespace detail { +struct external_worker_wakeup { + std::mutex mutex; + io::io_context* context = nullptr; +}; + // Allocated with the operation, not with its cancellation request. Only the // owner reads next/keep_alive after publication; attempt returns false when // admission must be retried after a backend poll. @@ -57,6 +62,7 @@ inline std::atomic worker_io_backend_for_test{ /// of rejecting accepted worker-bound resumptions. class worker_thread { friend class scheduler; + friend class lease_current_worker_awaitable; public: worker_thread(scheduler* sched, size_t worker_id, @@ -71,14 +77,18 @@ class worker_thread { , io_context_(io::io_context::make_worker_owned( worker_id, detail::worker_io_backend_for_test.load( - std::memory_order_acquire))) { + std::memory_order_acquire))) #else - , io_context_(io::io_context::make_worker_owned(worker_id)) { + , io_context_(io::io_context::make_worker_owned(worker_id)) #endif + { + external_wakeup_->context = io_context_.get(); } ~worker_thread() { stop(); + std::lock_guard lock(external_wakeup_->mutex); + external_wakeup_->context = nullptr; } worker_thread(const worker_thread&) = delete; @@ -299,7 +309,10 @@ class worker_thread { } void enter_draining_mode() noexcept { - draining_.store(true, std::memory_order_release); + { + std::lock_guard lock(schedule_mutex_); + draining_.store(true, std::memory_order_release); + } wake(); } @@ -431,6 +444,8 @@ class worker_thread { } void run(); + [[nodiscard]] io::detail::io_operation_guard acquire_external_lease( + std::shared_ptr execution_context); void drain_inbox() noexcept; void service_maintenance() noexcept; void abandon_maintenance() noexcept; @@ -505,6 +520,8 @@ class worker_thread { bool needs_sync_ = false; // Whether current task needs memory synchronization wait_strategy strategy_; // Configurable wait strategy std::unique_ptr io_context_; // Per-worker io_context + std::shared_ptr external_wakeup_{ + std::make_shared()}; static inline thread_local worker_thread* current_worker_ = nullptr; }; diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index 243aab97..f0e30a8d 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -54,6 +54,7 @@ set(TEST_SOURCES unit/test_chase_lev_deque.cpp unit/test_scheduler.cpp unit/test_scheduler_shutdown.cpp + unit/test_worker_lease.cpp unit/test_blocking_pool.cpp unit/test_spawn_blocking_uaf.cpp unit/test_affinity.cpp @@ -358,6 +359,11 @@ target_link_libraries(elio_core_sse_writer_compile PRIVATE elio Catch2::Catch2) target_compile_definitions(elio_core_sse_writer_compile PRIVATE ELIO_DEBUG) add_executable(elio_tests ${TEST_SOURCES}) +add_executable(elio_worker_lease_tests unit/test_worker_lease.cpp) +target_link_libraries(elio_worker_lease_tests PRIVATE elio Catch2::Catch2WithMain) +target_compile_definitions(elio_worker_lease_tests PRIVATE ELIO_DEBUG ELIO_RUNTIME_TEST_HOOKS) +add_test(NAME worker_lease COMMAND elio_worker_lease_tests) +set_tests_properties(worker_lease PROPERTIES TIMEOUT 120) target_link_libraries(elio_tests PRIVATE elio Catch2::Catch2WithMain) if(TARGET elio_http) target_link_libraries(elio_tests PRIVATE elio_http) @@ -619,7 +625,7 @@ add_custom_target(test_all COMMAND ${CMAKE_CTEST_COMMAND} --output-on-failure WORKING_DIRECTORY ${CMAKE_BINARY_DIR} ) -add_dependencies(test_all elio_tests ${ELIO_ADDITIONAL_TEST_TARGETS}) +add_dependencies(test_all elio_tests elio_worker_lease_tests ${ELIO_ADDITIONAL_TEST_TARGETS}) # CTest also invokes these executables directly or through script fixtures. # Optional targets must not introduce dependencies when their feature is off. diff --git a/tests/unit/test_worker_lease.cpp b/tests/unit/test_worker_lease.cpp new file mode 100644 index 00000000..2a361580 --- /dev/null +++ b/tests/unit/test_worker_lease.cpp @@ -0,0 +1,303 @@ +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include "../test_main.cpp" + +using namespace elio; +using namespace elio::runtime; +using namespace std::chrono_literals; + +namespace { + +template +bool eventually(Predicate predicate) { + const auto deadline = std::chrono::steady_clock::now() + test::scaled_sec(3); + while (!predicate() && std::chrono::steady_clock::now() < deadline) { + std::this_thread::yield(); + } + return predicate(); +} + +struct backend_selection { + explicit backend_selection(io::io_context::backend_type backend) + : previous(runtime::detail::worker_io_backend_for_test.exchange(backend)) {} + ~backend_selection() { + runtime::detail::worker_io_backend_for_test.store(previous); + } + io::io_context::backend_type previous; +}; + +} // namespace + +static_assert(!std::is_copy_constructible_v); +static_assert(std::is_nothrow_move_constructible_v); +static_assert(std::is_nothrow_move_assignable_v); + +TEST_CASE("worker lease retains an event waiter through shrink", "[worker_lease][scheduler]") { + const auto selected = GENERATE(io::io_context::backend_type::epoll, + io::io_context::backend_type::auto_detect); + backend_selection backend(selected); + scheduler sched(2); + auto* owner = sched.get_worker(1); + sync::event gate; + std::atomic acquired{false}, finished{false}, correct_owner{false}; + std::atomic move_valid{false}; + sched.start(); + sched.go_to(1, [&]() -> coro::task { + auto first = co_await lease_current_worker(); + auto second = co_await lease_current_worker(); + const auto generation = first.context_generation(); + external_worker_lease moved(std::move(first)); + first = std::move(second); + move_valid.store(!second.active() && moved.active() && first.active() && + owner->io_context().active_pin_count() == 2); + acquired.store(true, std::memory_order_release); + co_await gate.wait(); + correct_owner.store(worker_thread::current() == owner && + moved.context_generation() == generation); + moved.release(); + moved.release(); + first.release(); + finished.store(true, std::memory_order_release); + }); + const bool started = eventually([&] { return acquired.load(std::memory_order_acquire); }); + sched.set_thread_count(1); + const bool retained = owner->is_running() && sched.pending_tasks() >= 2 && + owner->io_context().active_pin_count() == 2; + gate.set(); + const bool completed = eventually([&] { return finished.load(std::memory_order_acquire); }); + const bool retired = eventually([&] { return !owner->is_running(); }); + // Reuse must wait for the old OS thread and must not inherit its pins. + sched.set_thread_count(2); + const bool reused_cleanly = owner->is_running() && + owner->io_context().active_pin_count() == 0; + sched.shutdown(); + REQUIRE(started); + REQUIRE(retained); + REQUIRE(completed); + REQUIRE(retired); + REQUIRE(correct_owner.load()); + REQUIRE(move_valid.load()); + REQUIRE(reused_cleanly); +} + +TEST_CASE("worker lease acquisition rejects after shrink admission closes", "[worker_lease][scheduler]") { + backend_selection backend(io::io_context::backend_type::epoll); + scheduler sched(2); + auto* owner = sched.get_worker(1); + std::atomic entered{false}, proceed{false}, finished{false}, rejected{false}; + sched.start(); + sched.go_to(1, [&]() -> coro::task { + entered.store(true, std::memory_order_release); + // A deterministic test barrier, not an application worker wait. + proceed.wait(false, std::memory_order_acquire); + try { + auto lease = co_await lease_current_worker(); + } catch (const std::runtime_error&) { + rejected.store(true); + } + finished.store(true, std::memory_order_release); + }); + const bool started = eventually([&] { return entered.load(std::memory_order_acquire); }); + sched.set_thread_count(1); + const bool draining = owner->is_draining(); + proceed.store(true, std::memory_order_release); + proceed.notify_all(); + const bool completed = eventually([&] { return finished.load(std::memory_order_acquire); }); + sched.shutdown(); + REQUIRE(started); + REQUIRE(draining); + REQUIRE(completed); + REQUIRE(rejected.load()); + REQUIRE(owner->io_context().active_pin_count() == 0); +} + +TEST_CASE("yielding leased owner cannot monopolize retirement service", "[worker_lease][scheduler]") { + backend_selection backend(io::io_context::backend_type::epoll); + scheduler sched(2); + auto* owner = sched.get_worker(1); + std::atomic acquired{false}, proceed{false}, finished{false}; + std::atomic polls{0}; + sched.start(); + sched.go_to(1, [&]() -> coro::task { + auto lease = co_await lease_current_worker(); + acquired.store(true, std::memory_order_release); + proceed.wait(false, std::memory_order_acquire); + const auto before = owner->blocking_poll_calls_for_test(); + for (size_t i = 0; i < 4096; ++i) co_await time::yield(); + polls.store(owner->blocking_poll_calls_for_test() - before); + lease.release(); + finished.store(true, std::memory_order_release); + }); + const bool started = eventually([&] { return acquired.load(std::memory_order_acquire); }); + sched.set_thread_count(1); + proceed.store(true, std::memory_order_release); + proceed.notify_all(); + const bool completed = eventually([&] { return finished.load(std::memory_order_acquire); }); + sched.shutdown(); + REQUIRE(started); + REQUIRE(completed); + REQUIRE(polls.load() >= 15); +} + +TEST_CASE("foreign lease release survives forced scheduler destruction", "[worker_lease][scheduler]") { + backend_selection backend(io::io_context::backend_type::epoll); + std::optional retained; + std::atomic acquired{false}; + { + scheduler sched(1); + sched.start(); + sched.go([&]() -> coro::task { + retained.emplace(co_await lease_current_worker()); + acquired.store(true, std::memory_order_release); + co_return; + }); + const bool started = eventually([&] { return acquired.load(std::memory_order_acquire); }); + if (!started) { + sched.shutdown_force(); + FAIL("lease acquisition did not run"); + } + REQUIRE(sched.pending_tasks() >= 1); + sched.shutdown_force(); + } + REQUIRE(retained->active()); + retained->release(); + retained->release(); + REQUIRE_FALSE(retained->active()); +} + +TEST_CASE("foreign release wakes a draining worker", "[worker_lease][scheduler]") { + backend_selection backend(io::io_context::backend_type::epoll); + scheduler sched(2); + auto* owner = sched.get_worker(1); + std::optional retained; + std::atomic acquired{false}; + sched.start(); + sched.go_to(1, [&]() -> coro::task { + retained.emplace(co_await lease_current_worker()); + acquired.store(true, std::memory_order_release); + co_return; + }); + const bool started = eventually([&] { return acquired.load(std::memory_order_acquire); }); + sched.set_thread_count(1); + const bool retained_owner = owner->is_running(); + if (started) retained->release(); + const bool retired = eventually([&] { return !owner->is_running(); }); + sched.shutdown(); + REQUIRE(started); + REQUIRE(retained_owner); + REQUIRE(retired); +} + +TEST_CASE("graceful shutdown waits for independent lease release", "[worker_lease][scheduler]") { + backend_selection backend(io::io_context::backend_type::epoll); + scheduler sched(1); + std::optional retained; + std::atomic acquired{false}, shutdown_returned{false}; + bool drained = false; + sched.start(); + sched.go([&]() -> coro::task { + retained.emplace(co_await lease_current_worker()); + acquired.store(true, std::memory_order_release); + co_return; + }); + const bool started = eventually([&] { return acquired.load(std::memory_order_acquire); }); + runtime::detail::graceful_admission_closed_for_test.store(false); + std::thread shutdown([&] { + drained = sched.shutdown(test::scaled_sec(3)); + shutdown_returned.store(true, std::memory_order_release); + }); + const bool closed = eventually([] { + return runtime::detail::graceful_admission_closed_for_test.load(std::memory_order_acquire); + }); + const bool waited = !shutdown_returned.load(std::memory_order_acquire); + if (started) retained->release(); + shutdown.join(); + REQUIRE(started); + REQUIRE(closed); + REQUIRE(waited); + REQUIRE(drained); +} + +namespace { +coro::task acquire_without_worker(bool* rejected) { + try { + auto lease = co_await lease_current_worker(); + } catch (const std::logic_error&) { + *rejected = true; + } +} + +coro::task throwing_lease_child() { + auto lease = co_await lease_current_worker(); + throw std::runtime_error("owner setup failed"); +} +} // namespace + +TEST_CASE("worker lease rejects off-worker acquisition", "[worker_lease][scheduler]") { + bool rejected = false; + auto value = acquire_without_worker(&rejected); + coro::detail::task_access::handle(value).resume(); + REQUIRE(rejected); +} + +TEST_CASE("nested lease setup failure unwinds its pin", "[worker_lease][scheduler]") { + backend_selection backend(io::io_context::backend_type::epoll); + scheduler sched(1); + std::atomic finished{false}, released{false}; + sched.start(); + sched.go([&]() -> coro::task { + try { + co_await throwing_lease_child(); + } catch (const std::runtime_error&) { + released.store(worker_thread::current()->io_context().active_pin_count() == 0); + } + finished.store(true, std::memory_order_release); + }); + const bool completed = eventually([&] { return finished.load(std::memory_order_acquire); }); + sched.shutdown(); + REQUIRE(completed); + REQUIRE(released.load()); +} + +TEST_CASE("stale lease release cannot affect another scheduler generation", "[worker_lease][scheduler]") { + backend_selection backend(io::io_context::backend_type::epoll); + std::optional stale, current; + std::atomic acquired{false}; + { + scheduler old(1); + old.start(); + old.go([&]() -> coro::task { + stale.emplace(co_await lease_current_worker()); + acquired.store(true, std::memory_order_release); + co_return; + }); + const bool started = eventually([&] { return acquired.load(std::memory_order_acquire); }); + old.shutdown_force(); + REQUIRE(started); + } + scheduler replacement(1); + acquired.store(false); + replacement.start(); + replacement.go([&]() -> coro::task { + current.emplace(co_await lease_current_worker()); + acquired.store(true, std::memory_order_release); + co_return; + }); + const bool started = eventually([&] { return acquired.load(std::memory_order_acquire); }); + stale->release(); + const bool unchanged = replacement.get_worker(0)->io_context().active_pin_count() == 1; + const bool distinct = started && stale->context_generation() != current->context_generation(); + if (started) current->release(); + replacement.shutdown(); + REQUIRE(started); + REQUIRE(unchanged); + REQUIRE(distinct); +} diff --git a/wiki/API-Contracts.md b/wiki/API-Contracts.md index 0dd449a4..ac750a24 100644 --- a/wiki/API-Contracts.md +++ b/wiki/API-Contracts.md @@ -82,6 +82,7 @@ by a broad module heading without checking the feature page or header comment. | `spawn_blocking()` and blocking pool | Runs blocking or CPU-heavy callables outside scheduler worker execution and resumes the awaiting coroutine on its scheduler. Without scheduler context it uses a standalone detached thread. If a scheduler exists but its blocking pool is unavailable, it stays on the current worker and throws `std::runtime_error`; it does not migrate to a detached thread. | Use it for work that would block or starve scheduler workers. Keep callable captures valid by value/reference rules. Keep the scheduler alive through completion and handle pool-rejection exceptions where the pool can be shut down independently. | | `runtime::current_worker_id()` | Reports the current worker ID when running on a scheduler worker. | Handle the no-current-worker case according to the return contract. | | `runtime::set_affinity()` | Requests migration or pinning to the target worker according to scheduler affinity rules. | Use valid worker IDs for deterministic placement and account for resizing. | +| `runtime::lease_current_worker()`, `external_worker_lease` | Atomically admits a move-only external-owner obligation on the current live worker or rejects closed admission. While active, retains that worker through normal retirement, routes the acquiring logical task to its exact owner, and contributes to pending accounting. Idempotent release removes the constraint and wakes retirement; serialized foreign-thread release and destruction after forced scheduler teardown are safe for token metadata. See [[Worker Leases]]. | Drive and drain external work yourself in bounded nonblocking turns; destroy thread-affine resources before release. Moving a token does not transfer placement to another logical task. Serialize access to each token. Do not treat a lease as an external event-loop driver, a cancellation mechanism, or protection against forced/timed-out shutdown. | | `runtime::clear_affinity()` | Clears caller-requested affinity. It does not clear an active operation-local I/O pin. | Use it only when the coroutine no longer needs caller-selected placement. Do not treat it as cancellation or migration of pending I/O. | | `runtime::bind_to_current_worker()` | Binds the current coroutine to its current worker when one exists. | Call it only after entering scheduler execution and before relying on worker-local resources. | | Process `fork()` boundary | Before any Elio runtime or I/O resource is used, a single-threaded process may fork and let parent and child create independent fresh runtimes. If a runtime or any other thread is active, the parent may continue while the child immediately calls an explicitly async-signal-safe exec function such as `execve()`, or `_exit()`. See [[Fork Safety]]. | Do not schedule, resume, cancel, poll, shut down, destroy, or replace inherited Elio runtime state in the child. Do not unwind inherited Elio owners or call `std::exit()`. Do not assume every exec-family wrapper is async-signal-safe. Elio installs no `pthread_atfork` repair hooks; satisfy the fork contracts of the C/C++ runtime and all other linked libraries as well. | diff --git a/wiki/API-Reference.md b/wiki/API-Reference.md index 62779a37..79cfcd87 100644 --- a/wiki/API-Reference.md +++ b/wiki/API-Reference.md @@ -184,6 +184,26 @@ coro::task main_task() { > **Note:** `go_to()` differs from spawning with `go()` + `co_await set_affinity()`. With `go_to()`, worker affinity is set before the task is scheduled and before it ever runs. A later steal attempt may briefly remove the task from a queue, but the scheduler checks affinity and requeues it instead of running it on the wrong worker. With `go()` + `set_affinity()`, the task may briefly run on any worker before migrating. +#### Worker Lifetime Lease + +Include `` (also in ``). + +```cpp +auto lease = co_await elio::runtime::lease_current_worker(); +size_t owner = lease.worker_id(); +uint64_t generation = lease.context_generation(); +bool active = lease.active(); +// Settle external work and destroy owner-thread resources before release. +lease.release(); +``` + +`external_worker_lease` is move-only with RAII release. It retains the acquiring +logical task's exact worker through normal shrink/drain, independently of +caller affinity. Foreign-thread release is supported with serialized access to +the token. Acquisition rejects retiring workers and shutdown. It does not +drive an external ring or prevent force shutdown. See [[Worker Leases]] for +ownership, progress, and teardown rules. + #### `elio::spawn()` - Joinable Spawn a coroutine and return a `join_handle` to await the result later. diff --git a/wiki/Home.md b/wiki/Home.md index cd9c0f95..53e26d24 100644 --- a/wiki/Home.md +++ b/wiki/Home.md @@ -79,6 +79,7 @@ Elio is built around a few key technical decisions: - [[Getting Started]] - Installation and first steps - [[Core Concepts]] - Coroutines, tasks, and scheduler +- [[Worker Leases]] - Lifetime and exact-owner execution for external resources - [[API Contracts]] - Per-interface library guarantees and caller responsibilities for audit and triage - [[Fork Safety]] - Supported process creation boundaries and inherited-runtime restrictions diff --git a/wiki/Worker-Leases.md b/wiki/Worker-Leases.md new file mode 100644 index 00000000..52d59e76 --- /dev/null +++ b/wiki/Worker-Leases.md @@ -0,0 +1,73 @@ +# Worker Lifetime Leases + +`` provides `elio::runtime::lease_current_worker()` +and the move-only `external_worker_lease` type. They are also available through +``. + +A lease represents a long-lived external obligation tied to one worker OS +thread. Examples include an external completion queue or a thread-affine device +runtime. Unlike best-effort affinity, an active lease delays normal worker +retirement even after a pool shrink removes that worker from `num_threads()`. +The acquiring logical task, including directly awaited Elio child frames, +continues to execute on that exact owner. Independent spawned roots do not +inherit the lease's execution constraint. + +```cpp +auto owner = co_await elio::runtime::lease_current_worker(); +// Initialize an external owner on this thread. +// Drive it in bounded turns, awaiting events between turns as appropriate. +// Drain all external commands and destroy thread-affine resources here. +owner.release(); +``` + +Acquisition does not suspend on success. It requires an Elio task currently +executing on a running scheduler worker. Outside a worker it throws +`std::logic_error`; after retirement admission or shutdown has closed it throws +`std::runtime_error`. Allocation or execution-context setup may also throw. +Acquisition is serialized with retirement admission: either the obligation is +counted before drain, or acquisition is rejected. + +## Ownership And Release + +The runnable `examples/worker_lease.cpp` demonstrates retaining an event waiter +on worker 1 while a controller shrinks the visible pool to one worker. + +- `active()` reports whether this token retains its obligation. A default or + moved-from token is inactive. Move assignment releases the destination's old + obligation. Destruction releases any remaining obligation. +- `worker_id()` and `context_generation()` identify the acquired context. These + are diagnostic snapshots, not reusable worker handles. After release they do + not authorize scheduling or access to a later worker incarnation. +- `release()` is idempotent and may run on a foreign thread. Serialize access + to the token itself; concurrent release/move/inspection of one token is not + supported. Release removes the acquiring task's ownership constraint and + wakes the owner so retirement can progress. +- Moving a token does not transfer the original logical task's execution + constraint to a different task. Keep the owner continuation in the acquiring + logical task, or arrange an explicit owner-service path before release. +- An inactive token and the metadata of a token retained after forced teardown + are safe to destroy after the scheduler. That does not make an external ring, + queue, borrowed buffer, or suspended task safe to abandon. + +## Progress And Shutdown + +The lease contributes to pending/active accounting even when the acquiring task +is waiting on a semaphore/event and has no pending native I/O. Draining workers +continue bounded owner-task service and native backend polling. A continuously +yielding owner cannot trap retirement in unbounded task redistribution. + +The application still owns the external protocol, completion progress, +cancellation, fairness, and resource teardown. A lease is not a polling hook, +does not expose the backend ring, and does not automatically drive another ring +through an eventfd. Never block the worker on an external blocking wait. + +Graceful `shutdown()` waits for the obligation within its existing timeout +policy. A timed-out shutdown and `shutdown_force()` retain Elio's existing +non-graceful behavior and may stop workers with active leases. Release external +owners before scheduler teardown for correctness; a lease does not make forced +frame destruction safe or turn cooperative cancellation into completion. + +Native protocol integrations can account for their obligations internally; +their users need not manually acquire a lease unless their own external +ownership requires one. See [[API Contracts]] for the general responsibility +boundary. From ec68a3f2eb3a35d1dae57b31b4aa94bab83ec605 Mon Sep 17 00:00:00 2001 From: Coldwings Date: Fri, 23 Oct 2026 12:19:33 +0800 Subject: [PATCH 2/5] fix(runtime): preserve leased owners across notification and scope return --- include/elio/coro/promise_base.hpp | 7 + include/elio/coro/task.hpp | 11 ++ include/elio/coro/task_execution_context.hpp | 32 +++- include/elio/io/io_operation_guard.hpp | 13 +- include/elio/runtime/scheduler.hpp | 38 ++++- include/elio/runtime/worker_thread.hpp | 11 +- tests/unit/test_worker_lease.cpp | 170 ++++++++++++++++++- wiki/Worker-Leases.md | 8 + 8 files changed, 280 insertions(+), 10 deletions(-) diff --git a/include/elio/coro/promise_base.hpp b/include/elio/coro/promise_base.hpp index 5d2d7cf2..e6577216 100644 --- a/include/elio/coro/promise_base.hpp +++ b/include/elio/coro/promise_base.hpp @@ -267,6 +267,13 @@ class promise_base { return execution_context_; } + /// @internal Lease-aware continuation handoff without materializing policy. + [[nodiscard]] std::shared_ptr + pinned_continuation_owner() const noexcept { + return execution_context_ + ? execution_context_->pinned_continuation_owner() : nullptr; + } + /// Bind an unstarted transparent child to the context of the Elio promise /// that actually awaits it. A task created outside the awaiter may already /// have a provisional independent context; no task body has run yet, so diff --git a/include/elio/coro/task.hpp b/include/elio/coro/task.hpp index f6178ad3..efd0cbfa 100644 --- a/include/elio/coro/task.hpp +++ b/include/elio/coro/task.hpp @@ -74,6 +74,17 @@ struct final_awaiter { h.promise().unlink_parent_cancellation(); h.promise().leave_frame_context(); if (continuation) { + // Cancellation-isolated scopes may execute elsewhere; returning + // from one must restore a still-leased parent's exact owner. + if (h.promise().direct_await_context_isolated()) { + if (auto* parent = h.promise().parent()) { + auto owner = parent->pinned_continuation_owner(); + if (owner && !owner->is_current()) { + owner->schedule(continuation); + return std::noop_coroutine(); + } + } + } return continuation; } else if (h.promise().detached_) { // Keep the externally owned join state after freeing the frame so diff --git a/include/elio/coro/task_execution_context.hpp b/include/elio/coro/task_execution_context.hpp index d2b90426..e47c5190 100644 --- a/include/elio/coro/task_execution_context.hpp +++ b/include/elio/coro/task_execution_context.hpp @@ -5,6 +5,7 @@ #include #include #include +#include #include #include #include @@ -21,6 +22,12 @@ class task_execution_context; class promise_base; namespace detail { +struct continuation_owner { + virtual ~continuation_owner() = default; + [[nodiscard]] virtual bool is_current() const noexcept = 0; + virtual void schedule(std::coroutine_handle<>) noexcept = 0; +}; + struct task_execution_control_block; [[nodiscard]] std::shared_ptr make_task_execution_context(); @@ -106,6 +113,14 @@ class task_execution_context final { return active_io_pins_.load(std::memory_order_acquire); } + /// @internal A lease also carries its scheduler domain for foreign wakes. + [[nodiscard]] std::shared_ptr + pinned_continuation_owner() const noexcept { + if (continuation_pins_.load(std::memory_order_acquire) == 0) return {}; + std::lock_guard lock(io_pin_mutex_); + return continuation_owner_; + } + [[nodiscard]] bool is_io_pin_owner( size_t worker_id, uint64_t context_generation) const noexcept { return has_active_io_pin() && @@ -151,7 +166,8 @@ class task_execution_context final { return cancellation_context_.link_parent(std::move(parent)); } - void acquire_io_pin(size_t worker_id, uint64_t context_generation) { + void acquire_io_pin(size_t worker_id, uint64_t context_generation, + std::shared_ptr owner = {}) { std::lock_guard lock(io_pin_mutex_); size_t count = active_io_pins_.load(std::memory_order_relaxed); if (count != 0 && @@ -169,11 +185,16 @@ class task_execution_context final { io_context_generation_.store( context_generation, std::memory_order_relaxed); } + if (owner) { + continuation_owner_ = std::move(owner); + continuation_pins_.fetch_add(1, std::memory_order_release); + } active_io_pins_.store(count + 1, std::memory_order_release); } void release_io_pin(size_t worker_id, - uint64_t context_generation) noexcept { + uint64_t context_generation, + bool routes_continuation = false) noexcept { std::lock_guard lock(io_pin_mutex_); size_t count = active_io_pins_.load(std::memory_order_relaxed); const bool valid = @@ -184,6 +205,11 @@ class task_execution_context final { assert(valid && "I/O operation guard released a mismatched task pin"); if (!valid) return; + if (routes_continuation && + continuation_pins_.fetch_sub(1, std::memory_order_acq_rel) == 1) { + continuation_owner_.reset(); + } + // Owner and generation deliberately remain as the last observed // identity when count reaches zero. The count is authoritative, and // retaining immutable snapshot fields avoids torn clear/read windows @@ -198,6 +224,8 @@ class task_execution_context final { std::atomic io_owner_worker_{NO_AFFINITY}; std::atomic io_context_generation_{0}; std::atomic active_io_pins_{0}; + std::atomic continuation_pins_{0}; + std::shared_ptr continuation_owner_; }; namespace detail { diff --git a/include/elio/io/io_operation_guard.hpp b/include/elio/io/io_operation_guard.hpp index 8916bd11..cc734a52 100644 --- a/include/elio/io/io_operation_guard.hpp +++ b/include/elio/io/io_operation_guard.hpp @@ -22,7 +22,8 @@ class io_operation_guard final { io_operation_guard( std::shared_ptr execution_context, - std::shared_ptr identity) + std::shared_ptr identity, + std::shared_ptr owner = {}) : execution_context_(std::move(execution_context)) , identity_(std::move(identity)) { if (!identity_) { @@ -32,8 +33,10 @@ class io_operation_guard final { if (identity_->is_worker_owned()) { if (execution_context_) { + routes_continuation_ = bool(owner); execution_context_->acquire_io_pin( - identity_->owner_worker_id, identity_->generation); + identity_->owner_worker_id, identity_->generation, + std::move(owner)); task_pinned_ = true; } } @@ -69,12 +72,14 @@ class io_operation_guard final { // during the release transition. if (task_pinned_) { execution_context_->release_io_pin( - identity_->owner_worker_id, identity_->generation); + identity_->owner_worker_id, identity_->generation, + routes_continuation_); } identity_->active_pins.fetch_sub(1, std::memory_order_acq_rel); active_ = false; task_pinned_ = false; + routes_continuation_ = false; execution_context_.reset(); identity_.reset(); } @@ -89,12 +94,14 @@ class io_operation_guard final { identity_ = std::move(other.identity_); active_ = std::exchange(other.active_, false); task_pinned_ = std::exchange(other.task_pinned_, false); + routes_continuation_ = std::exchange(other.routes_continuation_, false); } std::shared_ptr execution_context_; std::shared_ptr identity_; bool active_ = false; bool task_pinned_ = false; + bool routes_continuation_ = false; }; } // namespace elio::io::detail diff --git a/include/elio/runtime/scheduler.hpp b/include/elio/runtime/scheduler.hpp index 43404601..d3461d35 100644 --- a/include/elio/runtime/scheduler.hpp +++ b/include/elio/runtime/scheduler.hpp @@ -1658,6 +1658,14 @@ inline scheduler* get_current_scheduler() noexcept { inline void schedule_handle(std::coroutine_handle<> handle) noexcept { if (!handle) return; + // A foreign notifier's TLS scheduler is not the lease's scheduler domain. + if (auto* promise = coro::get_promise_base(handle.address())) { + if (auto owner = promise->pinned_continuation_owner()) { + owner->schedule(handle); + return; + } + } + auto* sched = scheduler::current(); if (sched) { // Rejection leaves this borrowed handle live. Prefer normal routing; @@ -1770,6 +1778,33 @@ inline void worker_thread::start() { thread_ = std::thread(&worker_thread::run, this); } +inline bool detail::external_worker_wakeup::is_current() const noexcept { + std::lock_guard lock(mutex); + return worker && worker_thread::current() == worker; +} + +inline void detail::external_worker_wakeup::schedule( + std::coroutine_handle<> handle) noexcept { + std::unique_lock lock(mutex); + auto* promise = coro::get_promise_base(handle.address()); + if (!promise || promise->pinned_continuation_owner().get() != this) { + // The final lease may have been released after the notifier captured + // this endpoint. Do not drop an otherwise movable continuation. + lock.unlock(); + schedule_handle(handle); + return; + } + if (!worker) return; + if (worker_thread::current() == worker) { + worker->schedule_local(handle); + return; + } + while (worker->is_running()) { + if (worker->schedule(handle)) return; + std::this_thread::yield(); + } +} + inline io::detail::io_operation_guard worker_thread::acquire_external_lease( std::shared_ptr execution_context) { if (current() != this || !execution_context) { @@ -1783,7 +1818,8 @@ inline io::detail::io_operation_guard worker_thread::acquire_external_lease( throw std::runtime_error("worker lease admission is closed"); } return io::detail::io_operation_guard( - std::move(execution_context), io_context_->operation_identity()); + std::move(execution_context), io_context_->operation_identity(), + external_wakeup_); } inline void worker_thread::request_stop() noexcept { diff --git a/include/elio/runtime/worker_thread.hpp b/include/elio/runtime/worker_thread.hpp index 3152a3e7..92d225f7 100644 --- a/include/elio/runtime/worker_thread.hpp +++ b/include/elio/runtime/worker_thread.hpp @@ -17,11 +17,16 @@ namespace elio::runtime { class scheduler; +class worker_thread; namespace detail { -struct external_worker_wakeup { - std::mutex mutex; +struct external_worker_wakeup final : coro::detail::continuation_owner { + [[nodiscard]] bool is_current() const noexcept override; + void schedule(std::coroutine_handle<> handle) noexcept override; + + mutable std::mutex mutex; io::io_context* context = nullptr; + worker_thread* worker = nullptr; }; // Allocated with the operation, not with its cancellation request. Only the @@ -83,12 +88,14 @@ class worker_thread { #endif { external_wakeup_->context = io_context_.get(); + external_wakeup_->worker = this; } ~worker_thread() { stop(); std::lock_guard lock(external_wakeup_->mutex); external_wakeup_->context = nullptr; + external_wakeup_->worker = nullptr; } worker_thread(const worker_thread&) = delete; diff --git a/tests/unit/test_worker_lease.cpp b/tests/unit/test_worker_lease.cpp index 2a361580..268ca434 100644 --- a/tests/unit/test_worker_lease.cpp +++ b/tests/unit/test_worker_lease.cpp @@ -1,9 +1,12 @@ #include #include #include +#include +#include #include #include #include +#include #include #include #include @@ -42,9 +45,19 @@ static_assert(std::is_nothrow_move_assignable_v); TEST_CASE("worker lease retains an event waiter through shrink", "[worker_lease][scheduler]") { const auto selected = GENERATE(io::io_context::backend_type::epoll, - io::io_context::backend_type::auto_detect); + io::io_context::backend_type::auto_detect, + io::io_context::backend_type::io_uring); backend_selection backend(selected); - scheduler sched(2); + std::optional storage; + try { + storage.emplace(2); + } catch (const std::runtime_error& error) { + if (selected == io::io_context::backend_type::io_uring) { + SKIP("Explicit io_uring is unavailable: " << error.what()); + } + throw; + } + auto& sched = *storage; auto* owner = sched.get_worker(1); sync::event gate; std::atomic acquired{false}, finished{false}, correct_owner{false}; @@ -53,9 +66,11 @@ TEST_CASE("worker lease retains an event waiter through shrink", "[worker_lease] sched.go_to(1, [&]() -> coro::task { auto first = co_await lease_current_worker(); auto second = co_await lease_current_worker(); + auto third = co_await lease_current_worker(); const auto generation = first.context_generation(); external_worker_lease moved(std::move(first)); first = std::move(second); + moved = std::move(third); move_valid.store(!second.active() && moved.active() && first.active() && owner->io_context().active_pin_count() == 2); acquired.store(true, std::memory_order_release); @@ -301,3 +316,154 @@ TEST_CASE("stale lease release cannot affect another scheduler generation", "[wo REQUIRE(unchanged); REQUIRE(distinct); } + +TEST_CASE("foreign notifier preserves a leased event waiter's exact owner", "[worker_lease][scheduler]") { + const bool different_scheduler = GENERATE(false, true); + backend_selection backend(io::io_context::backend_type::epoll); + scheduler sched(2); + sync::event gate; + std::atomic acquired{false}, finished{false}, correct{false}; + sched.start(); + sched.go_to(1, [&]() -> coro::task { + auto lease = co_await lease_current_worker(); + const auto owner = std::this_thread::get_id(); + acquired.store(true, std::memory_order_release); + co_await gate.wait(); + correct.store(std::this_thread::get_id() == owner); + lease.release(); + finished.store(true, std::memory_order_release); + }); + const bool started = eventually([&] { return acquired.load(std::memory_order_acquire); }); + // Ensure wait() has actually suspended, rather than taking its ready path. + const bool suspended = eventually([&] { return sched.get_worker(1)->is_idle(); }); + sched.set_thread_count(1); + if (different_scheduler) { + scheduler notifier(1); + notifier.start(); + notifier.go([&]() -> coro::task { gate.set(); co_return; }); + notifier.shutdown(); + } else { + std::thread notifier([&] { gate.set(); }); + notifier.join(); + } + const bool completed = eventually([&] { return finished.load(std::memory_order_acquire); }); + if (completed) sched.shutdown(); + else sched.shutdown_force(); + REQUIRE(started); + REQUIRE(suspended); + REQUIRE(completed); + REQUIRE(correct.load()); +} + +TEST_CASE("foreign cancellation preserves a leased waiter's owner", "[worker_lease][scheduler]") { + backend_selection backend(io::io_context::backend_type::epoll); + scheduler sched(2); + sync::event gate; + coro::cancel_source source; + std::atomic acquired{false}, finished{false}, correct{false}; + sched.start(); + sched.go_to(1, [&]() -> coro::task { + auto lease = co_await lease_current_worker(); + const auto owner = std::this_thread::get_id(); + acquired.store(true, std::memory_order_release); + const auto result = co_await gate.wait(source.get_token()); + correct.store(result == coro::cancel_result::cancelled && + std::this_thread::get_id() == owner); + lease.release(); + finished.store(true, std::memory_order_release); + }); + const bool started = eventually([&] { return acquired.load(std::memory_order_acquire); }); + const bool suspended = eventually([&] { return sched.get_worker(1)->is_idle(); }); + sched.set_thread_count(1); + std::thread canceller([&] { source.cancel(); }); + canceller.join(); + const bool completed = eventually([&] { return finished.load(std::memory_order_acquire); }); + if (completed) sched.shutdown(); + else sched.shutdown_force(); + REQUIRE(started); + REQUIRE(suspended); + REQUIRE(completed); + REQUIRE(correct.load()); +} + +TEST_CASE("foreign lease release races safely with context destruction", "[worker_lease][scheduler]") { + backend_selection backend(io::io_context::backend_type::epoll); + for (int iteration = 0; iteration < 16; ++iteration) { + auto sched = std::make_unique(1); + std::optional retained; + std::atomic acquired{false}; + sched->start(); + sched->go([&]() -> coro::task { + retained.emplace(co_await lease_current_worker()); + acquired.store(true, std::memory_order_release); + co_return; + }); + const bool started = eventually([&] { return acquired.load(std::memory_order_acquire); }); + if (!started) { sched->shutdown_force(); FAIL("lease acquisition did not run"); } + std::barrier start(2); + std::thread releaser([&] { start.arrive_and_wait(); retained->release(); }); + start.arrive_and_wait(); + sched.reset(); + releaser.join(); + REQUIRE_FALSE(retained->active()); + } +} + +TEST_CASE("isolated scope returns a leased parent to its owner", "[worker_lease][scheduler]") { + backend_selection backend(io::io_context::backend_type::epoll); + scheduler sched(2); + std::atomic finished{false}, correct{false}; + sched.start(); + sched.go_to(1, [&]() -> coro::task { + auto lease = co_await lease_current_worker(); + const auto owner = std::this_thread::get_id(); + co_await coro::task_scope([](coro::task_group&) -> coro::task { + co_await runtime::set_affinity(0); + }); + correct.store(std::this_thread::get_id() == owner); + lease.release(); + finished.store(true, std::memory_order_release); + }); + const bool completed = eventually([&] { return finished.load(std::memory_order_acquire); }); + sched.shutdown(); + REQUIRE(completed); + REQUIRE(correct.load()); +} + +TEST_CASE("captured owner endpoint falls back after final lease release", "[worker_lease][scheduler]") { + backend_selection backend(io::io_context::backend_type::epoll); + scheduler sched(2); + auto* old_owner = sched.get_worker(1); + std::optional retained; + std::shared_ptr endpoint; + std::atomic suspended{nullptr}; + std::atomic finished{false}; + struct manual_wait { + std::atomic* address; + bool await_ready() const noexcept { return false; } + void await_suspend(std::coroutine_handle<> handle) const noexcept { + address->store(handle.address(), std::memory_order_release); + } + void await_resume() const noexcept {} + }; + sched.start(); + sched.go_to(1, [&]() -> coro::task { + retained.emplace(co_await lease_current_worker()); + endpoint = coro::promise_base::current_frame()->pinned_continuation_owner(); + co_await manual_wait{&suspended}; + finished.store(true, std::memory_order_release); + }); + const bool parked = eventually([&] { return suspended.load(std::memory_order_acquire) != nullptr; }); + sched.set_thread_count(1); + if (parked) retained->release(); + const bool retired = eventually([&] { return !old_owner->is_running(); }); + // Models capture -> final release -> retire -> notifier dispatch, without + // relying on timing to land between lookup and endpoint invocation. + if (parked) endpoint->schedule(std::coroutine_handle<>::from_address(suspended.load())); + const bool completed = eventually([&] { return finished.load(std::memory_order_acquire); }); + if (completed) sched.shutdown(); + else sched.shutdown_force(); + REQUIRE(parked); + REQUIRE(retired); + REQUIRE(completed); +} diff --git a/wiki/Worker-Leases.md b/wiki/Worker-Leases.md index 52d59e76..cc1da4b4 100644 --- a/wiki/Worker-Leases.md +++ b/wiki/Worker-Leases.md @@ -12,6 +12,14 @@ The acquiring logical task, including directly awaited Elio child frames, continues to execute on that exact owner. Independent spawned roots do not inherit the lease's execution constraint. +Foreign-thread event notifications and cancellation wakeups return the leased +task to its retained scheduler/worker, not the notifier's scheduler. An isolated +`task_scope()` keeps its own cancellation and placement context and may migrate; +its return restores a still-leased parent to the parent's owner. Do not access +the parent's thread-affine resources from a scope body that migrates elsewhere. +An application-provided awaitable that directly resumes on a foreign executor +must arrange its own owner handoff before thread-affine code runs. + ```cpp auto owner = co_await elio::runtime::lease_current_worker(); // Initialize an external owner on this thread. From 2117bb4abd4b57f35d7817652c1a650d7ee9ffbf Mon Sep 17 00:00:00 2001 From: Coldwings Date: Fri, 23 Oct 2026 12:24:43 +0800 Subject: [PATCH 3/5] fix(coro): restore leased consumers across generator transfers --- include/elio/coro/generator.hpp | 35 +++++++++++++++--------- include/elio/coro/promise_base.hpp | 12 ++++++++ include/elio/coro/task.hpp | 9 ++---- tests/unit/test_worker_lease.cpp | 44 ++++++++++++++++++++++++++++++ wiki/Worker-Leases.md | 3 ++ 5 files changed, 83 insertions(+), 20 deletions(-) diff --git a/include/elio/coro/generator.hpp b/include/elio/coro/generator.hpp index 2c68beaf..5d61dabd 100644 --- a/include/elio/coro/generator.hpp +++ b/include/elio/coro/generator.hpp @@ -27,7 +27,8 @@ namespace detail { /// Minimal coroutine type for generator::for_each implementation. /// Acts as both the coroutine return type and an awaitable. struct for_each_task { - struct promise_type { + struct promise_type : promise_base { + promise_type() : promise_base(defer_nested_task_context) {} std::exception_ptr exception_; std::coroutine_handle<> consumer_; @@ -44,10 +45,8 @@ struct for_each_task { [[nodiscard]] std::coroutine_handle<> await_suspend( std::coroutine_handle h) noexcept { auto consumer = h.promise().consumer_; - if (consumer) { - return consumer; - } - return std::noop_coroutine(); + h.promise().leave_frame_context(); + return transfer_to_continuation(consumer, h.promise().parent()); } void await_resume() const noexcept {} }; @@ -64,7 +63,9 @@ struct for_each_task { using handle_type = std::coroutine_handle; handle_type handle_; - for_each_task(handle_type h) noexcept : handle_(h) {} + for_each_task(handle_type h) noexcept : handle_(h) { + handle_.promise().leave_creation_context(); + } for_each_task(for_each_task&& other) noexcept : handle_(std::exchange(other.handle_, nullptr)) {} @@ -82,9 +83,20 @@ struct for_each_task { return !handle_ || handle_.done(); } + template [[nodiscard]] std::coroutine_handle<> await_suspend( - std::coroutine_handle<> consumer) noexcept { - handle_.promise().consumer_ = consumer; + std::coroutine_handle consumer) { + auto& promise = handle_.promise(); + promise_base* parent = nullptr; + if constexpr (std::is_convertible_v) { + parent = std::addressof(consumer.promise()); + parent->ensure_independent_execution_context(); + promise.bind_direct_await_context(parent->execution_context()); + } else { + promise.ensure_independent_execution_context(); + } + promise.enter_frame_context(parent); + promise.consumer_ = consumer; return handle_; // symmetric transfer to the loop coroutine } @@ -182,10 +194,7 @@ class generator { auto& promise = h.promise(); auto consumer = promise.consumer_; promise.leave_frame_context(); - if (consumer) { - return consumer; // symmetric transfer → consumer - } - return std::noop_coroutine(); + return detail::transfer_to_continuation(consumer, promise.parent()); } void await_resume() const noexcept {} @@ -206,7 +215,7 @@ class generator { [[nodiscard]] std::coroutine_handle<> await_suspend( std::coroutine_handle h) noexcept { h.promise().leave_frame_context(); - return consumer; // symmetric transfer → consumer + return detail::transfer_to_continuation(consumer, h.promise().parent()); } void await_resume() noexcept {} diff --git a/include/elio/coro/promise_base.hpp b/include/elio/coro/promise_base.hpp index e6577216..50ca6cb5 100644 --- a/include/elio/coro/promise_base.hpp +++ b/include/elio/coro/promise_base.hpp @@ -480,6 +480,18 @@ static_assert(std::is_standard_layout_v, /// resumer. namespace detail { +[[nodiscard]] inline std::coroutine_handle<> transfer_to_continuation( + std::coroutine_handle<> continuation, promise_base* parent) noexcept { + if (continuation && parent) { + auto owner = parent->pinned_continuation_owner(); + if (owner && !owner->is_current()) { + owner->schedule(continuation); + return std::noop_coroutine(); + } + } + return continuation ? continuation : std::noop_coroutine(); +} + class frame_context_scope { public: explicit frame_context_scope(promise_base* frame) noexcept diff --git a/include/elio/coro/task.hpp b/include/elio/coro/task.hpp index efd0cbfa..401133f2 100644 --- a/include/elio/coro/task.hpp +++ b/include/elio/coro/task.hpp @@ -77,13 +77,8 @@ struct final_awaiter { // Cancellation-isolated scopes may execute elsewhere; returning // from one must restore a still-leased parent's exact owner. if (h.promise().direct_await_context_isolated()) { - if (auto* parent = h.promise().parent()) { - auto owner = parent->pinned_continuation_owner(); - if (owner && !owner->is_current()) { - owner->schedule(continuation); - return std::noop_coroutine(); - } - } + return transfer_to_continuation( + continuation, h.promise().parent()); } return continuation; } else if (h.promise().detached_) { diff --git a/tests/unit/test_worker_lease.cpp b/tests/unit/test_worker_lease.cpp index 268ca434..8f88c5fd 100644 --- a/tests/unit/test_worker_lease.cpp +++ b/tests/unit/test_worker_lease.cpp @@ -2,6 +2,7 @@ #include #include #include +#include #include #include #include @@ -242,6 +243,11 @@ TEST_CASE("graceful shutdown waits for independent lease release", "[worker_leas } namespace { +coro::generator gated_producer(sync::event* gate, bool yield_value) { + co_await gate->wait(); + if (yield_value) co_yield 42; +} + coro::task acquire_without_worker(bool* rejected) { try { auto lease = co_await lease_current_worker(); @@ -467,3 +473,41 @@ TEST_CASE("captured owner endpoint falls back after final lease release", "[work REQUIRE(retired); REQUIRE(completed); } + +TEST_CASE("generator yield and exhaustion restore a leased consumer", "[worker_lease][scheduler]") { + const bool yield_value = GENERATE(false, true); + const bool for_each = GENERATE(false, true); + backend_selection backend(io::io_context::backend_type::epoll); + scheduler sched(2); + sync::event gate; + auto producer = gated_producer(&gate, yield_value); + std::atomic started{false}, finished{false}, correct{false}, callback_correct{true}; + sched.start(); + sched.go_to(1, [&]() -> coro::task { + auto lease = co_await lease_current_worker(); + const auto owner = std::this_thread::get_id(); + started.store(true, std::memory_order_release); + if (for_each) { + co_await producer.for_each([&](int) { + callback_correct.store(std::this_thread::get_id() == owner); + }); + } else { + (void)co_await producer.next(); + } + correct.store(std::this_thread::get_id() == owner); + lease.release(); + finished.store(true, std::memory_order_release); + }); + const bool entered = eventually([&] { return started.load(std::memory_order_acquire); }); + const bool suspended = eventually([&] { return sched.get_worker(1)->is_idle(); }); + sched.set_thread_count(1); + gate.set(); + const bool completed = eventually([&] { return finished.load(std::memory_order_acquire); }); + if (completed) sched.shutdown(); + else sched.shutdown_force(); + REQUIRE(entered); + REQUIRE(suspended); + REQUIRE(completed); + REQUIRE(correct.load()); + REQUIRE(callback_correct.load()); +} diff --git a/wiki/Worker-Leases.md b/wiki/Worker-Leases.md index cc1da4b4..34d5b86e 100644 --- a/wiki/Worker-Leases.md +++ b/wiki/Worker-Leases.md @@ -17,6 +17,9 @@ task to its retained scheduler/worker, not the notifier's scheduler. An isolated `task_scope()` keeps its own cancellation and placement context and may migrate; its return restores a still-leased parent to the parent's owner. Do not access the parent's thread-affine resources from a scope body that migrates elsewhere. +Async generators likewise keep an independent producer context, but yield and +exhaustion restore a leased consumer to its owner. `for_each()`'s internal loop +shares its Elio caller's context so callbacks and loop return honor that lease. An application-provided awaitable that directly resumes on a foreign executor must arrange its own owner handoff before thread-affine code runs. From 575af21613b883a0b5429ccea831a9132080027b Mon Sep 17 00:00:00 2001 From: Coldwings Date: Fri, 23 Oct 2026 12:28:27 +0800 Subject: [PATCH 4/5] fix(runtime): retain leased owners through completion fallbacks --- CHANGELOG.md | 4 +- include/elio/coro/task_group.hpp | 6 +++ include/elio/runtime/spawn_blocking.hpp | 7 ++- include/elio/time/timer.hpp | 6 ++- tests/unit/test_worker_lease.cpp | 65 +++++++++++++++++++++++++ wiki/API-Contracts.md | 2 +- 6 files changed, 85 insertions(+), 5 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c6640954..dfc89207 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,7 +12,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Move-only worker lifetime leases for external thread-affine owners. Lease acquisition retains exact-owner continuations and worker drain accounting; release safely wakes retirement, including from a foreign thread. Retiring - workers service requeued owner tasks in bounded turns (#1301). + workers service requeued owner tasks in bounded turns. Foreign wakeups, + isolated-scope returns, generator consumers, and blocking/timer/group + completion paths honor the retained owner (#1301). ## [0.6.0] - 2026-10-12 diff --git a/include/elio/coro/task_group.hpp b/include/elio/coro/task_group.hpp index 6131a450..3cc809d9 100644 --- a/include/elio/coro/task_group.hpp +++ b/include/elio/coro/task_group.hpp @@ -87,6 +87,12 @@ inline void resume_task_group_join_waiter( if (!waiter) { return; } + if (auto* promise = get_promise_base(waiter.address())) { + if (auto owner = promise->pinned_continuation_owner()) { + owner->schedule(waiter); + return; + } + } if (completion.scheduler->try_schedule(waiter)) { return; } diff --git a/include/elio/runtime/spawn_blocking.hpp b/include/elio/runtime/spawn_blocking.hpp index e5e26cb0..77efe12b 100644 --- a/include/elio/runtime/spawn_blocking.hpp +++ b/include/elio/runtime/spawn_blocking.hpp @@ -166,7 +166,11 @@ class blocking_awaitable { // We claimed resume rights. The destructor will spin-wait until // we store kDone, so the frame stays alive throughout this block. - if (sched) { + auto* promise = coro::get_promise_base(caller.address()); + auto owner = promise ? promise->pinned_continuation_owner() : nullptr; + if (owner) { + owner->schedule(caller); + } else if (sched) { if (sched->is_running()) { // A live worker accepts through its bounded inbox or // overflow queue. shutdown_force() drains accepted pool @@ -176,7 +180,6 @@ class blocking_awaitable { } } else if (caller && !caller.done()) { inline_resume_guard guard(state.get()); - auto* promise = coro::get_promise_base(caller.address()); coro::detail::frame_context_scope frame_scope(promise); caller.resume(); } diff --git a/include/elio/time/timer.hpp b/include/elio/time/timer.hpp index 0564829c..5f316cf0 100644 --- a/include/elio/time/timer.hpp +++ b/include/elio/time/timer.hpp @@ -96,6 +96,11 @@ inline bool submit_blocking(F&& work) { inline void resume_via_scheduler(std::coroutine_handle<> handle, runtime::scheduler* sched) noexcept { if (!handle) return; + auto* promise = coro::get_promise_base(handle.address()); + if (auto owner = promise ? promise->pinned_continuation_owner() : nullptr) { + owner->schedule(handle); + return; + } if (sched) { if (sched->is_running() && sched->try_schedule(handle)) { return; @@ -105,7 +110,6 @@ inline void resume_via_scheduler(std::coroutine_handle<> handle, return; } if (!handle.done()) { - auto* promise = coro::get_promise_base(handle.address()); coro::detail::frame_context_scope frame_scope(promise); handle.resume(); } diff --git a/tests/unit/test_worker_lease.cpp b/tests/unit/test_worker_lease.cpp index 8f88c5fd..f450fd94 100644 --- a/tests/unit/test_worker_lease.cpp +++ b/tests/unit/test_worker_lease.cpp @@ -4,6 +4,7 @@ #include #include #include +#include #include #include #include @@ -511,3 +512,67 @@ TEST_CASE("generator yield and exhaustion restore a leased consumer", "[worker_l REQUIRE(correct.load()); REQUIRE(callback_correct.load()); } + +TEST_CASE("task group rejection fallback cannot resume a leased joiner off-owner", "[worker_lease][scheduler]") { + backend_selection backend(io::io_context::backend_type::epoll); + scheduler sched(2); + sync::event gate; + std::atomic child_waiting{false}, finished{false}, correct{false}; + sched.start(); + sched.go_to(1, [&]() -> coro::task { + auto lease = co_await lease_current_worker(); + const auto owner = std::this_thread::get_id(); + coro::task_group group(sched); + group.spawn([&]() -> coro::task { + co_await runtime::set_affinity(0); + child_waiting.store(true, std::memory_order_release); + co_await gate.wait(); + runtime::detail::reject_next_schedule_for_test.store(true); + }); + co_await group.join(); + correct.store(std::this_thread::get_id() == owner); + lease.release(); + finished.store(true, std::memory_order_release); + }); + const bool child_started = eventually([&] { return child_waiting.load(std::memory_order_acquire); }); + const bool parent_waiting = eventually([&] { return sched.get_worker(1)->is_idle(); }); + gate.set(); + const bool completed = eventually([&] { return finished.load(std::memory_order_acquire); }); + runtime::detail::reject_next_schedule_for_test.store(false); + if (completed) sched.shutdown(); + else sched.shutdown_force(); + REQUIRE(child_started); + REQUIRE(parent_waiting); + REQUIRE(completed); + REQUIRE(correct.load()); +} + +TEST_CASE("blocking and timer completions retain leased owners on rejection", "[worker_lease][scheduler]") { + const bool timer_fallback = GENERATE(false, true); + backend_selection backend(io::io_context::backend_type::epoll); + scheduler sched(1); + std::atomic finished{false}, correct{false}; + sched.start(); + sched.go([&]() -> coro::task { + auto lease = co_await lease_current_worker(); + const auto owner = std::this_thread::get_id(); + if (timer_fallback) { + time::detail::reject_next_timeout_prepare_for_test(); + runtime::detail::reject_next_schedule_for_test.store(true); + co_await time::sleep_for(1ms); + } else { + co_await elio::spawn_blocking([] { + runtime::detail::reject_next_schedule_for_test.store(true); + }); + } + correct.store(std::this_thread::get_id() == owner); + lease.release(); + finished.store(true, std::memory_order_release); + }); + const bool completed = eventually([&] { return finished.load(std::memory_order_acquire); }); + runtime::detail::reject_next_schedule_for_test.store(false); + if (completed) sched.shutdown(); + else sched.shutdown_force(); + REQUIRE(completed); + REQUIRE(correct.load()); +} diff --git a/wiki/API-Contracts.md b/wiki/API-Contracts.md index ac750a24..cd907948 100644 --- a/wiki/API-Contracts.md +++ b/wiki/API-Contracts.md @@ -60,7 +60,7 @@ by a broad module heading without checking the feature page or header comment. | `join_handle` | Awaiting a join handle returns the spawned task result or rethrows its exception. `is_ready()` is a non-blocking result-readiness check and can precede frame destruction. `is_destroyed()` observes frame release; after `wait_destroyed()` or `co_await wait_destroyed_async()` completes for a normally completed spawn, the root frame and its parameters have been destroyed, including any callable wrapper used by the callable overload. Multiple external blocking waiters and independent async destruction observers are supported. Each async awaitable retains shared state independently of later handle movement, does not consume/rethrow the result or cancel the task, and resumes in its registration scheduler domain. Already-destroyed observations complete without a scheduler; pending registration may allocate or throw for a missing scheduler. Observer teardown removes a registered or selected-but-unclaimed wake. `request_cancel()` publishes a cooperative request through the spawned root's shared execution context and remains safe after frame destruction; it does not rewrite a result that already completed. The request propagates down direct lazy-task awaits between Elio tasks. | Await or intentionally discard join handles according to the task lifetime you need. Call blocking `wait_destroyed()` only from a non-coroutine thread; scheduler workers must not block there. Keep an async observer's scheduler and coroutine owner alive through completion; use each awaitable once and do not move it during suspension or destroy its coroutine after scheduling ownership is claimed. Do not use a moved-from handle or concurrently mutate a handle while creating observations from it. Destruction observation does not cancel work, provide a deadline, drain orphaned I/O, or promise normal teardown ordering during forced frame destruction. Pass `this_coro::cancel_token()` into cancellation-aware waits inside the task. Do not assume `request_cancel()` forcibly destroys the task, stops it promptly, crosses a foreign coroutine promise, or cancels a separate explicit token. Registered callback exceptions can propagate from `request_cancel()`. | | `coro::task_group` | Owns registration, cancellation, failure collection, and completion accounting for children submitted to one scheduler. The default failure policy records the first child failure, requests sibling cancellation, waits for every registered child frame to leave the group, and then rethrows that failure from the direct single-use `join()` awaitable. `join()` does not create or link a nested task. `collect_all` records failures without cancelling siblings and reports them together through `task_group_error`. A nonzero `max_concurrency` bounds executing child bodies while preserving every accepted child registration. `outstanding_children()` counts accepted child submissions, not an active `task_scope()` body; before join closes admission, a newly accepted child may follow another child's transition to zero. Join-wake selection revalidates that no accepted child remains under the same slot lock used to register the waiter. A group created on its scheduler inherits the current task's cancellation token. Its join continuation always resumes on a worker of the selected scheduler. An ordinary external thread, including the thread that called `scheduler::start()`, synchronously dispatches cancellation there; a worker in another scheduler posts cancellation asynchronously to avoid cross-scheduler deadlock. | Call `join()` exactly once while executing on a worker of the group's scheduler, and keep the group alive until it completes. Pass `this_coro::cancel_token()` into cancellation-aware child operations; cancellation is cooperative and cannot finish a child blocked in work that ignores its token. Do not spawn after joining starts. Keep child captures alive through join. Treat an explicitly selected scheduler as the group's only execution domain and keep it running until the group drains. Account for synchronous callback execution and possible blocking on ordinary external threads; do not assume cancellation has run when a request returns on another scheduler's worker. The destructor requests cancellation but does not synchronously join. | | `coro::task_scope()` | Creates a task group for a callback-shaped lexical scope and runs the body under an isolated group cancellation context. Caller cancellation propagates into the scope, but group or fail-fast cancellation does not leave the caller's token cancelled after the scope joins. Caller user affinity is inherited and the scope's final user-affinity value flows back; active I/O pins remain owned by the scope's operation context. Normal return joins all children. A fail-fast child can therefore wake a token-aware suspended body before the scope joins and reports the child failure. If a body awaitable resumes elsewhere, scope cleanup hands execution back to a worker of the selected scheduler. The body callable and its captures remain alive until the children join. If the body throws, the scope requests child cancellation and joins every child. Fail-fast then rethrows the body failure; `collect_all` reports the body failure together with every child failure in `task_group_error`. | Use it when lexical cancel-and-join is required. The body must return an Elio task, must not call `join()` on its own group, and must not allow token-ignoring work to outlive captured resources. The explicit-scheduler overload must initially be awaited on a worker of that scheduler; it does not migrate a foreign worker or external thread into the scope. Automatic local objects in the body coroutine are destroyed when that coroutine returns, so children that can outlive the body must not retain references to those locals. Choose `collect_all` or a concurrency bound explicitly when fail-fast and unlimited execution are not suitable. Keep the selected scheduler running through cleanup and caller resumption. | -| `coro::generator` | `next()` delivers yielded values in producer order and returns completion through the documented optional result. It binds producer virtual-stack ancestry to the active consumer and restores the consumer context before yield and completion transfers, including after internal asynchronous suspension. | Use it as a single-consumer stream unless a future generator API documents sharing. Do not retain yielded references or views beyond their documented lifetime. | +| `coro::generator` | `next()` delivers yielded values in producer order and returns completion through the documented optional result. It binds producer virtual-stack ancestry to the active consumer and restores the consumer context before yield and completion transfers, including after internal asynchronous suspension. A leased consumer returns to its retained owner; the producer's execution context remains independent. The `for_each()` loop shares its Elio caller's context so callbacks honor its lease. | Use it as a single-consumer stream unless a future generator API documents sharing. Do not retain yielded references or views beyond their documented lifetime. Do not assume a consumer lease pins the independent producer body. | | `coro::cancel_source` | Calling `cancel()` publishes cooperative cancellation and synchronously dispatches callbacks outside the state lock. Same-dispatch reentrant teardown may remove a later selected callback. After dispatch, `cancel()` rethrows the first callback exception. Concurrent calls publish one request. | Keep callback code reentry-safe and account for execution on the requesting thread. Do not assume cancellation forcibly destroys operations that do not observe the token. Handle callback exceptions where callbacks may throw. | | `coro::cancel_token` | Cancellation-aware operations observe the token at their documented wait or poll points. Destroying or unregistering a callback registration prevents a callback not yet selected by cancellation. Outside callback dispatch, teardown waits when another thread has selected or started the callback. During callback reentry, cross-dispatch teardown is deferred to prevent mutual wait cycles; the callback payload remains dispatcher-owned. Self-unregistration and reentrant removal of a later callback by the same synchronous dispatcher do not deadlock. | Pass the token into every operation that should stop. Keep externally owned callback state synchronized: teardown called from another cancellation callback is not a cross-dispatch join point. Synchronize shared state accessed by different callbacks. An outer request cannot cancel waits that never receive or check the token. | | `coro::this_coro::cancel_token()` | Returns the current Elio task's runtime cancellation token. Outside an active Elio runtime frame it returns a default never-cancelled token. | Treat it as a cooperative request channel. Pass it into each operation that should react. Explicit token parameters remain independent unless application or framework code deliberately bridges them. | From dd8fe6cc9d5963ade2abbb8eb33459ce0bbc16c6 Mon Sep 17 00:00:00 2001 From: Coldwings Date: Fri, 23 Oct 2026 12:33:52 +0800 Subject: [PATCH 5/5] fix(runtime): preserve scheduler domains after final lease release --- include/elio/coro/generator.hpp | 5 +- include/elio/runtime/scheduler.hpp | 17 ++++-- include/elio/runtime/spawn_blocking.hpp | 14 +++++ include/elio/runtime/worker_thread.hpp | 3 + tests/unit/test_worker_lease.cpp | 81 ++++++++++++++++++++++++- wiki/Worker-Leases.md | 4 ++ 6 files changed, 116 insertions(+), 8 deletions(-) diff --git a/include/elio/coro/generator.hpp b/include/elio/coro/generator.hpp index 5d61dabd..7ab52764 100644 --- a/include/elio/coro/generator.hpp +++ b/include/elio/coro/generator.hpp @@ -282,8 +282,9 @@ class generator { // transferring control back, including after internal I/O. producer.promise().enter_frame_context( promise_base::current_frame()); - // Symmetric transfer → producer runs until co_yield / end. - return producer; + // Producer and consumer have independent ownership domains. + return detail::transfer_to_continuation( + producer, std::addressof(producer.promise())); } std::optional await_resume() { diff --git a/include/elio/runtime/scheduler.hpp b/include/elio/runtime/scheduler.hpp index d3461d35..e46545ad 100644 --- a/include/elio/runtime/scheduler.hpp +++ b/include/elio/runtime/scheduler.hpp @@ -1787,11 +1787,20 @@ inline void detail::external_worker_wakeup::schedule( std::coroutine_handle<> handle) noexcept { std::unique_lock lock(mutex); auto* promise = coro::get_promise_base(handle.address()); - if (!promise || promise->pinned_continuation_owner().get() != this) { + auto current_owner = promise ? promise->pinned_continuation_owner() : nullptr; + if (current_owner.get() != this) { // The final lease may have been released after the notifier captured - // this endpoint. Do not drop an otherwise movable continuation. - lock.unlock(); - schedule_handle(handle); + // this endpoint. Retain the original scheduler domain, never the + // notifier's TLS/trampoline (which may be a blocking-pool thread). + if (current_owner) { + lock.unlock(); + current_owner->schedule(handle); + return; + } + while (domain && domain->is_running()) { + if (domain->try_schedule(handle)) return; + std::this_thread::yield(); + } return; } if (!worker) return; diff --git a/include/elio/runtime/spawn_blocking.hpp b/include/elio/runtime/spawn_blocking.hpp index 77efe12b..bd69f02a 100644 --- a/include/elio/runtime/spawn_blocking.hpp +++ b/include/elio/runtime/spawn_blocking.hpp @@ -16,6 +16,11 @@ namespace elio { namespace detail { +#ifdef ELIO_RUNTIME_TEST_HOOKS +inline std::atomic pause_blocking_owner_dispatch_for_test{false}; +inline std::atomic blocking_owner_dispatch_paused_for_test{false}; +#endif + class blocking_admission_error : public std::runtime_error { public: blocking_admission_error() @@ -169,6 +174,15 @@ class blocking_awaitable { auto* promise = coro::get_promise_base(caller.address()); auto owner = promise ? promise->pinned_continuation_owner() : nullptr; if (owner) { +#ifdef ELIO_RUNTIME_TEST_HOOKS + if (pause_blocking_owner_dispatch_for_test.load(std::memory_order_acquire)) { + blocking_owner_dispatch_paused_for_test.store(true, std::memory_order_release); + while (pause_blocking_owner_dispatch_for_test.load(std::memory_order_acquire)) { + pause_blocking_owner_dispatch_for_test.wait(true, std::memory_order_acquire); + } + blocking_owner_dispatch_paused_for_test.store(false, std::memory_order_release); + } +#endif owner->schedule(caller); } else if (sched) { if (sched->is_running()) { diff --git a/include/elio/runtime/worker_thread.hpp b/include/elio/runtime/worker_thread.hpp index 92d225f7..9cb17d0b 100644 --- a/include/elio/runtime/worker_thread.hpp +++ b/include/elio/runtime/worker_thread.hpp @@ -27,6 +27,7 @@ struct external_worker_wakeup final : coro::detail::continuation_owner { mutable std::mutex mutex; io::io_context* context = nullptr; worker_thread* worker = nullptr; + scheduler* domain = nullptr; }; // Allocated with the operation, not with its cancellation request. Only the @@ -89,6 +90,7 @@ class worker_thread { { external_wakeup_->context = io_context_.get(); external_wakeup_->worker = this; + external_wakeup_->domain = scheduler_; } ~worker_thread() { @@ -96,6 +98,7 @@ class worker_thread { std::lock_guard lock(external_wakeup_->mutex); external_wakeup_->context = nullptr; external_wakeup_->worker = nullptr; + external_wakeup_->domain = nullptr; } worker_thread(const worker_thread&) = delete; diff --git a/tests/unit/test_worker_lease.cpp b/tests/unit/test_worker_lease.cpp index f450fd94..41a5b5a1 100644 --- a/tests/unit/test_worker_lease.cpp +++ b/tests/unit/test_worker_lease.cpp @@ -249,6 +249,16 @@ coro::generator gated_producer(sync::event* gate, bool yield_value) { if (yield_value) co_yield 42; } +coro::generator leased_producer(std::atomic* correct) { + auto lease = co_await lease_current_worker(); + const auto owner = std::this_thread::get_id(); + co_yield 1; + correct->store(std::this_thread::get_id() == owner); + co_yield 2; + correct->store(correct->load() && std::this_thread::get_id() == owner); + lease.release(); +} + coro::task acquire_without_worker(bool* rejected) { try { auto lease = co_await lease_current_worker(); @@ -438,13 +448,14 @@ TEST_CASE("isolated scope returns a leased parent to its owner", "[worker_lease] } TEST_CASE("captured owner endpoint falls back after final lease release", "[worker_lease][scheduler]") { + const bool foreign_dispatch = GENERATE(false, true); backend_selection backend(io::io_context::backend_type::epoll); scheduler sched(2); auto* old_owner = sched.get_worker(1); std::optional retained; std::shared_ptr endpoint; std::atomic suspended{nullptr}; - std::atomic finished{false}; + std::atomic finished{false}, correct_domain{false}; struct manual_wait { std::atomic* address; bool await_ready() const noexcept { return false; } @@ -458,6 +469,7 @@ TEST_CASE("captured owner endpoint falls back after final lease release", "[work retained.emplace(co_await lease_current_worker()); endpoint = coro::promise_base::current_frame()->pinned_continuation_owner(); co_await manual_wait{&suspended}; + correct_domain.store(scheduler::current() == &sched && worker_thread::current() != nullptr); finished.store(true, std::memory_order_release); }); const bool parked = eventually([&] { return suspended.load(std::memory_order_acquire) != nullptr; }); @@ -466,13 +478,46 @@ TEST_CASE("captured owner endpoint falls back after final lease release", "[work const bool retired = eventually([&] { return !old_owner->is_running(); }); // Models capture -> final release -> retire -> notifier dispatch, without // relying on timing to land between lookup and endpoint invocation. - if (parked) endpoint->schedule(std::coroutine_handle<>::from_address(suspended.load())); + if (parked) { + auto dispatch = [&] { + endpoint->schedule(std::coroutine_handle<>::from_address(suspended.load())); + }; + if (foreign_dispatch) { std::thread thread(dispatch); thread.join(); } + else dispatch(); + } const bool completed = eventually([&] { return finished.load(std::memory_order_acquire); }); if (completed) sched.shutdown(); else sched.shutdown_force(); REQUIRE(parked); REQUIRE(retired); REQUIRE(completed); + REQUIRE(correct_domain.load()); +} + +TEST_CASE("next restores a leased producer independently of its leased consumer", "[worker_lease][scheduler]") { + backend_selection backend(io::io_context::backend_type::epoll); + scheduler sched(2); + std::atomic producer_correct{true}, finished{false}, consumer_correct{false}; + auto producer = leased_producer(&producer_correct); + sched.start(); + sched.go_to(0, [&]() -> coro::task { + (void)co_await producer.next(); + co_await runtime::set_affinity(1); + auto consumer_lease = co_await lease_current_worker(); + const auto consumer_thread = std::this_thread::get_id(); + (void)co_await producer.next(); + const bool returned = std::this_thread::get_id() == consumer_thread; + (void)co_await producer.next(); + consumer_correct.store(returned && std::this_thread::get_id() == consumer_thread); + consumer_lease.release(); + finished.store(true, std::memory_order_release); + }); + const bool completed = eventually([&] { return finished.load(std::memory_order_acquire); }); + if (completed) sched.shutdown(); + else sched.shutdown_force(); + REQUIRE(completed); + REQUIRE(consumer_correct.load()); + REQUIRE(producer_correct.load()); } TEST_CASE("generator yield and exhaustion restore a leased consumer", "[worker_lease][scheduler]") { @@ -576,3 +621,35 @@ TEST_CASE("blocking and timer completions retain leased owners on rejection", "[ REQUIRE(completed); REQUIRE(correct.load()); } + +TEST_CASE("blocking completion keeps its scheduler after captured lease release", "[worker_lease][scheduler]") { + backend_selection backend(io::io_context::backend_type::epoll); + scheduler sched(2); + auto* old_owner = sched.get_worker(1); + std::optional retained; + std::atomic finished{false}, correct_domain{false}; + elio::detail::blocking_owner_dispatch_paused_for_test.store(false); + elio::detail::pause_blocking_owner_dispatch_for_test.store(true); + sched.start(); + sched.go_to(1, [&]() -> coro::task { + retained.emplace(co_await lease_current_worker()); + co_await elio::spawn_blocking([] {}); + correct_domain.store(scheduler::current() == &sched && worker_thread::current() != nullptr); + finished.store(true, std::memory_order_release); + }); + const bool captured = eventually([] { + return elio::detail::blocking_owner_dispatch_paused_for_test.load(std::memory_order_acquire); + }); + sched.set_thread_count(1); + if (captured) retained->release(); + const bool retired = eventually([&] { return !old_owner->is_running(); }); + elio::detail::pause_blocking_owner_dispatch_for_test.store(false, std::memory_order_release); + elio::detail::pause_blocking_owner_dispatch_for_test.notify_all(); + const bool completed = eventually([&] { return finished.load(std::memory_order_acquire); }); + if (completed) sched.shutdown(); + else sched.shutdown_force(); + REQUIRE(captured); + REQUIRE(retired); + REQUIRE(completed); + REQUIRE(correct_domain.load()); +} diff --git a/wiki/Worker-Leases.md b/wiki/Worker-Leases.md index 34d5b86e..8c49ed2e 100644 --- a/wiki/Worker-Leases.md +++ b/wiki/Worker-Leases.md @@ -20,6 +20,10 @@ the parent's thread-affine resources from a scope body that migrates elsewhere. Async generators likewise keep an independent producer context, but yield and exhaustion restore a leased consumer to its owner. `for_each()`'s internal loop shares its Elio caller's context so callbacks and loop return honor that lease. +If the producer itself holds a lease across yields, `next()` restores the +producer's owner before executing its body. Settle/exhaust such a producer on +its owner before destroying its frame: a lease does not make early generator +destruction asynchronous or move thread-affine resource destructors to an owner. An application-provided awaitable that directly resumes on a foreign executor must arrange its own owner handoff before thread-affine code runs.