From 758dd1b8204263e9882a02fb7a3fd3e1206e83a6 Mon Sep 17 00:00:00 2001 From: Steve Gerbino Date: Mon, 19 Oct 2026 21:02:53 +0200 Subject: [PATCH 1/4] refactor: create the timer service with the first timer Each scheduler created the timer service in its own constructor, so the timer service was older than the scheduler. It was therefore shut down after the scheduler's drain, and every scheduler shut it down early by hand so its waiters were drained too. The timer service is now created with a context's first timer, like every other service, and attaches itself to the scheduler. Being newer than the scheduler, it shuts down before the drain without help, and the early shutdown is gone. A context that never uses timers creates no timer service, and on IOCP starts no timer thread. The scheduler holds the attached service in an atomic pointer, since the first timer can be made on another thread than the one running the loop, and treats a null one as having no timers. Only the service the context keeps is attached: a concurrent first use can construct a second one that the context discards. --- doc/design/lifetime-contract.md | 12 ++-- include/boost/corosio/detail/scheduler.hpp | 51 ++++++++++++++++ include/boost/corosio/detail/timer.hpp | 3 + .../boost/corosio/detail/timer_service.hpp | 59 +++++++++++++++---- .../native/detail/epoll/epoll_scheduler.hpp | 29 +++++---- .../native/detail/iocp/win_scheduler.hpp | 53 +++++------------ .../native/detail/kqueue/kqueue_scheduler.hpp | 16 +++-- .../detail/reactor/reactor_scheduler.hpp | 19 +++--- .../native/detail/select/select_scheduler.hpp | 16 +++-- .../native/detail/uring/uring_scheduler.hpp | 39 ++++++------ src/corosio/src/timer.cpp | 17 +++++- test/unit/delay.cpp | 25 ++++++++ test/unit/fault/iocp_faults.cpp | 3 + test/unit/lazy_services.cpp | 53 ++++++++++++++++- 14 files changed, 277 insertions(+), 118 deletions(-) diff --git a/doc/design/lifetime-contract.md b/doc/design/lifetime-contract.md index de8958611..0f754c71e 100644 --- a/doc/design/lifetime-contract.md +++ b/doc/design/lifetime-contract.md @@ -181,17 +181,15 @@ reach an unrelated file: ## 6. Teardown `io_context` destruction runs every service's `shutdown()` newest -first, then deletes the services in the same order. Services created -after the scheduler come before it; the timer service, which the -scheduler's constructor creates, comes after it. Every service gives -up all its work in `shutdown()` and posts nothing from its destructor. +first, then deletes the services in the same order. The scheduler is +created with the context, and every other service after it, so every +service shuts down before the scheduler drains. Every service gives up +all its work in `shutdown()` and posts nothing from its destructor. So nothing reaches the queue after the drain in step 3. Waiting for the kernel or the OS to return operations a service submitted, as in step 2, is part of giving up its work. The contract for that sequence: -0. **Timers first.** Every scheduler shuts the timer service down - before it drains, so work a timer's teardown posts is drained too. 1. **Cancel.** Each service cancels or closes every live implementation. Its pool's `shutdown()` pins each one with `try_acquire` and switches the pool to delete-on-zero. @@ -260,5 +258,5 @@ How the code meets each rule: | §5.3 IOCP | Conforms: socket and acceptor initiators re-check cancellation before `on_pending`, as the handle initiators do. IOCP is checked only by Windows CI. | | §5.5 descriptor numbers | Conforms: posix file gate, uring flush before close or release, cancelled waits skip the probe on uring and the IOCP wait reactor. | | §5.4 executor queues | Conforms: every awaitable owns its continuation; no op embeds one. | -| §6 teardown | Conforms on all backends, timers first everywhere; implemented per backend with no shared statement. Every service gives up its work in `shutdown()` and posts nothing from its destructor, so every frame is destroyed before any service (`testShutdownDestroysFramesBeforeServices` in `teardown_destroy.cpp`). | +| §6 teardown | Conforms on all backends. Implemented per backend with no shared statement. Every service gives up its work in `shutdown()` and posts nothing from its destructor, so every frame is destroyed before any service (`testShutdownDestroysFramesBeforeServices` in `teardown_destroy.cpp`). | | §7 recycling | Mixed. Some `reuse()` methods assert fields others reset. | diff --git a/include/boost/corosio/detail/scheduler.hpp b/include/boost/corosio/detail/scheduler.hpp index 2c28a006b..c0df2821f 100644 --- a/include/boost/corosio/detail/scheduler.hpp +++ b/include/boost/corosio/detail/scheduler.hpp @@ -28,6 +28,7 @@ namespace boost::corosio::detail { class scheduler_op; +class timer_service; /** Define the abstract interface for the event loop scheduler. @@ -156,7 +157,56 @@ class BOOST_COROSIO_DECL scheduler cancels_posting_.fetch_sub(1, std::memory_order_release); } + /** Attach the context's timer service. + + The timer service is created with the context's first timer, + possibly on a thread other than the one running the loop; until + it is attached the loop has no timers to wait on. + + @par Preconditions + No timer service is attached. + + @par Exception Safety + Strong guarantee: if the scheduler cannot start waiting on + timers, nothing is attached. + + @throws std::system_error if the scheduler cannot start waiting + on timers. + + @param svc The service to attach. + */ + void attach_timer_service(timer_service& svc) + { + on_timer_service_attached(svc); + timer_svc_.store(&svc, std::memory_order_release); + } + + /// Detach the timer service as it is destroyed. + void detach_timer_service() noexcept + { + timer_svc_.store(nullptr, std::memory_order_release); + } + protected: + /// Return the attached timer service, or null before the first timer. + timer_service* timers() const noexcept + { + return timer_svc_.load(std::memory_order_acquire); + } + + /** Prepare to wait on a timer service about to be attached. + + Called before the service is published to the loop, so the loop + never sees a service it cannot yet wait on. + + @par Exception Safety + Strong guarantee: a throw leaves the scheduler as it was, and + the next timer calls this again. + + @param svc The service being attached. + */ + virtual void on_timer_service_attached(timer_service& svc) = 0; + /** Wait out the cancellers counted by @ref cancel_posting_begin. Called before the shutdown drain, so it finds their posts; the @@ -173,6 +223,7 @@ class BOOST_COROSIO_DECL scheduler BOOST_COROSIO_MSVC_WARNING_PUSH BOOST_COROSIO_MSVC_WARNING_DISABLE(4251) std::atomic cancels_posting_{0}; + std::atomic timer_svc_{nullptr}; BOOST_COROSIO_MSVC_WARNING_POP }; diff --git a/include/boost/corosio/detail/timer.hpp b/include/boost/corosio/detail/timer.hpp index 9add53dba..a92d9ceb2 100644 --- a/include/boost/corosio/detail/timer.hpp +++ b/include/boost/corosio/detail/timer.hpp @@ -219,6 +219,9 @@ class BOOST_COROSIO_DECL timer : public io_object throws (a timer service is required). @throws std::logic_error if @p ctx is not an io_context. + + @throws std::system_error if this is the context's first timer + and the context cannot start waiting on timers. */ explicit timer(capy::execution_context& ctx); diff --git a/include/boost/corosio/detail/timer_service.hpp b/include/boost/corosio/detail/timer_service.hpp index 846769ce7..45c54034a 100644 --- a/include/boost/corosio/detail/timer_service.hpp +++ b/include/boost/corosio/detail/timer_service.hpp @@ -156,6 +156,9 @@ class BOOST_COROSIO_DECL timer_service final timer_object_pool pool_; callback on_earliest_changed_; bool shutting_down_ = false; + // Set once the scheduler waits on this service; read without the + // mutex by every timer construction after the first. + std::atomic attached_{false}; // Cancellers that claimed a waiter and have yet to post it; // shutdown waits them out so the scheduler's drain finds the post. @@ -166,10 +169,21 @@ class BOOST_COROSIO_DECL timer_service final BOOST_COROSIO_MSVC_WARNING_POP public: - /// Construct the timer service bound to a scheduler. - inline timer_service(capy::execution_context&, scheduler& sched) - : sched_(&sched) + /** Construct the timer service for a context's scheduler. + + Created with the context's first timer. It does not attach to + the scheduler here: a concurrent first use may construct a + second service that the context then discards. See + @ref attach. + + @throws std::logic_error if the context has no scheduler. + */ + inline explicit timer_service(capy::execution_context& ctx) + : sched_(ctx.find_service()) { + if (!sched_) + detail::throw_logic_error( + "timer_service: the context has no scheduler"); } /// Return the associated scheduler. @@ -178,8 +192,35 @@ class BOOST_COROSIO_DECL timer_service final return *sched_; } - /// Destroy the timer service. - ~timer_service() override = default; + /// Destroy the timer service, detaching it from the scheduler. + ~timer_service() override + { + if (attached_.load(std::memory_order_acquire)) + sched_->detach_timer_service(); + } + + /** Attach this service to the scheduler, if not yet attached. + + Call once the context has settled which service it keeps; no + timer of this service can wait before this returns. + + @par Exception Safety + Strong guarantee: on failure the service stays detached and + the next call retries. + + @throws std::system_error if the scheduler cannot start waiting + on timers. + */ + inline void attach() + { + if (attached_.load(std::memory_order_acquire)) + return; + std::lock_guard lock(mutex_); + if (attached_.load(std::memory_order_relaxed)) + return; + sched_->attach_timer_service(*this); + attached_.store(true, std::memory_order_release); + } timer_service(timer_service const&) = delete; timer_service& operator=(timer_service const&) = delete; @@ -699,14 +740,6 @@ timer_service::swap_heap(std::size_t i1, std::size_t i2) // in timer.hpp (e.g. code that only reaches timer.hpp through // delay.hpp, without transitively including a scheduler header). -// Free functions - -inline timer_service& -get_timer_service(capy::execution_context& ctx, scheduler& sched) -{ - return ctx.make_service(sched); -} - } // namespace boost::corosio::detail #endif diff --git a/include/boost/corosio/native/detail/epoll/epoll_scheduler.hpp b/include/boost/corosio/native/detail/epoll/epoll_scheduler.hpp index b180357c4..77e756975 100644 --- a/include/boost/corosio/native/detail/epoll/epoll_scheduler.hpp +++ b/include/boost/corosio/native/detail/epoll/epoll_scheduler.hpp @@ -139,6 +139,7 @@ class BOOST_COROSIO_DECL epoll_scheduler final : public reactor_scheduler void run_task(lock_type& lock, context_type& ctx, long timeout_us) override; void interrupt_reactor() const override; void update_timerfd() const; + void on_timer_service_attached(timer_service& svc) override; int epoll_fd_; int event_fd_; @@ -159,7 +160,7 @@ class BOOST_COROSIO_DECL epoll_scheduler final : public reactor_scheduler std::vector event_buffer_; }; -inline epoll_scheduler::epoll_scheduler(capy::execution_context& ctx, int) +inline epoll_scheduler::epoll_scheduler(capy::execution_context&, int) : epoll_fd_(-1) , event_fd_(-1) , timer_fd_(-1) @@ -210,17 +211,19 @@ inline epoll_scheduler::epoll_scheduler(capy::execution_context& ctx, int) detail::throw_system_error(make_err(errn), "epoll_ctl (timerfd)"); } - timer_svc_ = &get_timer_service(ctx, *this); - timer_svc_->set_on_earliest_changed( - timer_service::callback(this, [](void* p) { - auto* self = static_cast(p); - self->timerfd_stale_.store(true, std::memory_order_release); - self->interrupt_reactor(); - })); - completed_ops_.push(&task_op_); } +inline void +epoll_scheduler::on_timer_service_attached(timer_service& svc) +{ + svc.set_on_earliest_changed(timer_service::callback(this, [](void* p) { + auto* self = static_cast(p); + self->timerfd_stale_.store(true, std::memory_order_release); + self->interrupt_reactor(); + })); +} + inline epoll_scheduler::~epoll_scheduler() { if (timer_fd_ >= 0) @@ -326,7 +329,10 @@ epoll_scheduler::interrupt_reactor() const inline void epoll_scheduler::update_timerfd() const { - auto nearest = timer_svc_->nearest_expiry(); + auto* tsvc = timers(); + if (!tsvc) + return; + auto nearest = tsvc->nearest_expiry(); itimerspec ts{}; int flags = 0; @@ -444,7 +450,8 @@ epoll_scheduler::run_task(lock_type& lock, context_type& ctx, long timeout_us) if (check_timers) { - timer_svc_->process_expired(); + if (auto* tsvc = timers()) + tsvc->process_expired(); update_timerfd(); } diff --git a/include/boost/corosio/native/detail/iocp/win_scheduler.hpp b/include/boost/corosio/native/detail/iocp/win_scheduler.hpp index 3139b539e..4b4529b5c 100644 --- a/include/boost/corosio/native/detail/iocp/win_scheduler.hpp +++ b/include/boost/corosio/native/detail/iocp/win_scheduler.hpp @@ -109,10 +109,11 @@ class BOOST_COROSIO_DECL win_scheduler final void on_completion(overlapped_op* op, DWORD error, DWORD bytes) const; // Timer service integration - void set_timer_service(timer_service* svc); void update_timeout(); private: + void on_timer_service_attached(timer_service& svc) override; + /// Destroy every packet and deferred entry still queued, without /// running them; stops once the port is empty. void sweep_port() noexcept; @@ -125,7 +126,6 @@ class BOOST_COROSIO_DECL win_scheduler final void post_deferred_completions(op_queue& ops); std::size_t do_one(unsigned long timeout_ms); - timer_service* timer_svc_ = nullptr; void* iocp_; mutable long outstanding_work_; @@ -538,8 +538,8 @@ win_scheduler::do_one(unsigned long timeout_ms) } post_deferred_completions(local_ops); - if (timer_svc_) - timer_svc_->process_expired(); + if (auto* tsvc = timers()) + tsvc->process_expired(); update_timeout(); } @@ -664,11 +664,10 @@ win_scheduler::on_timer_changed(void* ctx) } inline void -win_scheduler::set_timer_service(timer_service* svc) +win_scheduler::on_timer_service_attached(timer_service& svc) { - timer_svc_ = svc; // Pass 'this' as context - callback routes to correct instance - svc->set_on_earliest_changed( + svc.set_on_earliest_changed( timer_service::callback{this, &on_timer_changed}); if (timers_) timers_->start(); @@ -677,8 +676,9 @@ win_scheduler::set_timer_service(timer_service* svc) inline void win_scheduler::update_timeout() { - if (timer_svc_ && timers_) - timers_->update_timeout(timer_svc_->nearest_expiry()); + auto* tsvc = timers(); + if (tsvc && timers_) + timers_->update_timeout(tsvc->nearest_expiry()); } } // namespace boost::corosio::detail @@ -697,7 +697,7 @@ win_scheduler::update_timeout() namespace boost::corosio::detail { inline win_scheduler::win_scheduler( - capy::execution_context& ctx, int concurrency_hint) + capy::execution_context&, int concurrency_hint) : iocp_(nullptr) , outstanding_work_(0) , stopped_(0) @@ -716,7 +716,6 @@ inline win_scheduler::win_scheduler( try { timers_ = make_win_timers(iocp_, &dispatch_required_); - set_timer_service(&get_timer_service(ctx, *this)); // A scheduler whose wait reactor could not be built would // answer every wait with a parked op, so it refuses to exist @@ -728,21 +727,7 @@ inline win_scheduler::win_scheduler( catch (...) { // ~win_scheduler never runs for a constructor that throws, and - // the port is a raw handle nothing else owns. The timer thread - // is stopped first because it posts to that port. - // - // The services registered above are not unregistered here, and - // cannot be: the context owns them and offers no way to take - // one back. Each holds a scheduler reference this unwind is - // about to invalidate, and each survives to be shut down and - // destroyed with the context. What makes that safe is that - // none of them touches its scheduler while it holds nothing: - // a scheduler that never finished constructing handed out no - // timer, resolver or file, so every one of those shutdowns - // walks an empty list. Registering anything here that would - // reach back into the scheduler on an empty shutdown breaks - // that, and the fix would have to be a rollback in the - // context, not an ordering trick here. + // the port is a raw handle nothing else owns. timers_.reset(); ::CloseHandle(iocp_); iocp_ = nullptr; @@ -756,19 +741,9 @@ win_scheduler::shutdown() if (timers_) timers_->stop(); - // Drain timer heap before the work-counting loop. The timer_service - // is created from our constructor, so it finishes, and is linked, - // first; execution_context::shutdown() runs newest first, so it - // calls us first. - // Asio avoids this by owning timer queues directly inside the - // scheduler; we bridge the gap by shutting down the timer service - // early. The subsequent call from execution_context is a no-op. - if (timer_svc_) - timer_svc_->shutdown(); - - // Same problem for the auxiliary wait reactor: ops parked in it - // owe completion packets. Stop the reactor early so its loop - // posts them as cancelled and the pending count can reach zero. + // Ops parked in the auxiliary wait reactor owe completion packets. + // Stop the reactor early so its loop posts them as cancelled and + // the pending count can reach zero. wait_reactor_->stop(); wait_cancels_posting(); diff --git a/include/boost/corosio/native/detail/kqueue/kqueue_scheduler.hpp b/include/boost/corosio/native/detail/kqueue/kqueue_scheduler.hpp index a612ef85a..2414617cb 100644 --- a/include/boost/corosio/native/detail/kqueue/kqueue_scheduler.hpp +++ b/include/boost/corosio/native/detail/kqueue/kqueue_scheduler.hpp @@ -249,7 +249,7 @@ class BOOST_COROSIO_DECL kqueue_scheduler final : public reactor_scheduler std::vector event_buffer_; }; -inline kqueue_scheduler::kqueue_scheduler(capy::execution_context& ctx, int) +inline kqueue_scheduler::kqueue_scheduler(capy::execution_context&, int) : kq_fd_(-1) , event_buffer_(max_events_per_poll_) { @@ -273,12 +273,6 @@ inline kqueue_scheduler::kqueue_scheduler(capy::execution_context& ctx, int) detail::throw_system_error(make_err(errn), "kevent (EVFILT_USER)"); } - timer_svc_ = &get_timer_service(ctx, *this); - timer_svc_->set_on_earliest_changed( - timer_service::callback(this, [](void* p) { - static_cast(p)->interrupt_reactor(); - })); - completed_ops_.push(&task_op_); } @@ -417,7 +411,10 @@ kqueue_scheduler::calculate_timeout(long requested_timeout_us) const if (requested_timeout_us == 0) return 0; - auto nearest = timer_svc_->nearest_expiry(); + auto* tsvc = timers(); + if (!tsvc) + return requested_timeout_us; + auto nearest = tsvc->nearest_expiry(); if (nearest == timer_service::time_point::max()) return requested_timeout_us; @@ -521,7 +518,8 @@ kqueue_scheduler::run_task(lock_type& lock, context_type& ctx, long timeout_us) } } - timer_svc_->process_expired(); + if (auto* tsvc = timers()) + tsvc->process_expired(); lock.lock(); diff --git a/include/boost/corosio/native/detail/reactor/reactor_scheduler.hpp b/include/boost/corosio/native/detail/reactor/reactor_scheduler.hpp index a32f0a204..957ce7fef 100644 --- a/include/boost/corosio/native/detail/reactor/reactor_scheduler.hpp +++ b/include/boost/corosio/native/detail/reactor/reactor_scheduler.hpp @@ -295,7 +295,6 @@ class reactor_scheduler } protected: - timer_service* timer_svc_ = nullptr; bool scheduler_locking_disabled_ = false; bool reactor_io_locking_ = true; bool one_thread_ = false; @@ -381,6 +380,9 @@ class reactor_scheduler /// Wake a blocked reactor (e.g. write to eventfd or pipe). virtual void interrupt_reactor() const = 0; + /// Wake the reactor whenever the earliest timer changes. + void on_timer_service_attached(timer_service& svc) override; + private: struct work_cleanup { @@ -823,15 +825,16 @@ reactor_scheduler::settle_retired(lock_type& lock) const } inline void -reactor_scheduler::shutdown_drain() +reactor_scheduler::on_timer_service_attached(timer_service& svc) { - // This scheduler's constructor registered the timer service, so - // the context shuts it down later; its waiters' frames must be - // destroyed before the drain, which then destroys whatever their - // destructors post. The context's own call later finds no waiter. - if (timer_svc_) - timer_svc_->shutdown(); + svc.set_on_earliest_changed(timer_service::callback(this, [](void* p) { + static_cast(p)->interrupt_reactor(); + })); +} +inline void +reactor_scheduler::shutdown_drain() +{ wait_cancels_posting(); lock_type lock(mutex_); diff --git a/include/boost/corosio/native/detail/select/select_scheduler.hpp b/include/boost/corosio/native/detail/select/select_scheduler.hpp index b16207fb2..b65449b3d 100644 --- a/include/boost/corosio/native/detail/select/select_scheduler.hpp +++ b/include/boost/corosio/native/detail/select/select_scheduler.hpp @@ -164,7 +164,7 @@ class BOOST_COROSIO_DECL select_scheduler final : public reactor_scheduler mutable int max_fd_ = -1; }; -inline select_scheduler::select_scheduler(capy::execution_context& ctx, int) +inline select_scheduler::select_scheduler(capy::execution_context&, int) : pipe_fds_{-1, -1} , max_fd_(-1) { @@ -197,12 +197,6 @@ inline select_scheduler::select_scheduler(capy::execution_context& ctx, int) } } - timer_svc_ = &get_timer_service(ctx, *this); - timer_svc_->set_on_earliest_changed( - timer_service::callback(this, [](void* p) { - static_cast(p)->interrupt_reactor(); - })); - completed_ops_.push(&task_op_); } @@ -307,7 +301,10 @@ select_scheduler::calculate_timeout(long requested_timeout_us) const if (requested_timeout_us == 0) return 0; // LCOV_EXCL_LINE run_task passes 0 via task_interrupted_, never through this argument - auto nearest = timer_svc_->nearest_expiry(); + auto* tsvc = timers(); + if (!tsvc) + return requested_timeout_us; + auto nearest = tsvc->nearest_expiry(); if (nearest == timer_service::time_point::max()) return requested_timeout_us; @@ -433,7 +430,8 @@ select_scheduler::run_task(lock_type& lock, context_type& ctx, long timeout_us) } // Process timers outside the lock - timer_svc_->process_expired(); + if (auto* tsvc = timers()) + tsvc->process_expired(); ready_queue local_ops; diff --git a/include/boost/corosio/native/detail/uring/uring_scheduler.hpp b/include/boost/corosio/native/detail/uring/uring_scheduler.hpp index 81ce3e3c5..ae4dae5ba 100644 --- a/include/boost/corosio/native/detail/uring/uring_scheduler.hpp +++ b/include/boost/corosio/native/detail/uring/uring_scheduler.hpp @@ -416,6 +416,8 @@ class BOOST_COROSIO_DECL uring_scheduler final } private: + void on_timer_service_attached(timer_service& svc) override; + /// Destroy every queued operation without running it. void destroy_queued(lock_type& lock) noexcept; @@ -423,7 +425,6 @@ class BOOST_COROSIO_DECL uring_scheduler final // from const contexts like post()) can populate them on first use. mutable struct ::io_uring ring_{}; mutable int wakeup_eventfd_ = -1; - timer_service* timer_svc_ = nullptr; // dispatch_mutex_ protects completed_ops_, cond_, task_running_. // ring_mutex_ protects every userspace touch of ring_ (SQ tail, @@ -559,20 +560,12 @@ class BOOST_COROSIO_DECL uring_scheduler final }; inline uring_scheduler::uring_scheduler( - capy::execution_context& ctx, int /*concurrency_hint*/) + capy::execution_context&, int /*concurrency_hint*/) { // sched_ cannot be set in the member initialiser — `this` is not // available there. submit_op_.sched_ = this; - // Wire timer service. on_earliest_changed wakes the run loop so it - // recomputes its wait timeout. - timer_svc_ = &get_timer_service(ctx, *this); - timer_svc_->set_on_earliest_changed( - timer_service::callback(this, [](void* p) { - static_cast(p)->interrupt_reactor(); - })); - // Ring init is deferred so the options the io_context applies // after this constructor — the locking tier and SQPOLL — can feed // the flags io_uring_queue_init_params is given. The io_context @@ -695,10 +688,6 @@ uring_scheduler::shutdown() { stopped_.store(true, std::memory_order_release); - // See reactor_scheduler::shutdown_drain(). - if (timer_svc_) - timer_svc_->shutdown(); - // Cancel every request still in the kernel and wait for each to // come back: the services free their impls next, and an op still // armed would write into freed memory. One cancel-any covers ops @@ -848,6 +837,14 @@ uring_scheduler::interrupt_reactor() const noexcept std::ignore = ::write(wakeup_eventfd_, &v, sizeof(v)); } +inline void +uring_scheduler::on_timer_service_attached(timer_service& svc) +{ + svc.set_on_earliest_changed(timer_service::callback(this, [](void* p) { + static_cast(p)->interrupt_reactor(); + })); +} + inline void uring_scheduler::drain_wakeup_eventfd() const noexcept { @@ -1218,8 +1215,8 @@ uring_scheduler::do_one(long timeout_us) // (~25 pp on io_context:single_threaded). When a timer IS // registered the call runs exactly as before, preserving the // deadlock fix this guard was originally written to address. - if (!timer_svc_->empty()) - timer_svc_->process_expired(); + if (auto* tsvc = timers(); tsvc && !tsvc->empty()) + tsvc->process_expired(); lock_type lock(dispatch_mutex_); for (;;) @@ -1280,8 +1277,10 @@ uring_scheduler::do_one(long timeout_us) // work. __kernel_timespec ts{}; __kernel_timespec* ts_ptr = nullptr; - auto next_expiry = timer_svc_->nearest_expiry(); - auto now = std::chrono::steady_clock::now(); + auto* tsvc = timers(); + auto next_expiry = + tsvc ? tsvc->nearest_expiry() : timer_service::time_point::max(); + auto now = std::chrono::steady_clock::now(); if (timeout_us == 0) { @@ -1368,8 +1367,8 @@ uring_scheduler::do_one(long timeout_us) make_err(-rc), "io_uring_wait_cqe_timeout"); } - if (!timer_svc_->empty()) - timer_svc_->process_expired(); + if (tsvc && !tsvc->empty()) + tsvc->process_expired(); lock.lock(); task_running_ = false; diff --git a/src/corosio/src/timer.cpp b/src/corosio/src/timer.cpp index acdbbcb50..200eab7c7 100644 --- a/src/corosio/src/timer.cpp +++ b/src/corosio/src/timer.cpp @@ -16,8 +16,23 @@ namespace boost::corosio::detail { timer::~timer() = default; +namespace { + +// The context's first timer creates the timer service. Only the service +// the context keeps is attached; a concurrent first use may construct +// one more, which the context discards. +timer_service& +attached_timer_service(capy::execution_context& ctx) +{ + auto& svc = ctx.use_service(); + svc.attach(); + return svc; +} + +} // namespace + timer::timer(capy::execution_context& ctx) - : io_object(create_handle(ctx)) + : io_object(handle(ctx, attached_timer_service(ctx))) { } diff --git a/test/unit/delay.cpp b/test/unit/delay.cpp index f18a77887..13eb72094 100644 --- a/test/unit/delay.cpp +++ b/test/unit/delay.cpp @@ -208,6 +208,30 @@ struct delay_test BOOST_TEST(ok); } + // The first timer of a context may be made on another thread while + // run() waits for work; run() still wakes when it expires. + void testFirstTimerMadeOnAnotherThread() + { + io_context ioc(Backend); + auto ex = ioc.get_executor(); + ex.on_work_started(); + bool fired = false; + std::thread runner([&] { ioc.run(); }); + { + detail::timer t(ioc); + t.expires_after(std::chrono::milliseconds(1)); + capy::run_async(ex)( + [](detail::timer& tm, bool& f, + io_context::executor_type e) -> capy::task<> { + auto [ec] = co_await tm.wait(); + f = !ec; + e.on_work_finished(); + }(t, fired, ex)); + runner.join(); + } + BOOST_TEST(fired); + } + void testCancellation() { io_context ioc(Backend); @@ -1153,6 +1177,7 @@ struct delay_test testZeroDurationCompletesImmediately(); testPastTimePointCompletesImmediately(); testCancellation(); + testFirstTimerMadeOnAnotherThread(); testAlreadyStoppedCompletesCanceled(); testZeroDurationWithStopRequested(); testPastTimePointWithStopRequested(); diff --git a/test/unit/fault/iocp_faults.cpp b/test/unit/fault/iocp_faults.cpp index 2bf82ee3b..023605bb4 100644 --- a/test/unit/fault/iocp_faults.cpp +++ b/test/unit/fault/iocp_faults.cpp @@ -15,6 +15,7 @@ #include "test_utils.hpp" #include +#include #include #include #include @@ -236,6 +237,8 @@ struct iocp_faults sys::WaitForSingleObject, ERROR_INVALID_HANDLE, 1, any_thread); { io_context ioc(iocp); + // The first timer starts the timer thread. + corosio::detail::timer t(ioc); for (int i = 0; i < 2000 && !f.fired(); ++i) std::this_thread::sleep_for(std::chrono::milliseconds(1)); } diff --git a/test/unit/lazy_services.cpp b/test/unit/lazy_services.cpp index df84525e4..e6595703d 100644 --- a/test/unit/lazy_services.cpp +++ b/test/unit/lazy_services.cpp @@ -31,7 +31,10 @@ #include #include +#include #include +#include +#include #if BOOST_COROSIO_HAS_IOCP #include @@ -57,6 +60,9 @@ #include #endif +#include +#include +#include #include #include #include @@ -138,11 +144,54 @@ struct lazy_services_test ioc.template find_service() == nullptr); - // The scheduler's own infrastructure stays eager. + // Timers too: the first timer creates the timer service. + BOOST_TEST( + ioc.template find_service() == nullptr); + } + + void testTimerCreatesTimerService() + { + io_context ioc(Backend); + + detail::timer t(ioc); BOOST_TEST( ioc.template find_service() != nullptr); } + // The first timers may be made on several threads at once; one timer + // service is kept, and it drives the waits. + void testConcurrentFirstTimer() + { + io_context ioc(Backend); + + std::vector threads; + std::vector> timers(8); + // Spawning a thread takes longer than making a timer, so without + // a common start the timers would mostly be made one at a time. + std::latch start(static_cast(timers.size())); + for (std::size_t i = 0; i < timers.size(); ++i) + threads.emplace_back([&, i] { + start.arrive_and_wait(); + timers[i] = std::make_unique(ioc); + }); + for (auto& t : threads) + t.join(); + + int fired = 0; + for (auto& t : timers) + { + t->expires_after(std::chrono::milliseconds(1)); + capy::run_async(ioc.get_executor())( + [](detail::timer& tm, int& n) -> capy::task<> { + auto [ec] = co_await tm.wait(); + if (!ec) + ++n; + }(*t, fired)); + } + ioc.run(); + BOOST_TEST_EQ(fired, static_cast(timers.size())); + } + void testObjectCreatesOnlyItsOwnService() { io_context ioc(Backend); @@ -236,6 +285,8 @@ struct lazy_services_test testObjectCreatesOnlyItsOwnService(); testEachObjectCreatesItsService(); testConcurrentFirstUse(); + testTimerCreatesTimerService(); + testConcurrentFirstTimer(); } }; From 9a80f36473306f9b0998063e1f65c5792af5d499 Mon Sep 17 00:00:00 2001 From: Steve Gerbino Date: Tue, 20 Oct 2026 19:09:31 +0200 Subject: [PATCH 2/4] refactor: create epoll's timerfd with the first timer The epoll scheduler created and registered its timerfd in its constructor, so every context paid for a descriptor and two syscalls whether or not it used timers. It is now created when the timer service attaches, with the context's first timer. If creating or registering it fails, the epoll_wait timeout drives the timers instead: the first timer is often made inside a delay or timeout, where a throw could only terminate. --- .../native/detail/epoll/epoll_scheduler.hpp | 81 ++++++++++++------- test/unit/fault/epoll_faults.cpp | 34 +++++++- test/unit/lazy_services.cpp | 41 ++++++++++ 3 files changed, 124 insertions(+), 32 deletions(-) diff --git a/include/boost/corosio/native/detail/epoll/epoll_scheduler.hpp b/include/boost/corosio/native/detail/epoll/epoll_scheduler.hpp index 77e756975..5c103e93f 100644 --- a/include/boost/corosio/native/detail/epoll/epoll_scheduler.hpp +++ b/include/boost/corosio/native/detail/epoll/epoll_scheduler.hpp @@ -30,6 +30,7 @@ #include #include #include +#include #include #include @@ -65,8 +66,9 @@ class BOOST_COROSIO_DECL epoll_scheduler final : public reactor_scheduler public: /** Construct the scheduler. - Creates an epoll instance, eventfd for reactor interruption, - and timerfd for kernel-managed timer expiry. + Creates an epoll instance and an eventfd for reactor + interruption. The timerfd for timer expiry is created with the + context's first timer. @param ctx Reference to the owning execution_context. @param concurrency_hint Hint for expected thread count (unused). @@ -139,6 +141,7 @@ class BOOST_COROSIO_DECL epoll_scheduler final : public reactor_scheduler void run_task(lock_type& lock, context_type& ctx, long timeout_us) override; void interrupt_reactor() const override; void update_timerfd() const; + static int timer_timeout_ms(timer_service& svc, int timeout_ms) noexcept; void on_timer_service_attached(timer_service& svc) override; int epoll_fd_; @@ -178,45 +181,40 @@ inline epoll_scheduler::epoll_scheduler(capy::execution_context&, int) detail::throw_system_error(make_err(errn), "eventfd"); } - timer_fd_ = ::timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK | TFD_CLOEXEC); - if (timer_fd_ < 0) - { - int errn = errno; - ::close(event_fd_); - ::close(epoll_fd_); - detail::throw_system_error(make_err(errn), "timerfd_create"); - } - epoll_event ev{}; ev.events = EPOLLIN | EPOLLET; ev.data.ptr = nullptr; if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, event_fd_, &ev) < 0) { int errn = errno; - ::close(timer_fd_); ::close(event_fd_); ::close(epoll_fd_); detail::throw_system_error(make_err(errn), "epoll_ctl"); } - epoll_event timer_ev{}; - timer_ev.events = EPOLLIN | EPOLLERR; - timer_ev.data.ptr = &timer_fd_; - if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, timer_fd_, &timer_ev) < 0) - { - int errn = errno; - ::close(timer_fd_); - ::close(event_fd_); - ::close(epoll_fd_); - detail::throw_system_error(make_err(errn), "epoll_ctl (timerfd)"); - } - completed_ops_.push(&task_op_); } inline void epoll_scheduler::on_timer_service_attached(timer_service& svc) { + // Without a timerfd the epoll_wait timeout drives the timers: the + // first timer is often made where a throw can only terminate. + int fd = ::timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK | TFD_CLOEXEC); + if (fd >= 0) + { + // Set before registration, which orders it for every thread + // the timerfd's events can wake. + timer_fd_ = fd; + epoll_event timer_ev{}; + timer_ev.events = EPOLLIN | EPOLLERR; + timer_ev.data.ptr = &timer_fd_; + if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, fd, &timer_ev) < 0) + { + ::close(fd); + timer_fd_ = -1; + } + } svc.set_on_earliest_changed(timer_service::callback(this, [](void* p) { auto* self = static_cast(p); self->timerfd_stale_.store(true, std::memory_order_release); @@ -330,7 +328,7 @@ inline void epoll_scheduler::update_timerfd() const { auto* tsvc = timers(); - if (!tsvc) + if (!tsvc || timer_fd_ < 0) return; auto nearest = tsvc->nearest_expiry(); @@ -365,6 +363,26 @@ epoll_scheduler::update_timerfd() const detail::throw_system_error(make_err(errno), "timerfd_settime"); } +// Rounded up, so the wait never wakes before the timer is due. +inline int +epoll_scheduler::timer_timeout_ms(timer_service& svc, int timeout_ms) noexcept +{ + auto nearest = svc.nearest_expiry(); + if (nearest == timer_service::time_point::max()) + return timeout_ms; + auto now = std::chrono::steady_clock::now(); + if (nearest <= now) + return 0; + auto ms = + std::chrono::ceil(nearest - now).count(); + ms = (std::min)(ms, + static_cast( + (std::numeric_limits::max)())); + if (timeout_ms >= 0 && timeout_ms < ms) + return timeout_ms; + return static_cast(ms); +} + inline void epoll_scheduler::run_task(lock_type& lock, context_type& ctx, long timeout_us) { @@ -385,6 +403,13 @@ epoll_scheduler::run_task(lock_type& lock, context_type& ctx, long timeout_us) if (timerfd_stale_.exchange(false, std::memory_order_acquire)) update_timerfd(); + // timer_fd_ is read only once the service is seen, after which it + // no longer changes. + auto* tsvc = timers(); + bool const poll_timers = tsvc && timer_fd_ < 0; + if (poll_timers && timeout_ms != 0) + timeout_ms = timer_timeout_ms(*tsvc, timeout_ms); + int nfds = ::epoll_wait( epoll_fd_, event_buffer_.data(), static_cast(event_buffer_.size()), timeout_ms); @@ -392,7 +417,7 @@ epoll_scheduler::run_task(lock_type& lock, context_type& ctx, long timeout_us) if (nfds < 0 && errno != EINTR) detail::throw_system_error(make_err(errno), "epoll_wait"); - bool check_timers = false; + bool check_timers = poll_timers; ready_queue local_ops; for (int i = 0; i < nfds; ++i) @@ -450,8 +475,8 @@ epoll_scheduler::run_task(lock_type& lock, context_type& ctx, long timeout_us) if (check_timers) { - if (auto* tsvc = timers()) - tsvc->process_expired(); + if (auto* svc = timers()) + svc->process_expired(); update_timerfd(); } diff --git a/test/unit/fault/epoll_faults.cpp b/test/unit/fault/epoll_faults.cpp index 0699f9585..2ca5bfc04 100644 --- a/test/unit/fault/epoll_faults.cpp +++ b/test/unit/fault/epoll_faults.cpp @@ -57,11 +57,36 @@ struct epoll_faults expect_throw( sys::epoll_create1, 1, EMFILE, std::errc::too_many_files_open); expect_throw(sys::eventfd, 1, EMFILE, std::errc::too_many_files_open); - expect_throw( - sys::timerfd_create, 1, EMFILE, std::errc::too_many_files_open); - // 1 registers the eventfd, 2 the timerfd. expect_throw(sys::epoll_ctl, 1, ENOMEM, std::errc::not_enough_memory); - expect_throw(sys::epoll_ctl, 2, ENOMEM, std::errc::not_enough_memory); + } + + // A timerfd the first timer cannot get leaves timers to the + // epoll_wait timeout instead. The first timer is often made inside + // a delay's await_suspend, where a throw could only terminate. + void testFirstTimerWithoutTimerfd() + { + auto delay_completes = [](sys which, int err) { + in_child([=] { + // Timers that nothing drives would leave run() waiting. + ::alarm(10); + io_context ioc(epoll); + int const before = open_fds(); + bool done = false; + auto body = [&]() -> capy::task<> { + auto [ec] = + co_await corosio::delay(std::chrono::milliseconds(1)); + done = !ec; + }; + fault_scope f(which, err); + capy::run_async(ioc.get_executor())(body()); + ioc.run(); + return f.fired() && done && open_fds() == before; + }); + }; + delay_completes(sys::timerfd_create, EMFILE); + // The timerfd exists when registration fails; it is closed, not + // left open and unwatched. + delay_completes(sys::epoll_ctl, ENOMEM); } void testOpenFails() @@ -373,6 +398,7 @@ struct epoll_faults if (skip_under_valgrind()) return; testConstructorFails(); + testFirstTimerWithoutTimerfd(); testOpenFails(); testAcceptorRegisterFails(); testAcceptFails(); diff --git a/test/unit/lazy_services.cpp b/test/unit/lazy_services.cpp index e6595703d..67424f1ee 100644 --- a/test/unit/lazy_services.cpp +++ b/test/unit/lazy_services.cpp @@ -60,11 +60,16 @@ #include #endif +#if BOOST_COROSIO_HAS_EPOLL +#include +#endif + #include #include #include #include #include +#include #include #include "context.hpp" @@ -105,6 +110,21 @@ using file_service_t = detail::file_service; using random_access_file_service_t = detail::random_access_file_service; #endif +#if BOOST_COROSIO_HAS_EPOLL +int +open_descriptor_count() +{ + DIR* d = ::opendir("/proc/self/fd"); + if (!d) + return -1; + int n = 0; + while (::readdir(d)) + ++n; + ::closedir(d); + return n; +} +#endif + } // namespace template @@ -192,6 +212,26 @@ struct lazy_services_test BOOST_TEST_EQ(fired, static_cast(timers.size())); } + // On epoll the timerfd belongs to timers, so a context that makes + // none never opens it. + void testFirstTimerOpensTimerDescriptor() + { +#if BOOST_COROSIO_HAS_EPOLL + if constexpr (std::is_same_v< + std::remove_const_t, epoll_t>) + { + io_context ioc(Backend); + int const before = open_descriptor_count(); + detail::timer t1(ioc); + int const after = open_descriptor_count(); + detail::timer t2(ioc); + BOOST_TEST(before >= 0); + BOOST_TEST_EQ(after, before + 1); + BOOST_TEST_EQ(open_descriptor_count(), after); + } +#endif + } + void testObjectCreatesOnlyItsOwnService() { io_context ioc(Backend); @@ -287,6 +327,7 @@ struct lazy_services_test testConcurrentFirstUse(); testTimerCreatesTimerService(); testConcurrentFirstTimer(); + testFirstTimerOpensTimerDescriptor(); } }; From 7272084f2ba95edbddf50d1601794c741b74a6b5 Mon Sep 17 00:00:00 2001 From: Steve Gerbino Date: Wed, 21 Oct 2026 15:17:53 +0200 Subject: [PATCH 3/4] refactor: create IOCP's waitable timer with the first timer The IOCP scheduler created its waitable timer in its constructor but started the thread that waits on it with the first timer. It now creates both with the first timer, so a context that uses no timers holds neither. A waitable timer that could not be created was never reported: the thread was not started and timers never fired. If the waitable timer or its thread cannot be created now, the completion-port wait timeout drives the timers instead, as on epoll. A thread started by a timer made after shutdown is joined before the completion port closes. --- include/boost/corosio/detail/scheduler.hpp | 12 +-- include/boost/corosio/detail/timer.hpp | 3 - .../boost/corosio/detail/timer_service.hpp | 3 - .../native/detail/iocp/win_scheduler.hpp | 94 +++++++++++++++---- .../corosio/native/detail/iocp/win_timers.hpp | 4 +- .../native/detail/iocp/win_timers_none.hpp | 5 +- .../native/detail/iocp/win_timers_thread.hpp | 22 +++-- test/unit/fault/iocp_faults.cpp | 52 +++++----- 8 files changed, 135 insertions(+), 60 deletions(-) diff --git a/include/boost/corosio/detail/scheduler.hpp b/include/boost/corosio/detail/scheduler.hpp index c0df2821f..1c775d4d5 100644 --- a/include/boost/corosio/detail/scheduler.hpp +++ b/include/boost/corosio/detail/scheduler.hpp @@ -167,11 +167,7 @@ class BOOST_COROSIO_DECL scheduler No timer service is attached. @par Exception Safety - Strong guarantee: if the scheduler cannot start waiting on - timers, nothing is attached. - - @throws std::system_error if the scheduler cannot start waiting - on timers. + Strong guarantee: on a throw, nothing is attached. @param svc The service to attach. */ @@ -197,7 +193,11 @@ class BOOST_COROSIO_DECL scheduler /** Prepare to wait on a timer service about to be attached. Called before the service is published to the loop, so the loop - never sees a service it cannot yet wait on. + never sees a service it cannot yet wait on. A scheduler that + cannot get a kernel timer falls back to bounding its wait by the + nearest expiry rather than throwing: the first timer is often + made inside an `await_suspend`, where a throw could only + terminate. @par Exception Safety Strong guarantee: a throw leaves the scheduler as it was, and diff --git a/include/boost/corosio/detail/timer.hpp b/include/boost/corosio/detail/timer.hpp index a92d9ceb2..9add53dba 100644 --- a/include/boost/corosio/detail/timer.hpp +++ b/include/boost/corosio/detail/timer.hpp @@ -219,9 +219,6 @@ class BOOST_COROSIO_DECL timer : public io_object throws (a timer service is required). @throws std::logic_error if @p ctx is not an io_context. - - @throws std::system_error if this is the context's first timer - and the context cannot start waiting on timers. */ explicit timer(capy::execution_context& ctx); diff --git a/include/boost/corosio/detail/timer_service.hpp b/include/boost/corosio/detail/timer_service.hpp index 45c54034a..5f4001b0f 100644 --- a/include/boost/corosio/detail/timer_service.hpp +++ b/include/boost/corosio/detail/timer_service.hpp @@ -207,9 +207,6 @@ class BOOST_COROSIO_DECL timer_service final @par Exception Safety Strong guarantee: on failure the service stays detached and the next call retries. - - @throws std::system_error if the scheduler cannot start waiting - on timers. */ inline void attach() { diff --git a/include/boost/corosio/native/detail/iocp/win_scheduler.hpp b/include/boost/corosio/native/detail/iocp/win_scheduler.hpp index 4b4529b5c..609142e30 100644 --- a/include/boost/corosio/native/detail/iocp/win_scheduler.hpp +++ b/include/boost/corosio/native/detail/iocp/win_scheduler.hpp @@ -123,8 +123,11 @@ class BOOST_COROSIO_DECL win_scheduler final void post_copy(capy::continuation const& c) const; static void on_timer_changed(void* ctx); + static unsigned long + timer_timeout_ms(timer_service& svc, unsigned long timeout_ms) noexcept; void post_deferred_completions(op_queue& ops); std::size_t do_one(unsigned long timeout_ms); + std::size_t do_one_wait(unsigned long timeout_ms, bool& took_wake); void* iocp_; mutable long outstanding_work_; @@ -525,6 +528,24 @@ win_scheduler::post_deferred_completions(op_queue& ops) inline std::size_t win_scheduler::do_one(unsigned long timeout_ms) +{ + bool took_wake = false; + std::size_t n = do_one_wait(timeout_ms, took_wake); + // Without the timer thread, the wake is all that tells a thread + // blocked with no timeout about a new earliest timer; leaving + // with it would strand those threads. + if (took_wake && !timers_) + { + auto* tsvc = timers(); + if (tsvc && + tsvc->nearest_expiry() != (timer_service::time_point::max)()) + ::PostQueuedCompletionStatus(iocp_, 0, key_wake_dispatch, nullptr); + } + return n; +} + +inline std::size_t +win_scheduler::do_one_wait(unsigned long timeout_ms, bool& took_wake) { for (;;) { @@ -544,13 +565,20 @@ win_scheduler::do_one(unsigned long timeout_ms) update_timeout(); } + unsigned long wait_ms = timeout_ms; + if (auto* tsvc = timers(); tsvc && !timers_) + { + tsvc->process_expired(); + wait_ms = timer_timeout_ms(*tsvc, timeout_ms); + } + DWORD bytes = 0; ULONG_PTR key = 0; LPOVERLAPPED overlapped = nullptr; ::SetLastError(0); BOOL result = ::GetQueuedCompletionStatus( - iocp_, &bytes, &key, &overlapped, timeout_ms); + iocp_, &bytes, &key, &overlapped, wait_ms); DWORD dwError = ::GetLastError(); // Handle based on completion key @@ -623,6 +651,7 @@ win_scheduler::do_one(unsigned long timeout_ms) { case key_wake_dispatch: // Timer wakeup - loop to check dispatch_required_ + took_wake = true; continue; case key_shutdown: @@ -646,14 +675,17 @@ win_scheduler::do_one(unsigned long timeout_ms) } } - // Timeout or error. INFINITE never times out, so a WAIT_TIMEOUT - // can only be a finite caller timeout (wait_one/poll) elapsing - // with no work. run()/run_one() pass INFINITE and block until a - // real wake-up; they exit only when stop() posts key_shutdown - // (a failed post is fatal in stop()). + // Timeout or error. Unless a timer cut the wait short, a + // WAIT_TIMEOUT is a finite caller timeout (wait_one/poll) + // elapsing with no work. run()/run_one() pass INFINITE and + // block until a real wake-up; they exit only when stop() posts + // key_shutdown (a failed post is fatal in stop()). if (dwError != WAIT_TIMEOUT) detail::throw_system_error(make_err(dwError)); - return 0; + if (wait_ms == timeout_ms) + return 0; + if (timeout_ms != INFINITE) + timeout_ms -= wait_ms; } } @@ -663,22 +695,52 @@ win_scheduler::on_timer_changed(void* ctx) static_cast(ctx)->update_timeout(); } +// Rounds up, so a timer is never seen early and polled for again. +inline unsigned long +win_scheduler::timer_timeout_ms( + timer_service& svc, unsigned long timeout_ms) noexcept +{ + auto nearest = svc.nearest_expiry(); + if (nearest == (timer_service::time_point::max)()) + return timeout_ms; + auto now = std::chrono::steady_clock::now(); + if (nearest <= now) + return 0; + auto ms = + std::chrono::ceil(nearest - now).count(); + // Kept below INFINITE, which would never time out. + ms = (std::min)(ms, static_cast(INFINITE - 1)); + return (std::min)(timeout_ms, static_cast(ms)); +} + inline void win_scheduler::on_timer_service_attached(timer_service& svc) { - // Pass 'this' as context - callback routes to correct instance + // Without the timer thread the completion-port wait timeout drives + // timers instead. The first timer is often made inside a delay's + // await_suspend, where a throw could only terminate. + auto timers = make_win_timers(iocp_, &dispatch_required_); + if (timers->start()) + timers_ = std::move(timers); svc.set_on_earliest_changed( timer_service::callback{this, &on_timer_changed}); - if (timers_) - timers_->start(); } inline void win_scheduler::update_timeout() { auto* tsvc = timers(); - if (tsvc && timers_) + if (!tsvc) + return; + if (timers_) + { timers_->update_timeout(tsvc->nearest_expiry()); + } + else + { + // A run loop may be waiting past the new earliest expiry. + ::PostQueuedCompletionStatus(iocp_, 0, key_wake_dispatch, nullptr); + } } } // namespace boost::corosio::detail @@ -715,20 +777,15 @@ inline win_scheduler::win_scheduler( try { - timers_ = make_win_timers(iocp_, &dispatch_required_); - // A scheduler whose wait reactor could not be built would // answer every wait with a parked op, so it refuses to exist - // instead. Last, so the catch below is the whole cleanup: - // the reactor holds its own Winsock reference and needs no - // help from the order. + // instead. wait_reactor_ = std::make_unique(*this); } catch (...) { // ~win_scheduler never runs for a constructor that throws, and // the port is a raw handle nothing else owns. - timers_.reset(); ::CloseHandle(iocp_); iocp_ = nullptr; throw; @@ -845,6 +902,9 @@ win_scheduler::sweep_port() noexcept inline win_scheduler::~win_scheduler() { + // A timer made after shutdown starts a thread that posts to the + // port, so it is joined before the port closes. + timers_.reset(); wait_reactor_->stop(); wait_reactor_.reset(); diff --git a/include/boost/corosio/native/detail/iocp/win_timers.hpp b/include/boost/corosio/native/detail/iocp/win_timers.hpp index 41d1ea9e8..16c8e376a 100644 --- a/include/boost/corosio/native/detail/iocp/win_timers.hpp +++ b/include/boost/corosio/native/detail/iocp/win_timers.hpp @@ -42,7 +42,9 @@ class win_timers virtual ~win_timers() = default; - virtual void start() = 0; + /// Start delivering wakeups; return false, keeping nothing, if + /// the resources for them cannot be created. + virtual bool start() = 0; virtual void stop() = 0; virtual void update_timeout(time_point next_expiry) = 0; }; diff --git a/include/boost/corosio/native/detail/iocp/win_timers_none.hpp b/include/boost/corosio/native/detail/iocp/win_timers_none.hpp index 53a1dfa7e..8eb3011bc 100644 --- a/include/boost/corosio/native/detail/iocp/win_timers_none.hpp +++ b/include/boost/corosio/native/detail/iocp/win_timers_none.hpp @@ -26,7 +26,10 @@ class win_timers_none final : public win_timers public: win_timers_none() = default; - void start() override {} + bool start() override + { + return true; + } void stop() override {} void update_timeout(time_point) override {} }; diff --git a/include/boost/corosio/native/detail/iocp/win_timers_thread.hpp b/include/boost/corosio/native/detail/iocp/win_timers_thread.hpp index 9465be551..4d11bbb64 100644 --- a/include/boost/corosio/native/detail/iocp/win_timers_thread.hpp +++ b/include/boost/corosio/native/detail/iocp/win_timers_thread.hpp @@ -18,6 +18,7 @@ #include #include #include +#include #include namespace boost::corosio::detail { @@ -36,7 +37,7 @@ class win_timers_thread final : public win_timers win_timers_thread(win_timers_thread const&) = delete; win_timers_thread& operator=(win_timers_thread const&) = delete; - void start() override; + bool start() override; void stop() override; void update_timeout(time_point next_expiry) override; @@ -49,7 +50,6 @@ inline win_timers_thread::win_timers_thread( : win_timers(dispatch_required) , iocp_(iocp_handle) { - waitable_timer_ = ::CreateWaitableTimerW(nullptr, FALSE, nullptr); } inline win_timers_thread::~win_timers_thread() @@ -59,13 +59,23 @@ inline win_timers_thread::~win_timers_thread() ::CloseHandle(waitable_timer_); } -inline void +inline bool win_timers_thread::start() { + waitable_timer_ = ::CreateWaitableTimerW(nullptr, FALSE, nullptr); if (!waitable_timer_) - return; - - thread_ = std::thread([this] { thread_func(); }); + return false; + try + { + thread_ = std::thread([this] { thread_func(); }); + } + catch (std::system_error const&) + { + ::CloseHandle(waitable_timer_); + waitable_timer_ = nullptr; + return false; + } + return true; } inline void diff --git a/test/unit/fault/iocp_faults.cpp b/test/unit/fault/iocp_faults.cpp index 023605bb4..170108648 100644 --- a/test/unit/fault/iocp_faults.cpp +++ b/test/unit/fault/iocp_faults.cpp @@ -182,30 +182,36 @@ struct iocp_faults } } - void testTimerCreationFails() + // A waitable timer or timer thread the first timer cannot get + // leaves timers to the completion-port wait timeout instead. The + // first timer is often made inside a delay's await_suspend, where a + // throw could only terminate. + void testFirstTimerWithoutTimerThread() { - // A null waitable timer is never reported: start() returns - // without a thread and update_timeout() does nothing, so - // timers simply never fire. - fault_scope f(sys::CreateWaitableTimerW, ERROR_NOT_ENOUGH_MEMORY); - io_context ioc(iocp); - BOOST_TEST(f.fired()); - - bool fired_timer = false; - auto body = [&]() -> capy::task<> { - std::ignore = co_await corosio::delay(std::chrono::milliseconds(1)); - fired_timer = true; + auto delay_completes = [](sys which, int err) { + expect_no_handle_leak( + [&] { + io_context ioc(iocp); + bool done = false; + auto body = [&]() -> capy::task<> { + auto [ec] = co_await corosio::delay( + std::chrono::milliseconds(1)); + done = !ec; + }; + fault_scope f(which, err); + capy::run_async(ioc.get_executor())(body()); + // Bounds a hang, not a timing assertion. + ioc.run_for(std::chrono::seconds(10)); + BOOST_TEST(f.fired()); + BOOST_TEST(done); + }, + 16, 8); }; - capy::run_async(ioc.get_executor())(body()); - // Nothing else can end the loop, so the run is bounded by the - // stop rather than by the timer. - ioc.stop(); - ioc.run(); - // A smoke check rather than a proof: the stop latches before - // run() begins, so the loop would have returned early even with - // a working timer. What the arm proves is that construction - // swallowed the failure; this only shows nothing fired anyway. - BOOST_TEST(!fired_timer); + delay_completes(sys::CreateWaitableTimerW, ERROR_NOT_ENOUGH_MEMORY); + // The waitable timer exists when the thread is refused; it is + // closed, not left open and unwatched. + if (hook_is_live(sys::pthread_create)) + delay_completes(sys::pthread_create, ERROR_MAX_THRDS_REACHED); } void testTimerThreadWaitFails() @@ -1954,7 +1960,7 @@ struct iocp_faults void run() { testSchedulerConstructFails(); - testTimerCreationFails(); + testFirstTimerWithoutTimerThread(); testTimerThreadWaitFails(); testStopPostFails(); testPostFallbackRuns(); From 804651d93ced9727ff5346a30b45e51522eb7f1e Mon Sep 17 00:00:00 2001 From: Steve Gerbino Date: Wed, 21 Oct 2026 15:17:54 +0200 Subject: [PATCH 4/4] refactor: resume timer waiters by dispatch A timer's completion runs on the scheduler, then posted the waiter's continuation, so every expiry and cancellation went through the queue twice. It now dispatches the continuation, as I/O completions do: the waiter resumes at once when its executor's context is running on this thread, and is queued as before otherwise. --- include/boost/corosio/delay.hpp | 2 +- include/boost/corosio/detail/timer.hpp | 4 ++-- .../boost/corosio/detail/timer_service.hpp | 2 +- src/corosio/src/timer.cpp | 11 +++++----- test/unit/delay.cpp | 21 +++++++++++++++++++ 5 files changed, 30 insertions(+), 10 deletions(-) diff --git a/include/boost/corosio/delay.hpp b/include/boost/corosio/delay.hpp index ad41d5f53..55935360e 100644 --- a/include/boost/corosio/delay.hpp +++ b/include/boost/corosio/delay.hpp @@ -234,7 +234,7 @@ class clock_delay_awaitable } // Runs on the scheduler thread executing the completion op, - // before the continuation is posted, so the frame cannot die + // before the continuation is dispatched, so the frame cannot die // concurrently. static bool on_fire(void* ctx) noexcept { diff --git a/include/boost/corosio/detail/timer.hpp b/include/boost/corosio/detail/timer.hpp index 9add53dba..a1ee5a67f 100644 --- a/include/boost/corosio/detail/timer.hpp +++ b/include/boost/corosio/detail/timer.hpp @@ -427,10 +427,10 @@ struct BOOST_COROSIO_SYMBOL_VISIBLE waiter_node /// The timer service that completes this waiter. timer_service* svc_ = nullptr; - /// The continuation posted to resume the coroutine. + /// The continuation that resumes the coroutine. capy::continuation cont_; - /// The executor the continuation is posted through. + /// The executor the coroutine resumes on. capy::executor_ref d_; // Borrowed from the awaiting chain's await_context, which outlives the diff --git a/include/boost/corosio/detail/timer_service.hpp b/include/boost/corosio/detail/timer_service.hpp index 5f4001b0f..8fbc2b8cb 100644 --- a/include/boost/corosio/detail/timer_service.hpp +++ b/include/boost/corosio/detail/timer_service.hpp @@ -81,7 +81,7 @@ class scheduler; pointer on waiter_node is used as a "still in list" marker. A waiter_node's storage is the suspended coroutine's frame: every completion path must finish touching the node before - posting the continuation or destroying the handle. + dispatching the continuation or destroying the handle. */ inline void timer_service_invalidate_cache() noexcept; diff --git a/src/corosio/src/timer.cpp b/src/corosio/src/timer.cpp index 200eab7c7..762501b57 100644 --- a/src/corosio/src/timer.cpp +++ b/src/corosio/src/timer.cpp @@ -10,6 +10,7 @@ #include #include +#include #include namespace boost::corosio::detail { @@ -164,9 +165,9 @@ waiter_node::completion_op::do_complete( void waiter_node::completion_op::operator()() noexcept { - // The node lives in the resuming coroutine's frame: posting the - // continuation is the last access, since the frame (and node) - // may complete and die on another thread immediately after. + // The node lives in the resuming coroutine's frame: dispatching + // the continuation is the last access, since the frame (and node) + // may complete and die, here or on another thread, once it runs. auto* w = waiter_; // A true return means the waiter re-published itself: the frame // stays suspended, the wait's work count stays live, and the @@ -177,9 +178,7 @@ waiter_node::completion_op::operator()() noexcept w->reset_stop_cb(); auto d = w->d_; auto& sched = w->svc_->get_scheduler(); - // A throw terminates: the waiter is already off the heap, so nothing - // else could resume or abandon the coroutine. - d.post(w->cont_); + dispatch_coro(d, w->cont_).resume(); sched.work_finished(); } diff --git a/test/unit/delay.cpp b/test/unit/delay.cpp index 13eb72094..66f247262 100644 --- a/test/unit/delay.cpp +++ b/test/unit/delay.cpp @@ -251,6 +251,26 @@ struct delay_test BOOST_TEST(canceled); } + // The completion runs on the context the waiter belongs to, so it + // resumes the waiter itself rather than queueing it a second time. + void testCompletionResumesWaiterInline() + { + io_context ioc(Backend); + std::stop_source src; + bool resumed = false; + + auto t = [](bool& resumed_out) -> capy::task<> { + std::ignore = co_await delay(std::chrono::hours(1)); + resumed_out = true; + }; + capy::run_async(ioc.get_executor(), src.get_token())(t(resumed)); + + ioc.poll(); + src.request_stop(); + BOOST_TEST_EQ(ioc.poll_one(), 1u); + BOOST_TEST(resumed); + } + void testAlreadyStoppedCompletesCanceled() { io_context ioc(Backend); @@ -1177,6 +1197,7 @@ struct delay_test testZeroDurationCompletesImmediately(); testPastTimePointCompletesImmediately(); testCancellation(); + testCompletionResumesWaiterInline(); testFirstTimerMadeOnAnotherThread(); testAlreadyStoppedCompletesCanceled(); testZeroDurationWithStopRequested();