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_IO_ANY_READ_STREAM_HPP

#ifndef BOOST_CAPY_IO_ANY_READ_STREAM_HPP

#define BOOST_CAPY_IO_ANY_READ_STREAM_HPP

#define BOOST_CAPY_IO_ANY_READ_STREAM_HPP

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

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

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

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

#include <boost/capy/buffers.hpp>

#include <boost/capy/buffers.hpp>

#include <boost/capy/buffers/buffer_array.hpp>

#include <boost/capy/buffers/buffer_array.hpp>

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

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

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

#include <boost/capy/concept/read_stream.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 <concepts>

#include <concepts>

#include <coroutine>

#include <coroutine>

#include <cstddef>

#include <cstddef>

#include <exception>

#include <exception>

#include <new>

#include <new>

#include <span>

#include <span>

#include <stop_token>

#include <stop_token>

#include <system_error>

#include <system_error>

#include <utility>

#include <utility>

namespace boost {

namespace boost {

namespace capy {

namespace capy {

/** Type-erased wrapper for any ReadStream.

/** Type-erased wrapper for any ReadStream.

    This class provides type erasure for any type satisfying the

    This class provides type erasure for any type satisfying the

    @ref ReadStream concept, enabling runtime polymorphism for

    @ref ReadStream concept, enabling runtime polymorphism for

    read operations. It uses cached awaitable storage to achieve

    read operations. It uses cached awaitable storage to achieve

    zero steady-state allocation after construction.

    zero steady-state allocation after construction.

    The wrapper supports two construction modes:

    The wrapper supports two construction modes:

    - **Owning**: Pass by value to transfer ownership. The wrapper

    - **Owning**: Pass by value to transfer ownership. The wrapper

      allocates storage and owns the stream.

      allocates storage and owns the stream.

    - **Reference**: Pass a pointer to wrap without ownership. The

    - **Reference**: Pass a pointer to wrap without ownership. The

      pointed-to stream must outlive this wrapper.

      pointed-to stream must outlive this wrapper.

    @par Awaitable Preallocation

    @par Awaitable Preallocation

    The constructor preallocates storage for the type-erased awaitable.

    The constructor preallocates storage for the type-erased awaitable.

    This reserves all virtual address space at server startup

    This reserves all virtual address space at server startup

    so memory usage can be measured up front, rather than

    so memory usage can be measured up front, rather than

    allocating piecemeal as traffic arrives.

    allocating piecemeal as traffic arrives.

    @par Immediate Completion

    @par Immediate Completion

    When the underlying stream's awaitable reports ready immediately

    When the underlying stream's awaitable reports ready immediately

    (e.g. buffered data already available), the wrapper skips

    (e.g. buffered data already available), the wrapper skips

    coroutine suspension entirely and returns the result inline.

    coroutine suspension entirely and returns the result inline.

    @par Thread Safety

    @par Thread Safety

    Not thread-safe. Concurrent operations on the same wrapper

    Not thread-safe. Concurrent operations on the same wrapper

    are undefined behavior.

    are undefined behavior.

    @par Example

    @par Example

    @code

    @code

    // Owning - takes ownership of the stream

    // Owning - takes ownership of the stream

    any_read_stream stream(socket{ioc});

    any_read_stream stream(socket{ioc});

    // Reference - wraps without ownership

    // Reference - wraps without ownership

    socket sock(ioc);

    socket sock(ioc);

    any_read_stream stream(&sock);

    any_read_stream stream(&sock);

    mutable_buffer buf(data, size);

    mutable_buffer buf(data, size);

    auto [ec, n] = co_await stream.read_some(buf);

    auto [ec, n] = co_await stream.read_some(buf);

    @endcode

    @endcode

    @see any_write_stream, any_stream, ReadStream

    @see any_write_stream, any_stream, ReadStream

*/

*/

class any_read_stream

class any_read_stream

{

{

    struct vtable;

    struct vtable;

    template<ReadStream S>

    template<ReadStream S>

    struct vtable_for_impl;

    struct vtable_for_impl;

    // ordered for cache line coherence

    // ordered for cache line coherence

    void* stream_ = nullptr;

    void* stream_ = nullptr;

    vtable const* vt_ = nullptr;

    vtable const* vt_ = nullptr;

    void* cached_awaitable_ = nullptr;

    void* cached_awaitable_ = nullptr;

    void* storage_ = nullptr;

    void* storage_ = nullptr;

    bool awaitable_active_ = false;

    bool awaitable_active_ = false;

public:

public:

    /** Destructor.

    /** Destructor.

        Destroys the owned stream (if any) and releases the cached

        Destroys the owned stream (if any) and releases the cached

        awaitable storage.

        awaitable storage.

    */

    */

    ~any_read_stream();

    ~any_read_stream();

    /** Construct a default instance.

    /** Construct a default instance.

        Constructs an empty wrapper. Operations on a default-constructed

        Constructs an empty wrapper. Operations on a default-constructed

        wrapper result in undefined behavior.

        wrapper result in undefined behavior.

    */

    */

    /** Non-copyable.

    /** Non-copyable.

        The awaitable cache is per-instance and cannot be shared.

        The awaitable cache is per-instance and cannot be shared.

    */

    */

    any_read_stream(any_read_stream const&) = delete;

    any_read_stream(any_read_stream const&) = delete;

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

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

    /** Construct by moving.

    /** Construct by moving.

        Transfers ownership of the wrapped stream (if owned) and

        Transfers ownership of the wrapped stream (if owned) and

        cached awaitable storage from `other`. After the move, `other` is

        cached awaitable storage from `other`. After the move, `other` is

        in a default-constructed state.

        in a default-constructed state.

        @param other The wrapper to move from.

        @param other The wrapper to move from.

    */

    */

    {

    {

    /** Assign by moving.

    /** Assign by moving.

        Destroys any owned stream and releases existing resources,

        Destroys any owned stream and releases existing resources,

        then transfers ownership from `other`.

        then transfers ownership from `other`.

        @param other The wrapper to move from.

        @param other The wrapper to move from.

        @return Reference to this wrapper.

        @return Reference to this wrapper.

    */

    */

    any_read_stream&

    any_read_stream&

    operator=(any_read_stream&& other) noexcept;

    operator=(any_read_stream&& other) noexcept;

    /** Construct by taking ownership of a ReadStream.

    /** Construct by taking ownership of a ReadStream.

        Allocates storage and moves the stream into this wrapper.

        Allocates storage and moves the stream into this wrapper.

        The wrapper owns the stream and will destroy it.

        The wrapper owns the stream and will destroy it.

        @param s The stream to take ownership of.

        @param s The stream to take ownership of.

    */

    */

    template<ReadStream S>

    template<ReadStream S>

        requires (!std::same_as<std::decay_t<S>, any_read_stream>)

        requires (!std::same_as<std::decay_t<S>, any_read_stream>)

    any_read_stream(S s);

    any_read_stream(S s);

    /** Construct by wrapping a ReadStream without ownership.

    /** Construct by wrapping a ReadStream without ownership.

        Wraps the given stream by pointer. The stream must remain

        Wraps the given stream by pointer. The stream must remain

        valid for the lifetime of this wrapper.

        valid for the lifetime of this wrapper.

        @param s Pointer to the stream to wrap.

        @param s Pointer to the stream to wrap.

    */

    */

    template<ReadStream S>

    template<ReadStream S>

    any_read_stream(S* s);

    any_read_stream(S* s);

    /** Check if the wrapper contains a valid stream.

    /** Check if the wrapper contains a valid stream.

        @return `true` if wrapping a stream, `false` if default-constructed

        @return `true` if wrapping a stream, `false` if default-constructed

            or moved-from.

            or moved-from.

    */

    */

    bool

    bool

    {

    {

    }

    }

    /** Check if the wrapper contains a valid stream.

    /** Check if the wrapper contains a valid stream.

        @return `true` if wrapping a stream, `false` if default-constructed

        @return `true` if wrapping a stream, `false` if default-constructed

            or moved-from.

            or moved-from.

    */

    */

    explicit

    explicit

    {

    {

    }

    }

    /** Initiate an asynchronous read operation.

    /** Initiate an asynchronous read operation.

        Reads data into the provided buffer sequence. The operation

        Reads data into the provided buffer sequence. The operation

        completes when at least one byte has been read, or an error

        completes when at least one byte has been read, or an error

        occurs.

        occurs.

        @param buffers The buffer sequence to read into. Passed by

        @param buffers The buffer sequence to read into. Passed by

            value to ensure the sequence lives in the coroutine frame

            value to ensure the sequence lives in the coroutine frame

            across suspension points.

            across suspension points.

        @return An awaitable yielding `(error_code,std::size_t)`.

        @return An awaitable yielding `(error_code,std::size_t)`.

        @par Immediate Completion

        @par Immediate Completion

        The operation completes immediately without suspending

        The operation completes immediately without suspending

        the calling coroutine when the underlying stream's

        the calling coroutine when the underlying stream's

        awaitable reports immediate readiness via `await_ready`.

        awaitable reports immediate readiness via `await_ready`.

        @note This is a partial operation and may not process the

        @note This is a partial operation and may not process the

        entire buffer sequence. Use the composed @ref read algorithm

        entire buffer sequence. Use the composed @ref read algorithm

        for guaranteed complete transfer.

        for guaranteed complete transfer.

        @par Preconditions

        @par Preconditions

        The wrapper must contain a valid stream (`has_value() == true`).

        The wrapper must contain a valid stream (`has_value() == true`).

        The caller must not call this function again after a prior

        The caller must not call this function again after a prior

        call returned an error (including EOF).

        call returned an error (including EOF).

    */

    */

    template<MutableBufferSequence MB>

    template<MutableBufferSequence MB>

    auto

    auto

    read_some(MB buffers);

    read_some(MB buffers);

protected:

protected:

    /** Rebind to a new stream after move.

    /** Rebind to a new stream after move.

        Updates the internal pointer to reference a new stream object.

        Updates the internal pointer to reference a new stream object.

        Used by owning wrappers after move assignment when the owned

        Used by owning wrappers after move assignment when the owned

        object has moved to a new location.

        object has moved to a new location.

        @param new_stream The new stream to bind to. Must be the same

        @param new_stream The new stream to bind to. Must be the same

            type as the original stream.

            type as the original stream.

        @note Terminates if called with a stream of different type

        @note Terminates if called with a stream of different type

            than the original.

            than the original.

    */

    */

    template<ReadStream S>

    template<ReadStream S>

    void

    void

    rebind(S& new_stream) noexcept

    rebind(S& new_stream) noexcept

    {

    {

        if(vt_ != &vtable_for_impl<S>::value)

        if(vt_ != &vtable_for_impl<S>::value)

            std::terminate();

            std::terminate();

        stream_ = &new_stream;

        stream_ = &new_stream;

    }

    }

};

};

struct any_read_stream::vtable

struct any_read_stream::vtable

{

{

    // ordered by call frequency for cache line coherence

    // ordered by call frequency for cache line coherence

    void (*construct_awaitable)(

    void (*construct_awaitable)(

        void* stream,

        void* stream,

        void* storage,

        void* storage,

        std::span<mutable_buffer const> buffers);

        std::span<mutable_buffer const> buffers);

    bool (*await_ready)(void*);

    bool (*await_ready)(void*);

    std::coroutine_handle<> (*await_suspend)(void*, std::coroutine_handle<>, io_env const*);

    std::coroutine_handle<> (*await_suspend)(void*, std::coroutine_handle<>, io_env const*);

    io_result<std::size_t> (*await_resume)(void*);

    io_result<std::size_t> (*await_resume)(void*);

    void (*destroy_awaitable)(void*) noexcept;

    void (*destroy_awaitable)(void*) noexcept;

    std::size_t awaitable_size;

    std::size_t awaitable_size;

    std::size_t awaitable_align;

    std::size_t awaitable_align;

    void (*destroy)(void*) noexcept;

    void (*destroy)(void*) noexcept;

};

};

template<ReadStream S>

template<ReadStream S>

struct any_read_stream::vtable_for_impl

struct any_read_stream::vtable_for_impl

{

{

    using Awaitable = decltype(std::declval<S&>().read_some(

    using Awaitable = decltype(std::declval<S&>().read_some(

        std::span<mutable_buffer const>{}));

        std::span<mutable_buffer const>{}));

    static void

    static void

    {

    {

    static void

    static void

        void* stream,

        void* stream,

        void* storage,

        void* storage,

        std::span<mutable_buffer const> buffers)

        std::span<mutable_buffer const> buffers)

    {

    {

    static constexpr vtable value = {

    static constexpr vtable value = {

        &construct_awaitable_impl,

        &construct_awaitable_impl,

        },

        },

        },

        },

        },

        },

        },

        },

        sizeof(Awaitable),

        sizeof(Awaitable),

        alignof(Awaitable),

        alignof(Awaitable),

        &do_destroy_impl

        &do_destroy_impl

    };

    };

};

};

inline

inline

{

{

    {

    {

    }

    }

    {

    {

    }

    }

inline any_read_stream&

inline any_read_stream&

{

{

    {

    {

        {

        {

        }

        }

        {

        {

        }

        }

    }

    }

}

}

template<ReadStream S>

template<ReadStream S>

    requires (!std::same_as<std::decay_t<S>, any_read_stream>)

    requires (!std::same_as<std::decay_t<S>, any_read_stream>)

{

{

    struct guard {

    struct guard {

        any_read_stream* self;

        any_read_stream* self;

        bool committed = false;

        bool committed = false;

            }

            }

    // Preallocate the awaitable storage

    // Preallocate the awaitable storage

template<ReadStream S>

template<ReadStream S>

{

{

    // Preallocate the awaitable storage

    // Preallocate the awaitable storage

template<MutableBufferSequence MB>

template<MutableBufferSequence MB>

auto

auto

{

{

    // VFALCO in theory, we could use if constexpr to detect a

    // VFALCO in theory, we could use if constexpr to detect a

    // span and then pass that through to read_some without the array

    // span and then pass that through to read_some without the array

    struct awaitable

    struct awaitable

    {

    {

        any_read_stream* self_;

        any_read_stream* self_;

        mutable_buffer_array<detail::max_iovec_> ba_;

        mutable_buffer_array<detail::max_iovec_> ba_;

        bool

        bool

        {

        {

        }

        }

        std::coroutine_handle<>

        std::coroutine_handle<>

        {

        {

        }

        }

        io_result<std::size_t>

        io_result<std::size_t>

        {

        {

            struct guard {

            struct guard {

                any_read_stream* self;

                any_read_stream* self;

    };

    };

    return awaitable{this,

    return awaitable{this,

} // namespace capy

} // namespace capy

} // namespace boost

} // namespace boost

#endif

#endif