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_SRC_EX_DETAIL_STRAND_QUEUE_HPP

#ifndef BOOST_CAPY_SRC_EX_DETAIL_STRAND_QUEUE_HPP

#define BOOST_CAPY_SRC_EX_DETAIL_STRAND_QUEUE_HPP

#define BOOST_CAPY_SRC_EX_DETAIL_STRAND_QUEUE_HPP

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

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

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

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

#include <coroutine>

#include <coroutine>

#include <cstddef>

#include <cstddef>

#include <exception>

#include <exception>

namespace boost {

namespace boost {

namespace capy {

namespace capy {

namespace detail {

namespace detail {

class strand_queue;

class strand_queue;

//----------------------------------------------------------

//----------------------------------------------------------

// Metadata stored before the coroutine frame

// Metadata stored before the coroutine frame

struct frame_prefix

struct frame_prefix

{

{

    frame_prefix* next;

    frame_prefix* next;

    strand_queue* queue;

    strand_queue* queue;

    std::size_t alloc_size;

    std::size_t alloc_size;

};

};

//----------------------------------------------------------

//----------------------------------------------------------

/** Wrapper coroutine for strand queue dispatch operations.

/** Wrapper coroutine for strand queue dispatch operations.

    This coroutine wraps a target coroutine handle and resumes

    This coroutine wraps a target coroutine handle and resumes

    it when dispatched. The wrapper ensures control returns to

    it when dispatched. The wrapper ensures control returns to

    the dispatch loop after the target suspends or completes.

    the dispatch loop after the target suspends or completes.

    The promise contains an intrusive list node for queue

    The promise contains an intrusive list node for queue

    storage and supports a custom allocator that recycles

    storage and supports a custom allocator that recycles

    coroutine frames via a free list.

    coroutine frames via a free list.

*/

*/

struct strand_op

struct strand_op

{

{

    struct promise_type

    struct promise_type

    {

    {

        promise_type* next = nullptr;

        promise_type* next = nullptr;

        void*

        void*

        operator new(

        operator new(

            std::size_t size,

            std::size_t size,

            strand_queue& q,

            strand_queue& q,

            std::coroutine_handle<void>);

            std::coroutine_handle<void>);

        void

        void

        operator delete(void* p, std::size_t);

        operator delete(void* p, std::size_t);

        strand_op

        strand_op

        {

        {

        }

        }

        std::suspend_always

        std::suspend_always

        {

        {

        }

        }

        std::suspend_always

        std::suspend_always

        {

        {

        }

        }

        void

        void

        {

        {

        void

        void

        unhandled_exception()

        unhandled_exception()

        {

        {

            std::terminate();

            std::terminate();

        }

        }

    };

    };

    std::coroutine_handle<promise_type> h_;

    std::coroutine_handle<promise_type> h_;

};

};

//----------------------------------------------------------

//----------------------------------------------------------

/** Single-threaded dispatch queue for coroutine handles.

/** Single-threaded dispatch queue for coroutine handles.

    This queue stores coroutine handles and resumes them

    This queue stores coroutine handles and resumes them

    sequentially when dispatch() is called. Each pushed

    sequentially when dispatch() is called. Each pushed

    handle is wrapped in a strand_op coroutine that ensures

    handle is wrapped in a strand_op coroutine that ensures

    control returns to the dispatch loop after the target

    control returns to the dispatch loop after the target

    suspends or completes.

    suspends or completes.

    The queue uses an intrusive singly-linked list through

    The queue uses an intrusive singly-linked list through

    the promise type to avoid separate node allocations.

    the promise type to avoid separate node allocations.

    A free list recycles wrapper coroutine frames to reduce

    A free list recycles wrapper coroutine frames to reduce

    allocation overhead during repeated push/dispatch cycles.

    allocation overhead during repeated push/dispatch cycles.

    @par Thread Safety

    @par Thread Safety

    This class is not thread-safe. All operations must be

    This class is not thread-safe. All operations must be

    called from a single thread.

    called from a single thread.

*/

*/

class strand_queue

class strand_queue

{

{

    using promise_type = strand_op::promise_type;

    using promise_type = strand_op::promise_type;

    promise_type* head_ = nullptr;

    promise_type* head_ = nullptr;

    promise_type* tail_ = nullptr;

    promise_type* tail_ = nullptr;

    frame_prefix* free_list_ = nullptr;

    frame_prefix* free_list_ = nullptr;

    friend struct strand_op::promise_type;

    friend struct strand_op::promise_type;

    static

    static

    strand_op

    strand_op

        strand_queue& q,

        strand_queue& q,

        std::coroutine_handle<void> target)

        std::coroutine_handle<void> target)

    {

    {

        (void)q;

        (void)q;

        safe_resume(target);

        safe_resume(target);

        co_return;

        co_return;

public:

public:

    strand_queue(strand_queue const&) = delete;

    strand_queue(strand_queue const&) = delete;

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

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

    /** Destructor.

    /** Destructor.

        Destroys any pending wrappers without resuming them,

        Destroys any pending wrappers without resuming them,

        then frees all memory in the free list.

        then frees all memory in the free list.

    */

    */

    {

    {

        // Destroy pending wrappers

        // Destroy pending wrappers

        {

        {

        }

        }

        // Free the free list memory

        // Free the free list memory

        {

        {

        }

        }

    /** Returns true if there are no pending operations.

    /** Returns true if there are no pending operations.

    */

    */

    bool

    bool

    {

    {

    }

    }

    /** Push a coroutine handle to the queue.

    /** Push a coroutine handle to the queue.

        Creates a wrapper coroutine and appends it to the

        Creates a wrapper coroutine and appends it to the

        queue. The wrapper will resume the target handle

        queue. The wrapper will resume the target handle

        when dispatch() processes it.

        when dispatch() processes it.

        @param h The coroutine handle to dispatch.

        @param h The coroutine handle to dispatch.

    */

    */

    void

    void

    {

    {

        else

        else

    /** Resume all queued coroutines in sequence.

    /** Resume all queued coroutines in sequence.

        Processes each wrapper in FIFO order, resuming its

        Processes each wrapper in FIFO order, resuming its

        target coroutine. After each target suspends or

        target coroutine. After each target suspends or

        completes, the wrapper is destroyed and its frame

        completes, the wrapper is destroyed and its frame

        is added to the free list for reuse.

        is added to the free list for reuse.

        Coroutines resumed during dispatch may push new

        Coroutines resumed during dispatch may push new

        handles, which will also be processed in the same

        handles, which will also be processed in the same

        dispatch call.

        dispatch call.

        @warning Not thread-safe. Do not call while another

        @warning Not thread-safe. Do not call while another

            thread may be calling push().

            thread may be calling push().

    */

    */

    void

    void

    dispatch()

    dispatch()

    {

    {

        while(head_)

        while(head_)

        {

        {

            promise_type* p = head_;

            promise_type* p = head_;

            head_ = p->next;

            head_ = p->next;

            if(!head_)

            if(!head_)

                tail_ = nullptr;

                tail_ = nullptr;

            auto h = std::coroutine_handle<promise_type>::from_promise(*p);

            auto h = std::coroutine_handle<promise_type>::from_promise(*p);

            safe_resume(h);

            safe_resume(h);

            h.destroy();

            h.destroy();

        }

        }

    }

    }

    /** Batch of taken items for thread-safe dispatch. */

    /** Batch of taken items for thread-safe dispatch. */

    struct taken_batch

    struct taken_batch

    {

    {

        promise_type* head = nullptr;

        promise_type* head = nullptr;

        promise_type* tail = nullptr;

        promise_type* tail = nullptr;

    };

    };

    /** Take all pending items atomically.

    /** Take all pending items atomically.

        Removes all items from the queue and returns them

        Removes all items from the queue and returns them

        as a batch. The queue is left empty.

        as a batch. The queue is left empty.

        @return The batch of taken items.

        @return The batch of taken items.

    */

    */

    taken_batch

    taken_batch

    {

    {

    }

    }

    /** Dispatch a batch of taken items.

    /** Dispatch a batch of taken items.

        @param batch The batch to dispatch.

        @param batch The batch to dispatch.

        @note This is thread-safe w.r.t. push() because it doesn't

        @note This is thread-safe w.r.t. push() because it doesn't

            access the queue's free_list_. Frames are deleted directly

            access the queue's free_list_. Frames are deleted directly

            rather than recycled.

            rather than recycled.

    */

    */

    static

    static

    void

    void

    {

    {

        {

        {

            // Don't use h.destroy() - it would call operator delete which

            // Don't use h.destroy() - it would call operator delete which

            // accesses the queue's free_list_ (race with push).

            // accesses the queue's free_list_ (race with push).

            // Instead, manually free the frame without recycling.

            // Instead, manually free the frame without recycling.

            // h.address() returns the frame base (what operator new returned).

            // h.address() returns the frame base (what operator new returned).

        }

        }

};

};

//----------------------------------------------------------

//----------------------------------------------------------

inline

inline

void*

void*

    std::size_t size,

    std::size_t size,

    strand_queue& q,

    strand_queue& q,

    std::coroutine_handle<void>)

    std::coroutine_handle<void>)

{

{

    // Total size includes prefix

    // Total size includes prefix

    void* raw;

    void* raw;

    // Try to reuse from free list

    // Try to reuse from free list

    {

    {

    }

    }

    else

    else

    {

    {

    }

    }

    // Initialize prefix

    // Initialize prefix

    // Return pointer AFTER the prefix (this is where coroutine frame goes)

    // Return pointer AFTER the prefix (this is where coroutine frame goes)

}

}

inline

inline

void

void

{

{

    // Calculate back to get the prefix

    // Calculate back to get the prefix

    // Add to free list

    // Add to free list

} // namespace detail

} // namespace detail

} // namespace capy

} // namespace capy

} // namespace boost

} // namespace boost

#endif

#endif