Diff coverage report for

//

//

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

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

//

//

// 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/capy

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

//

//

#ifndef BOOST_CAPY_ASYNC_EVENT_HPP

#ifndef BOOST_CAPY_ASYNC_EVENT_HPP

#define BOOST_CAPY_ASYNC_EVENT_HPP

#define BOOST_CAPY_ASYNC_EVENT_HPP

#include <boost/capy/detail/config.hpp>

#include <boost/capy/detail/config.hpp>

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

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

#include <boost/capy/concept/executor.hpp>

#include <boost/capy/concept/executor.hpp>

#include <boost/capy/coro.hpp>

#include <boost/capy/coro.hpp>

#include <boost/capy/error.hpp>

#include <boost/capy/error.hpp>

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

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

#include <boost/capy/io_result.hpp>

#include <boost/capy/io_result.hpp>

#include <stop_token>

#include <stop_token>

#include <atomic>

#include <atomic>

#include <coroutine>

#include <coroutine>

#include <new>

#include <new>

#include <utility>

#include <utility>

/*  async_event implementation notes

/*  async_event implementation notes

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

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

    Same cancellation pattern as async_mutex (see that file for the

    Same cancellation pattern as async_mutex (see that file for the

    full discussion on claimed_, stop_cb lifetime, member ordering,

    full discussion on claimed_, stop_cb lifetime, member ordering,

    and threading assumptions).

    and threading assumptions).

    Key difference: set() wakes ALL waiters (broadcast), not one.

    Key difference: set() wakes ALL waiters (broadcast), not one.

    It pops every waiter from the list and dispatches the ones it

    It pops every waiter from the list and dispatches the ones it

    claims. Waiters already claimed by a stop callback are skipped.

    claims. Waiters already claimed by a stop callback are skipped.

    Because set() pops all waiters, a canceled waiter may have been

    Because set() pops all waiters, a canceled waiter may have been

    removed from the list by set() before its await_resume runs.

    removed from the list by set() before its await_resume runs.

    This requires a separate in_list_ flag (unlike async_mutex where

    This requires a separate in_list_ flag (unlike async_mutex where

    active_ served double duty). await_resume only calls remove()

    active_ served double duty). await_resume only calls remove()

    when in_list_ is true.

    when in_list_ is true.

*/

*/

namespace boost {

namespace boost {

namespace capy {

namespace capy {

/** An asynchronous event for coroutines.

/** An asynchronous event for coroutines.

    This event provides a way to notify multiple coroutines that some

    This event provides a way to notify multiple coroutines that some

    condition has occurred. When a coroutine awaits an unset event, it

    condition has occurred. When a coroutine awaits an unset event, it

    suspends and is added to a wait queue. When the event is set, all

    suspends and is added to a wait queue. When the event is set, all

    waiting coroutines are resumed.

    waiting coroutines are resumed.

    @par Cancellation

    @par Cancellation

    When a coroutine is suspended waiting for the event and its stop

    When a coroutine is suspended waiting for the event and its stop

    token is triggered, the waiter completes with `error::canceled`

    token is triggered, the waiter completes with `error::canceled`

    instead of waiting for `set()`.

    instead of waiting for `set()`.

    Cancellation only applies while the coroutine is suspended in the

    Cancellation only applies while the coroutine is suspended in the

    wait queue. If the event is already set when `wait()` is called,

    wait queue. If the event is already set when `wait()` is called,

    the wait completes immediately even if the stop token is already

    the wait completes immediately even if the stop token is already

    signaled.

    signaled.

    @par Zero Allocation

    @par Zero Allocation

    No heap allocation occurs for wait operations.

    No heap allocation occurs for wait operations.

    @par Thread Safety

    @par Thread Safety

    The event operations are designed for single-threaded use on one

    The event operations are designed for single-threaded use on one

    executor. The stop callback may fire from any thread.

    executor. The stop callback may fire from any thread.

    @par Example

    @par Example

    @code

    @code

    async_event event;

    async_event event;

    task<> waiter() {

    task<> waiter() {

        auto [ec] = co_await event.wait();

        auto [ec] = co_await event.wait();

        if(ec)

        if(ec)

            co_return;

            co_return;

        // ... event was set ...

        // ... event was set ...

    }

    }

    task<> notifier() {

    task<> notifier() {

        // ... do some work ...

        // ... do some work ...

        event.set();  // Wake all waiters

        event.set();  // Wake all waiters

    }

    }

    @endcode

    @endcode

*/

*/

class async_event

class async_event

{

{

public:

public:

    class wait_awaiter;

    class wait_awaiter;

private:

private:

    bool set_ = false;

    bool set_ = false;

    detail::intrusive_list<wait_awaiter> waiters_;

    detail::intrusive_list<wait_awaiter> waiters_;

public:

public:

    /** Awaiter returned by wait().

    /** Awaiter returned by wait().

    */

    */

    class wait_awaiter

    class wait_awaiter

        : public detail::intrusive_list<wait_awaiter>::node

        : public detail::intrusive_list<wait_awaiter>::node

    {

    {

        friend class async_event;

        friend class async_event;

        async_event* e_;

        async_event* e_;

        std::coroutine_handle<> h_;

        std::coroutine_handle<> h_;

        executor_ref ex_;

        executor_ref ex_;

        // Declared before stop_cb_buf_: the callback

        // Declared before stop_cb_buf_: the callback

        // accesses these members, so they must still be

        // accesses these members, so they must still be

        // alive if the stop_cb_ destructor blocks.

        // alive if the stop_cb_ destructor blocks.

        std::atomic<bool> claimed_{false};

        std::atomic<bool> claimed_{false};

        bool canceled_ = false;

        bool canceled_ = false;

        bool active_ = false;

        bool active_ = false;

        bool in_list_ = false;

        bool in_list_ = false;

        struct cancel_fn

        struct cancel_fn

        {

        {

            wait_awaiter* self_;

            wait_awaiter* self_;

            {

            {

                    true, std::memory_order_acq_rel))

                    true, std::memory_order_acq_rel))

                {

                {

                }

                }

        };

        };

        using stop_cb_t =

        using stop_cb_t =

            std::stop_callback<cancel_fn>;

            std::stop_callback<cancel_fn>;

        // Aligned storage for stop_cb_t. Declared last:

        // Aligned storage for stop_cb_t. Declared last:

        // its destructor may block while the callback

        // its destructor may block while the callback

        // accesses the members above.

        // accesses the members above.

#ifdef _MSC_VER

#ifdef _MSC_VER

# pragma warning(push)

# pragma warning(push)

# pragma warning(disable: 4324) // padded due to alignas

# pragma warning(disable: 4324) // padded due to alignas

#endif

#endif

        alignas(stop_cb_t)

        alignas(stop_cb_t)

            unsigned char stop_cb_buf_[sizeof(stop_cb_t)];

            unsigned char stop_cb_buf_[sizeof(stop_cb_t)];

#ifdef _MSC_VER

#ifdef _MSC_VER

# pragma warning(pop)

# pragma warning(pop)

#endif

#endif

        {

        {

            return *reinterpret_cast<stop_cb_t*>(

            return *reinterpret_cast<stop_cb_t*>(

        }

        }

    public:

    public:

        {

        {

        {

        {

                std::memory_order_relaxed))

                std::memory_order_relaxed))

        {

        {

        wait_awaiter(wait_awaiter const&) = delete;

        wait_awaiter(wait_awaiter const&) = delete;

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

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

        wait_awaiter& operator=(wait_awaiter&&) = delete;

        wait_awaiter& operator=(wait_awaiter&&) = delete;

        {

        {

        }

        }

        /** IoAwaitable protocol overload. */

        /** IoAwaitable protocol overload. */

        std::coroutine_handle<>

        std::coroutine_handle<>

            std::coroutine_handle<> h,

            std::coroutine_handle<> h,

            executor_ref ex,

            executor_ref ex,

            std::stop_token token) noexcept

            std::stop_token token) noexcept

        {

        {

            {

            {

            }

            }

        }

        }

        {

        {

            {

            {

            }

            }

            {

            {

                {

                {

                }

                }

            }

            }

        }

        }

    };

    };

    // Non-copyable, non-movable

    // Non-copyable, non-movable

    async_event(async_event const&) = delete;

    async_event(async_event const&) = delete;

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

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

    /** Returns an awaiter that waits until the event is set.

    /** Returns an awaiter that waits until the event is set.

        If the event is already set, completes immediately.

        If the event is already set, completes immediately.

        @return An awaitable yielding `(error_code)`.

        @return An awaitable yielding `(error_code)`.

    */

    */

    {

    {

    }

    }

    /** Sets the event.

    /** Sets the event.

        All waiting coroutines are resumed. Canceled waiters

        All waiting coroutines are resumed. Canceled waiters

        are skipped. Subsequent calls to wait() complete

        are skipped. Subsequent calls to wait() complete

        immediately until clear() is called.

        immediately until clear() is called.

    */

    */

    {

    {

        for(;;)

        for(;;)

        {

        {

                true, std::memory_order_acq_rel))

                true, std::memory_order_acq_rel))

            {

            {

            }

            }

    /** Clears the event.

    /** Clears the event.

        Subsequent calls to wait() will suspend until

        Subsequent calls to wait() will suspend until

        set() is called again.

        set() is called again.

    */

    */

    {

    {

    /** Returns true if the event is currently set.

    /** Returns true if the event is currently set.

    */

    */

    {

    {

    }

    }

};

};

} // namespace capy

} // namespace capy

} // namespace boost

} // namespace boost

#endif

#endif