Class rpc_asio_server

less_slow.cpp:7661–7720

#include <asio.hpp>

class rpc_asio_server {

    asio::io_context context_;
    asio::ip::udp::socket socket_;
    std::thread context_thread_;

    /// @brief Buffers, one per concurrent request
    std::vector<rpc_buffer_t> buffers_;
    /// @brief Where did the packets come from
    std::vector<asio::ip::udp::endpoint> clients_;
    /// @brief Which buffers are available?
    std::vector<std::size_t> buffers_available_;
    /// @brief Flag to stop the server without corrupting the state
    std::atomic_bool should_stop_;
    // Use a work guard so the io_context doesn’t run out of work and exit.
    asio::executor_work_guard<asio::io_context::executor_type> work_guard_;

    std::size_t failed_receptions_ = 0;
    std::size_t failed_responses_ = 0;

  public:
    rpc_asio_server(std::string const &address, std::uint16_t port, std::size_t max_concurrency)
        : context_(), socket_(context_), buffers_(max_concurrency), clients_(max_concurrency),
          work_guard_(asio::make_work_guard(context_)) {
        // Use your helper function to create and bind the native socket.
        auto server = rpc_server_socket(port, address);
        // Now assign the native socket to the ASIO socket.
        socket_.assign(asio::ip::udp::v4(), server.socket_descriptor);
    }

    void stop() { should_stop_.store(true, std::memory_order_seq_cst); }
    void close() {
        socket_.cancel();
        context_.stop();
        if (context_thread_.joinable()) context_thread_.join();
    }

    void operator()() {
        // For per-operation cancellations we could use the `asio::cancellation_signal`.
        // Let's issue a receive operation for each buffer, which will call a chain of
        // operations to process the packet and send a response, and repeat again.
        for (std::size_t job = 0; job < buffers_.size(); ++job) reuse_buffer(job);
        // Start listening for incoming packets.
        context_thread_ = std::thread([this] { context_.run(); });
    }

  private:
    void reuse_buffer(std::size_t job) {
        auto finalize = [this, job](std::error_code error, std::size_t) {
            if (error) failed_responses_++;
            if (should_stop_.load(std::memory_order_seq_cst)) return;
            reuse_buffer(job);
        };
        auto respond = [this, finalize, job](std::error_code error, std::size_t bytes) {
            if (error) { reuse_buffer(job); }
            else { socket_.async_send_to(asio::buffer(buffers_[job], bytes), clients_[job], finalize); }
        };
        socket_.async_receive_from(asio::buffer(buffers_[job]), clients_[job], respond);