Diff coverage report for

//

//

// Copyright (c) 2025 Vinnie Falco (vinnie.falco@gmail.com)

// Copyright (c) 2025 Vinnie Falco (vinnie.falco@gmail.com)

// Copyright (c) 2026 Steve Gerbino

// Copyright (c) 2026 Steve Gerbino

//

//

// Distributed under the Boost Software License, Version 1.0. (See accompanying

// Distributed under the Boost Software License, Version 1.0. (See accompanying

// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)

// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)

//

//

// Official repository: https://github.com/cppalliance/corosio

// Official repository: https://github.com/cppalliance/corosio

//

//

#ifndef BOOST_COROSIO_DETAIL_TIMER_SERVICE_HPP

#ifndef BOOST_COROSIO_DETAIL_TIMER_SERVICE_HPP

#define BOOST_COROSIO_DETAIL_TIMER_SERVICE_HPP

#define BOOST_COROSIO_DETAIL_TIMER_SERVICE_HPP

#include <boost/corosio/timer.hpp>

#include <boost/corosio/timer.hpp>

#include <boost/corosio/io_context.hpp>

#include <boost/corosio/io_context.hpp>

#include <boost/corosio/detail/scheduler_op.hpp>

#include <boost/corosio/detail/scheduler_op.hpp>

#include <boost/corosio/native/native_scheduler.hpp>

#include <boost/corosio/native/native_scheduler.hpp>

#include <boost/corosio/detail/intrusive.hpp>

#include <boost/corosio/detail/intrusive.hpp>

#include <boost/corosio/detail/thread_local_ptr.hpp>

#include <boost/corosio/detail/thread_local_ptr.hpp>

#include <boost/capy/error.hpp>

#include <boost/capy/error.hpp>

#include <boost/capy/ex/execution_context.hpp>

#include <boost/capy/ex/execution_context.hpp>

#include <boost/capy/ex/executor_ref.hpp>

#include <boost/capy/ex/executor_ref.hpp>

#include <system_error>

#include <system_error>

#include <atomic>

#include <atomic>

#include <chrono>

#include <chrono>

#include <coroutine>

#include <coroutine>

#include <cstddef>

#include <cstddef>

#include <limits>

#include <limits>

#include <mutex>

#include <mutex>

#include <optional>

#include <optional>

#include <stop_token>

#include <stop_token>

#include <utility>

#include <utility>

#include <vector>

#include <vector>

namespace boost::corosio::detail {

namespace boost::corosio::detail {

struct scheduler;

struct scheduler;

/*

/*

    Timer Service

    Timer Service

    =============

    =============

    Data Structures

    Data Structures

    ---------------

    ---------------

    waiter_node holds per-waiter state: coroutine handle, executor,

    waiter_node holds per-waiter state: coroutine handle, executor,

    error output, stop_token, embedded completion_op. Each concurrent

    error output, stop_token, embedded completion_op. Each concurrent

    co_await t.wait() allocates one waiter_node.

    co_await t.wait() allocates one waiter_node.

    timer_service::implementation holds per-timer state: expiry,

    timer_service::implementation holds per-timer state: expiry,

    heap index, and an intrusive_list of waiter_nodes. Multiple

    heap index, and an intrusive_list of waiter_nodes. Multiple

    coroutines can wait on the same timer simultaneously.

    coroutines can wait on the same timer simultaneously.

    timer_service owns a min-heap of active timers, a free list

    timer_service owns a min-heap of active timers, a free list

    of recycled impls, and a free list of recycled waiter_nodes. The

    of recycled impls, and a free list of recycled waiter_nodes. The

    heap is ordered by expiry time; the scheduler queries

    heap is ordered by expiry time; the scheduler queries

    nearest_expiry() to set the epoll/timerfd timeout.

    nearest_expiry() to set the epoll/timerfd timeout.

    Optimization Strategy

    Optimization Strategy

    ---------------------

    ---------------------

    1. Deferred heap insertion — expires_after() stores the expiry

    1. Deferred heap insertion — expires_after() stores the expiry

       but does not insert into the heap. Insertion happens in wait().

       but does not insert into the heap. Insertion happens in wait().

    2. Thread-local impl cache — single-slot per-thread cache.

    2. Thread-local impl cache — single-slot per-thread cache.

    3. Embedded completion_op — eliminates heap allocation per fire/cancel.

    3. Embedded completion_op — eliminates heap allocation per fire/cancel.

    4. Cached nearest expiry — atomic avoids mutex in nearest_expiry().

    4. Cached nearest expiry — atomic avoids mutex in nearest_expiry().

    5. might_have_pending_waits_ flag — skips lock when no wait issued.

    5. might_have_pending_waits_ flag — skips lock when no wait issued.

    6. Thread-local waiter cache — single-slot per-thread cache.

    6. Thread-local waiter cache — single-slot per-thread cache.

    Concurrency

    Concurrency

    -----------

    -----------

    stop_token callbacks can fire from any thread. The impl_

    stop_token callbacks can fire from any thread. The impl_

    pointer on waiter_node is used as a "still in list" marker.

    pointer on waiter_node is used as a "still in list" marker.

*/

*/

struct BOOST_COROSIO_SYMBOL_VISIBLE waiter_node;

struct BOOST_COROSIO_SYMBOL_VISIBLE waiter_node;

inline void timer_service_invalidate_cache() noexcept;

inline void timer_service_invalidate_cache() noexcept;

// timer_service class body — member function definitions are

// timer_service class body — member function definitions are

// out-of-class (after implementation and waiter_node are complete)

// out-of-class (after implementation and waiter_node are complete)

class BOOST_COROSIO_DECL timer_service final

class BOOST_COROSIO_DECL timer_service final

    : public capy::execution_context::service

    : public capy::execution_context::service

    , public io_object::io_service

    , public io_object::io_service

{

{

public:

public:

    using clock_type = std::chrono::steady_clock;

    using clock_type = std::chrono::steady_clock;

    using time_point = clock_type::time_point;

    using time_point = clock_type::time_point;

    /// Type-erased callback for earliest-expiry-changed notifications.

    /// Type-erased callback for earliest-expiry-changed notifications.

    class callback

    class callback

    {

    {

        void* ctx_         = nullptr;

        void* ctx_         = nullptr;

        void (*fn_)(void*) = nullptr;

        void (*fn_)(void*) = nullptr;

    public:

    public:

        /// Construct an empty callback.

        /// Construct an empty callback.

        /// Construct a callback with the given context and function.

        /// Construct a callback with the given context and function.

        /// Return true if the callback is non-empty.

        /// Return true if the callback is non-empty.

        explicit operator bool() const noexcept

        explicit operator bool() const noexcept

        {

        {

            return fn_ != nullptr;

            return fn_ != nullptr;

        }

        }

        /// Invoke the callback.

        /// Invoke the callback.

        {

        {

    };

    };

    struct implementation;

    struct implementation;

private:

private:

    struct heap_entry

    struct heap_entry

    {

    {

        time_point time_;

        time_point time_;

        implementation* timer_;

        implementation* timer_;

    };

    };

    scheduler* sched_ = nullptr;

    scheduler* sched_ = nullptr;

    mutable std::mutex mutex_;

    mutable std::mutex mutex_;

    std::vector<heap_entry> heap_;

    std::vector<heap_entry> heap_;

    implementation* free_list_     = nullptr;

    implementation* free_list_     = nullptr;

    waiter_node* waiter_free_list_ = nullptr;

    waiter_node* waiter_free_list_ = nullptr;

    callback on_earliest_changed_;

    callback on_earliest_changed_;

    bool shutting_down_ = false;

    bool shutting_down_ = false;

    // Avoids mutex in nearest_expiry() and empty()

    // Avoids mutex in nearest_expiry() and empty()

    mutable std::atomic<std::int64_t> cached_nearest_ns_{

    mutable std::atomic<std::int64_t> cached_nearest_ns_{

        (std::numeric_limits<std::int64_t>::max)()};

        (std::numeric_limits<std::int64_t>::max)()};

public:

public:

    /// Construct the timer service bound to a scheduler.

    /// Construct the timer service bound to a scheduler.

    {

    {

    /// Return the associated scheduler.

    /// Return the associated scheduler.

    {

    {

    }

    }

    /// Destroy the timer service.

    /// Destroy the timer service.

    timer_service(timer_service const&)            = delete;

    timer_service(timer_service const&)            = delete;

    timer_service& operator=(timer_service const&) = delete;

    timer_service& operator=(timer_service const&) = delete;

    /// Register a callback invoked when the earliest expiry changes.

    /// Register a callback invoked when the earliest expiry changes.

    {

    {

    /// Return true if no timers are in the heap.

    /// Return true if no timers are in the heap.

    inline bool empty() const noexcept

    inline bool empty() const noexcept

    {

    {

        return cached_nearest_ns_.load(std::memory_order_acquire) ==

        return cached_nearest_ns_.load(std::memory_order_acquire) ==

            (std::numeric_limits<std::int64_t>::max)();

            (std::numeric_limits<std::int64_t>::max)();

    }

    }

    /// Return the nearest timer expiry without acquiring the mutex.

    /// Return the nearest timer expiry without acquiring the mutex.

    {

    {

    }

    }

    /// Cancel all pending timers and free cached resources.

    /// Cancel all pending timers and free cached resources.

    inline void shutdown() override;

    inline void shutdown() override;

    /// Construct a new timer implementation.

    /// Construct a new timer implementation.

    inline io_object::implementation* construct() override;

    inline io_object::implementation* construct() override;

    /// Destroy a timer implementation, cancelling pending waiters.

    /// Destroy a timer implementation, cancelling pending waiters.

    inline void destroy(io_object::implementation* p) override;

    inline void destroy(io_object::implementation* p) override;

    /// Cancel and recycle a timer implementation.

    /// Cancel and recycle a timer implementation.

    inline void destroy_impl(implementation& impl);

    inline void destroy_impl(implementation& impl);

    /// Create or recycle a waiter node.

    /// Create or recycle a waiter node.

    inline waiter_node* create_waiter();

    inline waiter_node* create_waiter();

    /// Return a waiter node to the cache or free list.

    /// Return a waiter node to the cache or free list.

    inline void destroy_waiter(waiter_node* w);

    inline void destroy_waiter(waiter_node* w);

    /// Update the timer expiry, cancelling existing waiters.

    /// Update the timer expiry, cancelling existing waiters.

    inline std::size_t update_timer(implementation& impl, time_point new_time);

    inline std::size_t update_timer(implementation& impl, time_point new_time);

    /// Insert a waiter into the timer's waiter list and the heap.

    /// Insert a waiter into the timer's waiter list and the heap.

    inline void insert_waiter(implementation& impl, waiter_node* w);

    inline void insert_waiter(implementation& impl, waiter_node* w);

    /// Cancel all waiters on a timer.

    /// Cancel all waiters on a timer.

    inline std::size_t cancel_timer(implementation& impl);

    inline std::size_t cancel_timer(implementation& impl);

    /// Cancel a single waiter ( stop_token callback path ).

    /// Cancel a single waiter ( stop_token callback path ).

    inline void cancel_waiter(waiter_node* w);

    inline void cancel_waiter(waiter_node* w);

    /// Cancel one waiter on a timer.

    /// Cancel one waiter on a timer.

    inline std::size_t cancel_one_waiter(implementation& impl);

    inline std::size_t cancel_one_waiter(implementation& impl);

    /// Complete all waiters whose timers have expired.

    /// Complete all waiters whose timers have expired.

    inline std::size_t process_expired();

    inline std::size_t process_expired();

private:

private:

    {

    {

    inline void remove_timer_impl(implementation& impl);

    inline void remove_timer_impl(implementation& impl);

    inline void up_heap(std::size_t index);

    inline void up_heap(std::size_t index);

    inline void down_heap(std::size_t index);

    inline void down_heap(std::size_t index);

    inline void swap_heap(std::size_t i1, std::size_t i2);

    inline void swap_heap(std::size_t i1, std::size_t i2);

};

};

struct BOOST_COROSIO_SYMBOL_VISIBLE waiter_node

struct BOOST_COROSIO_SYMBOL_VISIBLE waiter_node

    : intrusive_list<waiter_node>::node

    : intrusive_list<waiter_node>::node

{

{

    // Embedded completion op — avoids heap allocation per fire/cancel

    // Embedded completion op — avoids heap allocation per fire/cancel

    struct completion_op final : scheduler_op

    struct completion_op final : scheduler_op

    {

    {

        waiter_node* waiter_ = nullptr;

        waiter_node* waiter_ = nullptr;

        static void do_complete(

        static void do_complete(

            void* owner, scheduler_op* base, std::uint32_t, std::uint32_t);

            void* owner, scheduler_op* base, std::uint32_t, std::uint32_t);

        void operator()() override;

        void operator()() override;

        void destroy() override;

        void destroy() override;

    };

    };

    // Per-waiter stop_token cancellation

    // Per-waiter stop_token cancellation

    struct canceller

    struct canceller

    {

    {

        waiter_node* waiter_;

        waiter_node* waiter_;

        void operator()() const;

        void operator()() const;

    };

    };

    // nullptr once removed from timer's waiter list (concurrency marker)

    // nullptr once removed from timer's waiter list (concurrency marker)

    timer_service::implementation* impl_ = nullptr;

    timer_service::implementation* impl_ = nullptr;

    timer_service* svc_                  = nullptr;

    timer_service* svc_                  = nullptr;

    std::coroutine_handle<> h_;

    std::coroutine_handle<> h_;

    capy::continuation* cont_            = nullptr;

    capy::continuation* cont_            = nullptr;

    capy::executor_ref d_;

    capy::executor_ref d_;

    std::error_code* ec_out_ = nullptr;

    std::error_code* ec_out_ = nullptr;

    std::stop_token token_;

    std::stop_token token_;

    std::optional<std::stop_callback<canceller>> stop_cb_;

    std::optional<std::stop_callback<canceller>> stop_cb_;

    completion_op op_;

    completion_op op_;

    std::error_code ec_value_;

    std::error_code ec_value_;

    waiter_node* next_free_ = nullptr;

    waiter_node* next_free_ = nullptr;

};

};

struct timer_service::implementation final : timer::implementation

struct timer_service::implementation final : timer::implementation

{

{

    using clock_type = std::chrono::steady_clock;

    using clock_type = std::chrono::steady_clock;

    using time_point = clock_type::time_point;

    using time_point = clock_type::time_point;

    using duration   = clock_type::duration;

    using duration   = clock_type::duration;

    timer_service* svc_ = nullptr;

    timer_service* svc_ = nullptr;

    intrusive_list<waiter_node> waiters_;

    intrusive_list<waiter_node> waiters_;

    // Free list linkage (reused when impl is on free_list)

    // Free list linkage (reused when impl is on free_list)

    implementation* next_free_ = nullptr;

    implementation* next_free_ = nullptr;

    inline explicit implementation(timer_service& svc) noexcept;

    inline explicit implementation(timer_service& svc) noexcept;

    inline std::coroutine_handle<> wait(

    inline std::coroutine_handle<> wait(

        std::coroutine_handle<>,

        std::coroutine_handle<>,

        capy::executor_ref,

        capy::executor_ref,

        std::stop_token,

        std::stop_token,

        std::error_code*,

        std::error_code*,

        capy::continuation*) override;

        capy::continuation*) override;

};

};

// Thread-local caches avoid hot-path mutex acquisitions:

// Thread-local caches avoid hot-path mutex acquisitions:

// 1. Impl cache — single-slot, validated by comparing svc_

// 1. Impl cache — single-slot, validated by comparing svc_

// 2. Waiter cache — single-slot, no service affinity

// 2. Waiter cache — single-slot, no service affinity

// All caches are cleared by timer_service_invalidate_cache() during shutdown.

// All caches are cleared by timer_service_invalidate_cache() during shutdown.

inline thread_local_ptr<timer_service::implementation> tl_cached_impl;

inline thread_local_ptr<timer_service::implementation> tl_cached_impl;

inline thread_local_ptr<waiter_node> tl_cached_waiter;

inline thread_local_ptr<waiter_node> tl_cached_waiter;

inline timer_service::implementation*

inline timer_service::implementation*

{

{

    {

    {

        // Stale impl from a destroyed service

        // Stale impl from a destroyed service

    }

    }

}

}

inline bool

inline bool

{

{

    {

    {

    }

    }

}

}

inline waiter_node*

inline waiter_node*

{

{

    {

    {

    }

    }

}

}

inline bool

inline bool

{

{

    {

    {

    }

    }

}

}

inline void

inline void

{

{

// timer_service out-of-class member function definitions

// timer_service out-of-class member function definitions

{

{

inline void

inline void

{

{

    // Snapshot impls and detach them from the heap so that

    // Snapshot impls and detach them from the heap so that

    // coroutine-owned timer destructors (triggered by h.destroy()

    // coroutine-owned timer destructors (triggered by h.destroy()

    // below) cannot re-enter remove_timer_impl() and mutate the

    // below) cannot re-enter remove_timer_impl() and mutate the

    // vector during iteration.

    // vector during iteration.

    {

    {

    }

    }

        (std::numeric_limits<std::int64_t>::max)(), std::memory_order_release);

        (std::numeric_limits<std::int64_t>::max)(), std::memory_order_release);

    // Cancel waiting timers. Each waiter called work_started()

    // Cancel waiting timers. Each waiter called work_started()

    // in implementation::wait(). On IOCP the scheduler shutdown

    // in implementation::wait(). On IOCP the scheduler shutdown

    // loop exits when outstanding_work_ reaches zero, so we must

    // loop exits when outstanding_work_ reaches zero, so we must

    // call work_finished() here to balance it. On other backends

    // call work_finished() here to balance it. On other backends

    // this is harmless.

    // this is harmless.

    {

    {

        {

        {

    }

    }

    // Delete free-listed impls

    // Delete free-listed impls

    {

    {

    }

    }

    // Delete free-listed waiters

    // Delete free-listed waiters

    {

    {

    }

    }

inline io_object::implementation*

inline io_object::implementation*

{

{

    {

    {

    }

    }

    {

    {

    }

    }

    else

    else

    {

    {

    }

    }

inline void

inline void

{

{

inline void

inline void

{

{

    // During shutdown the impl is owned by the shutdown loop.

    // During shutdown the impl is owned by the shutdown loop.

    // Re-entering here (from a coroutine-owned timer destructor

    // Re-entering here (from a coroutine-owned timer destructor

    // triggered by h.destroy()) must not modify the heap or

    // triggered by h.destroy()) must not modify the heap or

    // recycle the impl — shutdown deletes it directly.

    // recycle the impl — shutdown deletes it directly.

    {

    {

inline waiter_node*

inline waiter_node*

{

{

    {

    {

    }

    }

inline void

inline void

{

{

inline std::size_t

inline std::size_t

{

{

    bool in_heap =

    bool in_heap =

    {

    {

        {

        {