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/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/scheduler.hpp b/include/boost/corosio/detail/scheduler.hpp index 2c28a006b..1c775d4d5 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: on a throw, nothing is attached. + + @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. 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 + 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..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 846769ce7..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; @@ -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,32 @@ 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. + */ + 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 +737,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..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,8 @@ 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_; int event_fd_; @@ -159,7 +163,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) @@ -177,48 +181,45 @@ inline epoll_scheduler::epoll_scheduler(capy::execution_context& ctx, 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) + 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) { - int errn = errno; - ::close(timer_fd_); - ::close(event_fd_); - ::close(epoll_fd_); - detail::throw_system_error(make_err(errn), "epoll_ctl (timerfd)"); + // 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; + } } - - 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_); + 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() @@ -326,7 +327,10 @@ epoll_scheduler::interrupt_reactor() const inline void epoll_scheduler::update_timerfd() const { - auto nearest = timer_svc_->nearest_expiry(); + auto* tsvc = timers(); + if (!tsvc || timer_fd_ < 0) + return; + auto nearest = tsvc->nearest_expiry(); itimerspec ts{}; int flags = 0; @@ -359,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) { @@ -379,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); @@ -386,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) @@ -444,7 +475,8 @@ epoll_scheduler::run_task(lock_type& lock, context_type& ctx, long timeout_us) if (check_timers) { - timer_svc_->process_expired(); + if (auto* svc = timers()) + svc->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..609142e30 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; @@ -122,10 +123,12 @@ 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); - timer_service* timer_svc_ = nullptr; 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 (;;) { @@ -538,19 +559,26 @@ 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(); } + 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::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( + // 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() { - if (timer_svc_ && timers_) - timers_->update_timeout(timer_svc_->nearest_expiry()); + auto* 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 @@ -697,7 +759,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) @@ -715,35 +777,15 @@ 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 - // 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. 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. - timers_.reset(); + // the port is a raw handle nothing else owns. ::CloseHandle(iocp_); iocp_ = nullptr; throw; @@ -756,19 +798,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(); @@ -870,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/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..762501b57 100644 --- a/src/corosio/src/timer.cpp +++ b/src/corosio/src/timer.cpp @@ -10,14 +10,30 @@ #include #include +#include #include 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))) { } @@ -149,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 @@ -162,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 f18a77887..66f247262 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); @@ -227,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); @@ -1153,6 +1197,8 @@ struct delay_test testZeroDurationCompletesImmediately(); testPastTimePointCompletesImmediately(); testCancellation(); + testCompletionResumesWaiterInline(); + testFirstTimerMadeOnAnotherThread(); testAlreadyStoppedCompletesCanceled(); testZeroDurationWithStopRequested(); testPastTimePointWithStopRequested(); 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/fault/iocp_faults.cpp b/test/unit/fault/iocp_faults.cpp index 2bf82ee3b..170108648 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 @@ -181,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() @@ -236,6 +243,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)); } @@ -1951,7 +1960,7 @@ struct iocp_faults void run() { testSchedulerConstructFails(); - testTimerCreationFails(); + testFirstTimerWithoutTimerThread(); testTimerThreadWaitFails(); testStopPostFails(); testPostFallbackRuns(); diff --git a/test/unit/lazy_services.cpp b/test/unit/lazy_services.cpp index df84525e4..67424f1ee 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,8 +60,16 @@ #include #endif +#if BOOST_COROSIO_HAS_EPOLL +#include +#endif + +#include +#include +#include #include #include +#include #include #include "context.hpp" @@ -99,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 @@ -138,11 +164,74 @@ 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())); + } + + // 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); @@ -236,6 +325,9 @@ struct lazy_services_test testObjectCreatesOnlyItsOwnService(); testEachObjectCreatesItsService(); testConcurrentFirstUse(); + testTimerCreatesTimerService(); + testConcurrentFirstTimer(); + testFirstTimerOpensTimerDescriptor(); } };