When building high-throughput UDP applications—such as market data feeds, real-time telemetry, or multicast media streaming—handling datagrams one by one via async_receive_from introduces significant CPU overhead. Each individual receive call incurs context switches and system call overhead that can quickly become a performance bottleneck.

While ASIO supports scatter-gather I/O for reading parts of a single datagram into multiple buffers, it does not provide an out-of-the-box cross-platform abstraction for reading multiple distinct datagrams in a single call. However, you can easily achieve true batch reception in ASIO by pairing ASIO's readiness model (async_wait) with OS-level batching primitives like Linux's recvmmsg.

The Core Strategy: Reactor Pattern with async_wait

Instead of dispatching an async read operation for each packet, you can treat ASIO as a reactor:

  1. Call socket_.async_wait(udp::socket::wait_read, handler) to asynchronously wait until the socket has incoming data.
  2. When notified that the socket is readable, use a custom read_batch() function that invokes recvmmsg(2) on socket_.native_handle() to fetch up to $N$ datagrams directly into pre-allocated memory buffers.
  3. Process the batch, then call async_wait again to wait for the next burst.

Implementing read_batch and the Batch Receiver

Below is a production-ready example demonstrating how to implement a batched UDP multicast receiver using standalone ASIO and POSIX recvmmsg.

#include <iostream>      
#include <vector>        
#include <array>         
#include <system_error>  
#include <sys/socket.h> 
#include <netinet/in.h>  
#include <asio.hpp>      

using asio::ip::udp;

struct Packet {
    static constexpr std::size_t max_size = 1500;
    std::array<char, max_size> data;
    std::size_t size{0};
    sockaddr_storage sender_address{};
    socklen_t sender_address_len{sizeof(sockaddr_storage)};
};

class BatchUdpServer {
public:
    static constexpr std::size_t BATCH_SIZE = 64;

    BatchUdpServer(asio::io_context& io_context, unsigned short port, const std::string& mcast_addr)
        : socket_(io_context, udp::v4())
    {
        socket_.set_option(udp::socket::reuse_address(true));
        socket_.bind(udp::endpoint(udp::v4(), port));

        // Join multicast group
        const asio::ip::address_v4 group = asio::ip::make_address_v4(mcast_addr);
        socket_.set_option(asio::ip::multicast::join_group(group));

        // Set socket to non-blocking for recvmmsg
        socket_.non_blocking(true);
    }

    void start()
    {
        do_wait();
    }

    // Batch reading function
    std::size_t read_batch(std::span<Packet> packets, std::error_code& ec)
    {
        if (packets.empty()) return 0;

        std::vector<mmsghdr> msgs(packets.size());
        std::vector<iovec> iovs(packets.size());

        for (std::size_t i = 0; i < packets.size(); ++i) {
            iovs[i].iov_base = packets[i].data.data();
            iovs[i].iov_len = Packet::max_size;

            msgs[i].msg_hdr.msg_name = &packets[i].sender_address;
            msgs[i].msg_hdr.msg_namelen = sizeof(sockaddr_storage);
            msgs[i].msg_hdr.msg_iov = &iovs[i];
            msgs[i].msg_hdr.msg_iovlen = 1;
            msgs[i].msg_hdr.msg_control = nullptr;
            msgs[i].msg_hdr.msg_controllen = 0;
            msgs[i].msg_hdr.msg_flags = 0;
        }

        // Perform non-blocking batch receive
        int retval = ::recvmmsg(socket_.native_handle(), msgs.data(), msgs.size(), MSG_DONTWAIT, nullptr);

        if (retval < 0) {
            if (errno != EAGAIN && errno != EWOULDBLOCK) {
                ec = std::error_code(errno, std::generic_category());
            }
            return 0;
        }

        for (int i = 0; i < retval; ++i) {
            packets[i].size = msgs[i].msg_len;
            packets[i].sender_address_len = msgs[i].msg_hdr.msg_namelen;
        }

        return static_cast<std::size_t>(retval);
    }

private:
    void do_wait()
    {
        socket_.async_wait(
            udp::socket::wait_read,
            [this](std::error_code ec) {
                if (ec == asio::error::operation_aborted) return;

                if (!ec) {
                    std::array<Packet, BATCH_SIZE> batch;
                    std::error_code batch_ec;
                    std::size_t count = read_batch(batch, batch_ec);

                    if (!batch_ec && count > 0) {
                        handle_batch(std::span<Packet>(batch.data(), count));
                    }
                } else {
                    std::cerr << "Async wait error: " << ec.message() << '\n';
                }

                // Wait for the next readable event
                do_wait();
            });
    }

    void handle_batch(std::span<Packet> packets)
    {
        std::cout << "Received batch of " << packets.size() << " datagrams.\n";
        for (const auto& pkt : packets) {
            std::cout << "  - Packet size: " << pkt.size << " bytes\n";
        }
    }

    udp::socket socket_;
};

Key Benefits of This Approach

  • Drastic Syscall Reduction: Instead of executing 64 system calls to receive 64 packets, recvmmsg issues a single kernel transition, freeing up CPU cycles for payload parsing.
  • Seamless Integration with ASIO: By leveraging socket::async_wait, your application remains fully asynchronous and event-driven without blocking the io_context event loop.
  • Cache Locality: Processing contiguous arrays of packets keeps buffer addresses hot in the CPU cache, reducing cache misses.

Optimization Tips

  • Increase Socket Buffer Size: For high-throughput scenarios, increase the OS receive buffer using socket_.set_option(asio::socket_base::receive_buffer_size(4 * 1024 * 1024)); to prevent UDP drops during traffic bursts.
  • Loop Until Exhaustion: In high-load systems, you can call read_batch() in a loop within the completion handler until it returns 0 before re-registering async_wait(), minimizing reactor loop turns.
  • Cross-Platform Considerations: recvmmsg is Linux-specific. For Windows systems handling extreme UDP throughput, look into Windows Registered I/O (RIO) via RIORecvEx.