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_MUTEX_HPP

#ifndef BOOST_CAPY_ASYNC_MUTEX_HPP

#define BOOST_CAPY_ASYNC_MUTEX_HPP

#define BOOST_CAPY_ASYNC_MUTEX_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/error.hpp>

#include <boost/capy/error.hpp>

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

#include <boost/capy/ex/io_env.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_mutex implementation notes

/*  async_mutex implementation notes

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

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

    Waiters form a doubly-linked intrusive list (fair FIFO). lock_awaiter

    Waiters form a doubly-linked intrusive list (fair FIFO). lock_awaiter

    inherits intrusive_list<lock_awaiter>::node; the list is owned by

    inherits intrusive_list<lock_awaiter>::node; the list is owned by

    async_mutex::waiters_.

    async_mutex::waiters_.

    Cancellation via stop_token

    Cancellation via stop_token

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

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

    A std::stop_callback is registered in await_suspend. Two actors can

    A std::stop_callback is registered in await_suspend. Two actors can

    race to resume the suspended coroutine: unlock() and the stop callback.

    race to resume the suspended coroutine: unlock() and the stop callback.

    An atomic bool `claimed_` resolves the race -- whoever does

    An atomic bool `claimed_` resolves the race -- whoever does

    claimed_.exchange(true) and reads false wins. The loser does nothing.

    claimed_.exchange(true) and reads false wins. The loser does nothing.

    The stop callback calls ex_.dispatch(h_). If dispatch runs inline

    The stop callback calls ex_.dispatch(h_). If dispatch runs inline

    (same thread), the stop_callback is destroyed from within its own

    (same thread), the stop_callback is destroyed from within its own

    operator() via await_resume. This is safe: cancel_fn touches no

    operator() via await_resume. This is safe: cancel_fn touches no

    members after dispatch returns (same pattern as delete-this).

    members after dispatch returns (same pattern as delete-this).

    unlock() pops waiters from the front. If the popped waiter was

    unlock() pops waiters from the front. If the popped waiter was

    already claimed by the stop callback, unlock() skips it and tries

    already claimed by the stop callback, unlock() skips it and tries

    the next. await_resume removes the (still-linked) canceled waiter

    the next. await_resume removes the (still-linked) canceled waiter

    via waiters_.remove(this).

    via waiters_.remove(this).

    The stop_callback lives in a union to suppress automatic

    The stop_callback lives in a union to suppress automatic

    construction/destruction. Placement new in await_suspend, explicit

    construction/destruction. Placement new in await_suspend, explicit

    destructor call in await_resume and ~lock_awaiter.

    destructor call in await_resume and ~lock_awaiter.

    Member ordering constraint

    Member ordering constraint

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

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

    The union containing stop_cb_ must be declared AFTER the members

    The union containing stop_cb_ must be declared AFTER the members

    the callback accesses (h_, ex_, claimed_, canceled_). If the

    the callback accesses (h_, ex_, claimed_, canceled_). If the

    stop_cb_ destructor blocks waiting for a concurrent callback, those

    stop_cb_ destructor blocks waiting for a concurrent callback, those

    members must still be alive (C++ destroys in reverse declaration

    members must still be alive (C++ destroys in reverse declaration

    order).

    order).

    active_ flag

    active_ flag

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

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

    Tracks both list membership and stop_cb_ lifetime (they are always

    Tracks both list membership and stop_cb_ lifetime (they are always

    set and cleared together). Used by the destructor to clean up if the

    set and cleared together). Used by the destructor to clean up if the

    coroutine is destroyed while suspended (e.g. execution_context

    coroutine is destroyed while suspended (e.g. execution_context

    shutdown).

    shutdown).

    Cancellation scope

    Cancellation scope

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

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

    Cancellation only takes effect while the coroutine is suspended in

    Cancellation only takes effect while the coroutine is suspended in

    the wait queue. If the mutex is unlocked, await_ready acquires it

    the wait queue. If the mutex is unlocked, await_ready acquires it

    immediately without checking the stop token. This is intentional:

    immediately without checking the stop token. This is intentional:

    the fast path has no token access and no overhead.

    the fast path has no token access and no overhead.

    Threading assumptions

    Threading assumptions

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

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

    - All list mutations happen on the executor thread (await_suspend,

    - All list mutations happen on the executor thread (await_suspend,

      await_resume, unlock, ~lock_awaiter).

      await_resume, unlock, ~lock_awaiter).

    - The stop callback may fire from any thread, but only touches

    - The stop callback may fire from any thread, but only touches

      claimed_ (atomic) and then calls dispatch. It never touches the

      claimed_ (atomic) and then calls dispatch. It never touches the

      list.

      list.

    - ~lock_awaiter must be called from the executor thread. This is

    - ~lock_awaiter must be called from the executor thread. This is

      guaranteed during normal shutdown but NOT if the coroutine frame

      guaranteed during normal shutdown but NOT if the coroutine frame

      is destroyed from another thread while a stop callback could

      is destroyed from another thread while a stop callback could

      fire (precondition violation, same as cppcoro/folly).

      fire (precondition violation, same as cppcoro/folly).

*/

*/

namespace boost {

namespace boost {

namespace capy {

namespace capy {

/** An asynchronous mutex for coroutines.

/** An asynchronous mutex for coroutines.

    This mutex provides mutual exclusion for coroutines without blocking.

    This mutex provides mutual exclusion for coroutines without blocking.

    When a coroutine attempts to acquire a locked mutex, it suspends and

    When a coroutine attempts to acquire a locked mutex, it suspends and

    is added to an intrusive wait queue. When the holder unlocks, the next

    is added to an intrusive wait queue. When the holder unlocks, the next

    waiter is resumed with the lock held.

    waiter is resumed with the lock held.

    @par Cancellation

    @par Cancellation

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

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

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

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

    instead of acquiring the lock.

    instead of acquiring the lock.

    Cancellation only applies while the coroutine is suspended in the

    Cancellation only applies while the coroutine is suspended in the

    wait queue. If the mutex is unlocked when `lock()` is called, the

    wait queue. If the mutex is unlocked when `lock()` is called, the

    lock is acquired immediately even if the stop token is already

    lock is acquired immediately even if the stop token is already

    signaled.

    signaled.

    @par Zero Allocation

    @par Zero Allocation

    No heap allocation occurs for lock operations.

    No heap allocation occurs for lock operations.

    @par Thread Safety

    @par Thread Safety

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

    The mutex 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_mutex cm;

    async_mutex cm;

    task<> protected_operation() {

    task<> protected_operation() {

        auto [ec] = co_await cm.lock();

        auto [ec] = co_await cm.lock();

        if(ec)

        if(ec)

            co_return;

            co_return;

        // ... critical section ...

        // ... critical section ...

        cm.unlock();

        cm.unlock();

    }

    }

    // Or with RAII:

    // Or with RAII:

    task<> protected_operation() {

    task<> protected_operation() {

        auto [ec, guard] = co_await cm.scoped_lock();

        auto [ec, guard] = co_await cm.scoped_lock();

        if(ec)

        if(ec)

            co_return;

            co_return;

        // ... critical section ...

        // ... critical section ...

        // unlocks automatically

        // unlocks automatically

    }

    }

    @endcode

    @endcode

*/

*/

class async_mutex

class async_mutex

{

{

public:

public:

    class lock_awaiter;

    class lock_awaiter;

    class lock_guard;

    class lock_guard;

    class lock_guard_awaiter;

    class lock_guard_awaiter;

private:

private:

    bool locked_ = false;

    bool locked_ = false;

    detail::intrusive_list<lock_awaiter> waiters_;

    detail::intrusive_list<lock_awaiter> waiters_;

public:

public:

    /** Awaiter returned by lock().

    /** Awaiter returned by lock().

    */

    */

    class lock_awaiter

    class lock_awaiter

        : public detail::intrusive_list<lock_awaiter>::node

        : public detail::intrusive_list<lock_awaiter>::node

    {

    {

        friend class async_mutex;

        friend class async_mutex;

        async_mutex* m_;

        async_mutex* m_;

        std::coroutine_handle<> h_;

        std::coroutine_handle<> h_;

        executor_ref ex_;

        executor_ref ex_;

        // These members must be declared before stop_cb_

        // These members must be declared before stop_cb_

        // (see comment on the union below).

        // (see comment on the union below).

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

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

        bool canceled_ = false;

        bool canceled_ = false;

        bool active_ = false;

        bool active_ = false;

        struct cancel_fn

        struct cancel_fn

        {

        {

            lock_awaiter* self_;

            lock_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))

        {

        {

        lock_awaiter(lock_awaiter const&) = delete;

        lock_awaiter(lock_awaiter const&) = delete;

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

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

        lock_awaiter& operator=(lock_awaiter&&) = delete;

        lock_awaiter& operator=(lock_awaiter&&) = delete;

        {

        {

            {

            {

            }

            }

        }

        }

        /** IoAwaitable protocol overload. */

        /** IoAwaitable protocol overload. */

        std::coroutine_handle<>

        std::coroutine_handle<>

            std::coroutine_handle<> h,

            std::coroutine_handle<> h,

            io_env const& env) noexcept

            io_env const* env) noexcept

        {

        {

            {

            {

            }

            }

        }

        }

        {

        {

            {

            {

                {

                {

                }

                }

            }

            }

        }

        }

    };

    };

    /** RAII lock guard for async_mutex.

    /** RAII lock guard for async_mutex.

        Automatically unlocks the mutex when destroyed.

        Automatically unlocks the mutex when destroyed.

    */

    */

    class [[nodiscard]] lock_guard

    class [[nodiscard]] lock_guard

    {

    {

        async_mutex* m_;

        async_mutex* m_;

    public:

    public:

        {

        {

        {

        {

        {

        {

        {

        {

        lock_guard& operator=(lock_guard&& o) noexcept

        lock_guard& operator=(lock_guard&& o) noexcept

        {

        {

            if(this != &o)

            if(this != &o)

            {

            {

                if(m_)

                if(m_)

                    m_->unlock();

                    m_->unlock();

                m_ = std::exchange(o.m_, nullptr);

                m_ = std::exchange(o.m_, nullptr);

            }

            }

            return *this;

            return *this;

        }

        }

        lock_guard(lock_guard const&) = delete;

        lock_guard(lock_guard const&) = delete;

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

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

    };

    };

    /** Awaiter returned by scoped_lock() that returns a lock_guard on resume.

    /** Awaiter returned by scoped_lock() that returns a lock_guard on resume.

    */

    */

    class lock_guard_awaiter

    class lock_guard_awaiter

    {

    {

        async_mutex* m_;

        async_mutex* m_;

        lock_awaiter inner_;

        lock_awaiter inner_;

    public:

    public:

        {

        {

        {

        {

        }

        }

        /** IoAwaitable protocol overload. */

        /** IoAwaitable protocol overload. */

        std::coroutine_handle<>

        std::coroutine_handle<>

            std::coroutine_handle<> h,

            std::coroutine_handle<> h,

            io_env const& env) noexcept

            io_env const* env) noexcept

        {

        {

        }

        }

        {

        {

    };

    };

    async_mutex() = default;

    async_mutex() = default;

    // Non-copyable, non-movable

    // Non-copyable, non-movable

    async_mutex(async_mutex const&) = delete;

    async_mutex(async_mutex const&) = delete;

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

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

    /** Returns an awaiter that acquires the mutex.

    /** Returns an awaiter that acquires the mutex.

        @return An awaitable yielding `(error_code)`.

        @return An awaitable yielding `(error_code)`.

    */

    */

    {

    {

    }

    }

    /** Returns an awaiter that acquires the mutex with RAII.

    /** Returns an awaiter that acquires the mutex with RAII.

        @return An awaitable yielding `(error_code,lock_guard)`.

        @return An awaitable yielding `(error_code,lock_guard)`.

    */

    */

    {

    {

    }

    }

    /** Releases the mutex.

    /** Releases the mutex.

        If waiters are queued, the next eligible waiter is

        If waiters are queued, the next eligible waiter is

        resumed with the lock held. Canceled waiters are

        resumed with the lock held. Canceled waiters are

        skipped. If no eligible waiter remains, the mutex

        skipped. If no eligible waiter remains, the mutex

        becomes unlocked.

        becomes unlocked.

    */

    */

    {

    {

        for(;;)

        for(;;)

        {

        {

            {

            {

            }

            }

                true, std::memory_order_acq_rel))

                true, std::memory_order_acq_rel))

            {

            {

            }

            }

    }

    }

    /** Returns true if the mutex is currently locked.

    /** Returns true if the mutex is currently locked.

    */

    */

    {

    {

    }

    }

};

};

} // namespace capy

} // namespace capy

} // namespace boost

} // namespace boost

#endif

#endif