Class rpc_uring60_server

less_slow.cpp:7353–7460

* @brief A minimal RPC @b server using @b `io_uring` functionality * to setup the UDP socket, and process many requests concurrently. * * Unlike the `rpc_uring55_server`, this version: * - registers buffers and off-loads buffer selection to the kernel * - reduces the number of receive operations, using multi-shot receive */

 *  - reduces the number of receive operations, using multi-shot receive
 */
class rpc_uring60_server {

    int socket_descriptor_;
    sockaddr_in server_address_;
    std::atomic_bool should_stop_;
    io_uring ring_;

    // Pre-allocated resources
    mmap_array<message_t> messages_;
    std::size_t max_concurrency_;

  public:
    using status_t = message_t::message_status_t;

    rpc_uring60_server(std::string const &server_address_str, std::uint16_t port, std::size_t max_concurrency)
        : should_stop_(false), messages_(max_concurrency * 2), max_concurrency_(max_concurrency) {

        auto [socket_descriptor, server_address] = rpc_server_socket(port, server_address_str);
        socket_descriptor_ = socket_descriptor;
        server_address_ = server_address;

        // Zero copy operations would require more socket options
        int const one = 1;
        if (setsockopt(socket_descriptor_, SOL_SOCKET, SO_ZEROCOPY, &one, sizeof(one)) < 0)
            raise_system_error("Failed to enable zero-copy on socket");

        // Initialize `io_uring` with one slot for each receive/send operation
        // TODO: |= IORING_SETUP_COOP_TASKRUN | IORING_SETUP_SINGLE_ISSUER | IORING_SETUP_SUBMIT_ALL
        auto io_uring_setup_flags = 0;
        if (io_uring_queue_init(max_concurrency * 2, &ring_, io_uring_setup_flags) < 0)
            raise_system_error("Failed to initialize io_uring 6.0 server");
        if (io_uring_register_files(&ring_, &socket_descriptor_, 1) < 0)
            raise_system_error("Failed to register file descriptor with io_uring 6.0 server");

        // Initialize message resources
        for (message_t &message : messages_) {
            memset(&message.header, 0, sizeof(message.header));
            message.header.msg_name = &message.peer_address;
            message.header.msg_namelen = sizeof(sockaddr_in);
            // Each message will be made of just one buffer
            message.header.msg_iov = &message.io_vec;
            message.header.msg_iovlen = 1;
            // ... and that buffer is a member of our `message`
            message.io_vec.iov_base = message.buffer.data();
            message.io_vec.iov_len = message.buffer.size();
            message.status = status_t::pending_k;
        }

        // Let's register all of those with `IORING_REGISTER_BUFFERS`
        std::vector<struct iovec> iovecs_to_register;
        for (message_t &message : messages_) iovecs_to_register.push_back(message.io_vec);
        if (io_uring_register_buffers(&ring_, iovecs_to_register.data(), iovecs_to_register.size()) < 0)
            raise_system_error("Failed to register buffers with io_uring 6.0 server");
    }

    ~rpc_uring60_server() noexcept {}
    void close() noexcept {
        ::close(socket_descriptor_);