Diff coverage report for

//

//

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

// Copyright (c) 2026 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/corosio

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

//

//

#ifndef BOOST_COROSIO_TCP_SERVER_HPP

#ifndef BOOST_COROSIO_TCP_SERVER_HPP

#define BOOST_COROSIO_TCP_SERVER_HPP

#define BOOST_COROSIO_TCP_SERVER_HPP

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

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

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

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

#include <boost/corosio/tcp_acceptor.hpp>

#include <boost/corosio/tcp_acceptor.hpp>

#include <boost/corosio/tcp_socket.hpp>

#include <boost/corosio/tcp_socket.hpp>

#include <boost/corosio/io_context.hpp>

#include <boost/corosio/io_context.hpp>

#include <boost/corosio/endpoint.hpp>

#include <boost/corosio/endpoint.hpp>

#include <boost/capy/task.hpp>

#include <boost/capy/task.hpp>

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

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

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

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

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

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

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

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

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

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

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

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

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

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

#include <coroutine>

#include <coroutine>

#include <memory>

#include <memory>

#include <ranges>

#include <ranges>

#include <vector>

#include <vector>

namespace boost::corosio {

namespace boost::corosio {

#ifdef _MSC_VER

#ifdef _MSC_VER

#pragma warning(push)

#pragma warning(push)

#pragma warning(disable: 4251) // class needs to have dll-interface

#pragma warning(disable: 4251) // class needs to have dll-interface

#endif

#endif

/** TCP server with pooled workers.

/** TCP server with pooled workers.

    This class manages a pool of reusable worker objects that handle

    This class manages a pool of reusable worker objects that handle

    incoming connections. When a connection arrives, an idle worker

    incoming connections. When a connection arrives, an idle worker

    is dispatched to handle it. After the connection completes, the

    is dispatched to handle it. After the connection completes, the

    worker returns to the pool for reuse, avoiding allocation overhead

    worker returns to the pool for reuse, avoiding allocation overhead

    per connection.

    per connection.

    Workers are set via @ref set_workers as a forward range of

    Workers are set via @ref set_workers as a forward range of

    pointer-like objects (e.g., `unique_ptr<worker_base>`). The server

    pointer-like objects (e.g., `unique_ptr<worker_base>`). The server

    takes ownership of the container via type erasure.

    takes ownership of the container via type erasure.

    @par Thread Safety

    @par Thread Safety

    Distinct objects: Safe.

    Distinct objects: Safe.

    Shared objects: Unsafe.

    Shared objects: Unsafe.

    @par Lifecycle

    @par Lifecycle

    The server operates in three states:

    The server operates in three states:

    - **Stopped**: Initial state, or after @ref join completes.

    - **Stopped**: Initial state, or after @ref join completes.

    - **Running**: After @ref start, actively accepting connections.

    - **Running**: After @ref start, actively accepting connections.

    - **Stopping**: After @ref stop, draining active work.

    - **Stopping**: After @ref stop, draining active work.

    State transitions:

    State transitions:

    @code

    @code

    [Stopped] --start()--> [Running] --stop()--> [Stopping] --join()--> [Stopped]

    [Stopped] --start()--> [Running] --stop()--> [Stopping] --join()--> [Stopped]

    @endcode

    @endcode

    @par Running the Server

    @par Running the Server

    @code

    @code

    io_context ioc;

    io_context ioc;

    tcp_server srv(ioc, ioc.get_executor());

    tcp_server srv(ioc, ioc.get_executor());

    srv.set_workers(make_workers(ioc, 100));

    srv.set_workers(make_workers(ioc, 100));

    srv.bind(endpoint{address_v4::any(), 8080});

    srv.bind(endpoint{address_v4::any(), 8080});

    srv.start();

    srv.start();

    ioc.run();  // Blocks until all work completes

    ioc.run();  // Blocks until all work completes

    @endcode

    @endcode

    @par Graceful Shutdown

    @par Graceful Shutdown

    To shut down gracefully, call @ref stop then drain the io_context:

    To shut down gracefully, call @ref stop then drain the io_context:

    @code

    @code

    // From a signal handler or timer callback:

    // From a signal handler or timer callback:

    srv.stop();

    srv.stop();

    // ioc.run() returns after pending work drains.

    // ioc.run() returns after pending work drains.

    // Then from the thread that called ioc.run():

    // Then from the thread that called ioc.run():

    srv.join();  // Wait for accept loops to finish

    srv.join();  // Wait for accept loops to finish

    @endcode

    @endcode

    @par Restart After Stop

    @par Restart After Stop

    The server can be restarted after a complete shutdown cycle.

    The server can be restarted after a complete shutdown cycle.

    You must drain the io_context and call @ref join before restarting:

    You must drain the io_context and call @ref join before restarting:

    @code

    @code

    srv.start();

    srv.start();

    ioc.run_for( 10s );   // Run for a while

    ioc.run_for( 10s );   // Run for a while

    srv.stop();           // Signal shutdown

    srv.stop();           // Signal shutdown

    ioc.run();            // REQUIRED: drain pending completions

    ioc.run();            // REQUIRED: drain pending completions

    srv.join();           // REQUIRED: wait for accept loops

    srv.join();           // REQUIRED: wait for accept loops

    // Now safe to restart

    // Now safe to restart

    srv.start();

    srv.start();

    ioc.run();

    ioc.run();

    @endcode

    @endcode

    @par WARNING: What NOT to Do

    @par WARNING: What NOT to Do

    - Do NOT call @ref join from inside a worker coroutine (deadlock).

    - Do NOT call @ref join from inside a worker coroutine (deadlock).

    - Do NOT call @ref join from a thread running `ioc.run()` (deadlock).

    - Do NOT call @ref join from a thread running `ioc.run()` (deadlock).

    - Do NOT call @ref start without completing @ref join after @ref stop.

    - Do NOT call @ref start without completing @ref join after @ref stop.

    - Do NOT call `ioc.stop()` for graceful shutdown; use @ref stop instead.

    - Do NOT call `ioc.stop()` for graceful shutdown; use @ref stop instead.

    @par Example

    @par Example

    @code

    @code

    class my_worker : public tcp_server::worker_base

    class my_worker : public tcp_server::worker_base

    {

    {

        corosio::tcp_socket sock_;

        corosio::tcp_socket sock_;

        capy::any_executor ex_;

        capy::any_executor ex_;

    public:

    public:

        my_worker(io_context& ctx)

        my_worker(io_context& ctx)

            : sock_(ctx)

            : sock_(ctx)

            , ex_(ctx.get_executor())

            , ex_(ctx.get_executor())

        {

        {

        }

        }

        corosio::tcp_socket& socket() override { return sock_; }

        corosio::tcp_socket& socket() override { return sock_; }

        void run(launcher launch) override

        void run(launcher launch) override

        {

        {

            launch(ex_, [](corosio::tcp_socket* sock) -> capy::task<>

            launch(ex_, [](corosio::tcp_socket* sock) -> capy::task<>

            {

            {

                // handle connection using sock

                // handle connection using sock

                co_return;

                co_return;

            }(&sock_));

            }(&sock_));

        }

        }

    };

    };

    auto make_workers(io_context& ctx, int n)

    auto make_workers(io_context& ctx, int n)

    {

    {

        std::vector<std::unique_ptr<tcp_server::worker_base>> v;

        std::vector<std::unique_ptr<tcp_server::worker_base>> v;

        v.reserve(n);

        v.reserve(n);

        for(int i = 0; i < n; ++i)

        for(int i = 0; i < n; ++i)

            v.push_back(std::make_unique<my_worker>(ctx));

            v.push_back(std::make_unique<my_worker>(ctx));

        return v;

        return v;

    }

    }

    io_context ioc;

    io_context ioc;

    tcp_server srv(ioc, ioc.get_executor());

    tcp_server srv(ioc, ioc.get_executor());

    srv.set_workers(make_workers(ioc, 100));

    srv.set_workers(make_workers(ioc, 100));

    @endcode

    @endcode

    @see worker_base, set_workers, launcher

    @see worker_base, set_workers, launcher

*/

*/

class BOOST_COROSIO_DECL

class BOOST_COROSIO_DECL

    tcp_server

    tcp_server

{

{

public:

public:

    class worker_base;  ///< Abstract base for connection handlers.

    class worker_base;  ///< Abstract base for connection handlers.

    class launcher;     ///< Move-only handle to launch worker coroutines.

    class launcher;     ///< Move-only handle to launch worker coroutines.

private:

private:

    struct waiter

    struct waiter

    {

    {

        waiter* next;

        waiter* next;

        std::coroutine_handle<> h;

        std::coroutine_handle<> h;

        worker_base* w;

        worker_base* w;

    };

    };

    struct impl;

    struct impl;

    static impl* make_impl(capy::execution_context& ctx);

    static impl* make_impl(capy::execution_context& ctx);

    impl* impl_;

    impl* impl_;

    capy::any_executor ex_;

    capy::any_executor ex_;

    waiter* waiters_ = nullptr;

    waiter* waiters_ = nullptr;

    worker_base* idle_head_ = nullptr;    // Forward list: available workers

    worker_base* idle_head_ = nullptr;    // Forward list: available workers

    worker_base* active_head_ = nullptr;  // Doubly linked: workers handling connections

    worker_base* active_head_ = nullptr;  // Doubly linked: workers handling connections

    worker_base* active_tail_ = nullptr;  // Tail for O(1) push_back

    worker_base* active_tail_ = nullptr;  // Tail for O(1) push_back

    std::size_t active_accepts_ = 0;      // Number of active do_accept coroutines

    std::size_t active_accepts_ = 0;      // Number of active do_accept coroutines

    std::shared_ptr<void> storage_;       // Owns the worker container (type-erased)

    std::shared_ptr<void> storage_;       // Owns the worker container (type-erased)

    bool running_ = false;

    bool running_ = false;

    // Idle list (forward/singly linked) - push front, pop front

    // Idle list (forward/singly linked) - push front, pop front

    {

    {

    {

    {

    }

    }

    // Active list (doubly linked) - push back, remove anywhere

    // Active list (doubly linked) - push back, remove anywhere

    {

    {

        else

        else

    {

    {

        // Skip if not in active list (e.g., after failed accept)

        // Skip if not in active list (e.g., after failed accept)

        else

        else

        else

        else

    }

    }

    template<capy::Executor Ex>

    template<capy::Executor Ex>

    struct launch_wrapper

    struct launch_wrapper

    {

    {

        struct promise_type

        struct promise_type

        {

        {

            Ex ex;  // Executor stored directly in frame (outlives child tasks)

            Ex ex;  // Executor stored directly in frame (outlives child tasks)

            capy::io_env env_;

            capy::io_env env_;

            // For regular coroutines: first arg is executor, second is stop token

            // For regular coroutines: first arg is executor, second is stop token

            template<class E, class S, class... Args>

            template<class E, class S, class... Args>

                requires capy::Executor<std::decay_t<E>>

                requires capy::Executor<std::decay_t<E>>

            promise_type(E e, S s, Args&&...)

            promise_type(E e, S s, Args&&...)

                : ex(std::move(e))

                : ex(std::move(e))

                , env_{capy::executor_ref(ex), std::move(s),

                , env_{capy::executor_ref(ex), std::move(s),

                       capy::current_frame_allocator()}

                       capy::current_frame_allocator()}

            {

            {

            }

            }

            // For lambda coroutines: first arg is closure, second is executor, third is stop token

            // For lambda coroutines: first arg is closure, second is executor, third is stop token

            template<class Closure, class E, class S, class... Args>

            template<class Closure, class E, class S, class... Args>

                requires (!capy::Executor<std::decay_t<Closure>> && 

                requires (!capy::Executor<std::decay_t<Closure>> && 

                          capy::Executor<std::decay_t<E>>)

                          capy::Executor<std::decay_t<E>>)

            {

            {

            }

            }

            // Inject io_env for IoAwaitable

            // Inject io_env for IoAwaitable

            template<capy::IoAwaitable Awaitable>

            template<capy::IoAwaitable Awaitable>

            {

            {

                using AwaitableT = std::decay_t<Awaitable>;

                using AwaitableT = std::decay_t<Awaitable>;

                struct adapter

                struct adapter

                {

                {

                    AwaitableT aw;

                    AwaitableT aw;

                    capy::io_env const* env;

                    capy::io_env const* env;

                    {

                    {

                    }

                    }

                };

                };

        };

        };

        std::coroutine_handle<promise_type> h;

        std::coroutine_handle<promise_type> h;

        {

        {

        {

        {

        launch_wrapper(launch_wrapper&& o) noexcept

        launch_wrapper(launch_wrapper&& o) noexcept

            : h(std::exchange(o.h, nullptr))

            : h(std::exchange(o.h, nullptr))

        {

        {

        }

        }

        launch_wrapper(launch_wrapper const&) = delete;

        launch_wrapper(launch_wrapper const&) = delete;

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

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

        launch_wrapper& operator=(launch_wrapper&&) = delete;

        launch_wrapper& operator=(launch_wrapper&&) = delete;

    };

    };

    // Named functor to avoid incomplete lambda type in coroutine promise

    // Named functor to avoid incomplete lambda type in coroutine promise

    template<class Executor>

    template<class Executor>

    struct launch_coro

    struct launch_coro

    {

    {

            Executor,

            Executor,

            std::stop_token,

            std::stop_token,

            tcp_server* self,

            tcp_server* self,

            capy::task<void> t,

            capy::task<void> t,

            worker_base* wp)

            worker_base* wp)

        {

        {

            // Executor and stop token stored in promise via constructor

            // Executor and stop token stored in promise via constructor

            co_await std::move(t);

            co_await std::move(t);

            co_await self->push(*wp); // worker goes back to idle list

            co_await self->push(*wp); // worker goes back to idle list

    };

    };

    class push_awaitable

    class push_awaitable

    {

    {

        tcp_server& self_;

        tcp_server& self_;

        worker_base& w_;

        worker_base& w_;

    public:

    public:

            tcp_server& self,

            tcp_server& self,

            worker_base& w) noexcept

            worker_base& w) noexcept

        {

        {

        {

        {

        }

        }

        std::coroutine_handle<>

        std::coroutine_handle<>

            std::coroutine_handle<> h,

            std::coroutine_handle<> h,

            capy::io_env const*) noexcept

            capy::io_env const*) noexcept

        {

        {

            // Symmetric transfer to server's executor

            // Symmetric transfer to server's executor

        }

        }

        {

        {

            // Running on server executor - safe to modify lists

            // Running on server executor - safe to modify lists

            // Remove from active (if present), then wake waiter or add to idle

            // Remove from active (if present), then wake waiter or add to idle

            {

            {

            }

            }

            else

            else

            {

            {

            }

            }

    };

    };

    class pop_awaitable

    class pop_awaitable

    {

    {

        tcp_server& self_;

        tcp_server& self_;

        waiter wait_;

        waiter wait_;

    public:

    public:

        {

        {

        {

        {

        }

        }

        bool

        bool

            std::coroutine_handle<> h,

            std::coroutine_handle<> h,

            capy::io_env const*) noexcept

            capy::io_env const*) noexcept

        {

        {

            // Running on server executor (do_accept runs there)

            // Running on server executor (do_accept runs there)

        }

        }

        {

        {

            // Running on server executor

            // Running on server executor

        }

        }

    };

    };

    {

    {

    }

    }

    // Synchronous version for destructor/guard paths

    // Synchronous version for destructor/guard paths

    // Must be called from server executor context

    // Must be called from server executor context

    {

    {

        {

        {

        }

        }

        else

        else

        {

        {

        }

        }

    {

    {

    }

    }

    capy::task<void> do_accept(tcp_acceptor& acc);

    capy::task<void> do_accept(tcp_acceptor& acc);

public:

public:

    /** Abstract base class for connection handlers.

    /** Abstract base class for connection handlers.

        Derive from this class to implement custom connection handling.

        Derive from this class to implement custom connection handling.

        Each worker owns a socket and is reused across multiple

        Each worker owns a socket and is reused across multiple

        connections to avoid per-connection allocation.

        connections to avoid per-connection allocation.

        @see tcp_server, launcher

        @see tcp_server, launcher

    */

    */

    class BOOST_COROSIO_DECL

    class BOOST_COROSIO_DECL

        worker_base

        worker_base

    {

    {

        // Ordered largest to smallest for optimal packing

        // Ordered largest to smallest for optimal packing

        std::stop_source stop_;        // ~16 bytes

        std::stop_source stop_;        // ~16 bytes

        worker_base* next_ = nullptr;  // 8 bytes - used by idle and active lists

        worker_base* next_ = nullptr;  // 8 bytes - used by idle and active lists

        worker_base* prev_ = nullptr;  // 8 bytes - used only by active list

        worker_base* prev_ = nullptr;  // 8 bytes - used only by active list

        friend class tcp_server;

        friend class tcp_server;

    public:

    public:

        /// Destroy the worker.

        /// Destroy the worker.

        /** Handle an accepted connection.

        /** Handle an accepted connection.

            Called when this worker is dispatched to handle a new

            Called when this worker is dispatched to handle a new

            connection. The implementation must invoke the launcher

            connection. The implementation must invoke the launcher

            exactly once to start the handling coroutine.

            exactly once to start the handling coroutine.

            @param launch Handle to launch the connection coroutine.

            @param launch Handle to launch the connection coroutine.

        */

        */

        virtual void run(launcher launch) = 0;

        virtual void run(launcher launch) = 0;

        /// Return the socket used for connections.

        /// Return the socket used for connections.

        virtual corosio::tcp_socket& socket() = 0;

        virtual corosio::tcp_socket& socket() = 0;

    };

    };

    /** Move-only handle to launch a worker coroutine.

    /** Move-only handle to launch a worker coroutine.

        Passed to @ref worker_base::run to start the connection-handling

        Passed to @ref worker_base::run to start the connection-handling

        coroutine. The launcher ensures the worker returns to the idle

        coroutine. The launcher ensures the worker returns to the idle

        pool when the coroutine completes or if launching fails.

        pool when the coroutine completes or if launching fails.

        The launcher must be invoked exactly once via `operator()`.

        The launcher must be invoked exactly once via `operator()`.

        If destroyed without invoking, the worker is returned to the

        If destroyed without invoking, the worker is returned to the

        idle pool automatically.

        idle pool automatically.

        @see worker_base::run

        @see worker_base::run

    */

    */

    class BOOST_COROSIO_DECL

    class BOOST_COROSIO_DECL

        launcher

        launcher

    {

    {

        tcp_server* srv_;

        tcp_server* srv_;

        worker_base* w_;

        worker_base* w_;

        friend class tcp_server;

        friend class tcp_server;

        {

        {

    public:

    public:

        /// Return the worker to the pool if not launched.

        /// Return the worker to the pool if not launched.

        {

        {

        launcher(launcher&& o) noexcept

        launcher(launcher&& o) noexcept

            : srv_(o.srv_)

            : srv_(o.srv_)

            , w_(std::exchange(o.w_, nullptr))

            , w_(std::exchange(o.w_, nullptr))

        {

        {

        }

        }

        launcher(launcher const&) = delete;

        launcher(launcher const&) = delete;

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

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

        launcher& operator=(launcher&&) = delete;

        launcher& operator=(launcher&&) = delete;

        /** Launch the connection-handling coroutine.

        /** Launch the connection-handling coroutine.

            Starts the given coroutine on the specified executor. When

            Starts the given coroutine on the specified executor. When

            the coroutine completes, the worker is automatically returned

            the coroutine completes, the worker is automatically returned

            to the idle pool.

            to the idle pool.

            @param ex The executor to run the coroutine on.

            @param ex The executor to run the coroutine on.

            @param task The coroutine to execute.

            @param task The coroutine to execute.

            @throws std::logic_error If this launcher was already invoked.

            @throws std::logic_error If this launcher was already invoked.

        */

        */

        template<class Executor>

        template<class Executor>

        {

        {