Class rpc_libc_client

less_slow.cpp:6801–6861

* @brief A minimal RPC @b client using LibC functionality to setup the UDP socket, * and synchronous blocking POSIX calls to send and receive packets - * @b one at a time! */

 *          @b one at a time!
 */
class rpc_libc_client {
    int socket_descriptor_;
    sockaddr_in server_address_;
    std::size_t concurrency_;

  public:
    rpc_libc_client(std::string const &server_address_str, std::uint16_t port, std::size_t concurrency)
        : concurrency_(concurrency) {

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

        // Let's make sure we don't block forever on `recvfrom`
        struct timeval duration;
        duration.tv_sec = 0;
        duration.tv_usec = to_microseconds(rpc_batch_timeout_k).count();
        if (setsockopt(socket_descriptor_, SOL_SOCKET, SO_RCVTIMEO, &duration, sizeof(duration)) < 0)
            raise_system_error("Failed to set sockets batch timeout");
    }

    ~rpc_libc_client() noexcept { close(socket_descriptor_); }

    rpc_batch_result operator()() noexcept {
        rpc_batch_result result;

        sockaddr_in reply_addr;
        socklen_t reply_len = sizeof(reply_addr);
        rpc_buffer_t send_buffer, receive_buffer;
        send_buffer.fill('X');

        for (std::size_t i = 0; i < concurrency_; ++i) {
            auto send_time = std::chrono::steady_clock::now();
            ssize_t sent_length = sendto(socket_descriptor_, send_buffer.data(), send_buffer.size(), 0,
                                         reinterpret_cast<sockaddr *>(&server_address_), sizeof(server_address_));
            result.sent_packets++;
            if (sent_length <= 0) continue;

            // In general, `select` is used to monitor multiple file descriptors or sockets at once
            // to see if they are ready for I/O, but in this case we use it to constrain the time
            // we are willing to wait for a single response.
            struct timeval expiry;
            expiry.tv_sec = 0;
            expiry.tv_usec = to_microseconds(rpc_packet_timeout_k).count();
            fd_set available_descriptors;
            FD_ZERO(&available_descriptors);
            FD_SET(socket_descriptor_, &available_descriptors);
            if (select(socket_descriptor_ + 1, &available_descriptors, nullptr, nullptr, &expiry) <= 0) continue;

            ssize_t received_length = recvfrom(socket_descriptor_, receive_buffer.data(), receive_buffer.size(), 0,
                                               reinterpret_cast<sockaddr *>(&reply_addr), &reply_len);
            if (received_length <= 0) continue;
            auto response_time = std::chrono::steady_clock::now();
            auto diff = response_time - send_time;
            result.batch_latency += diff;
            result.max_packet_latency = std::max(result.max_packet_latency, diff);
            result.received_packets++;
        }