How to Receive UDP Datagrams in Batches Using C++ ASIO
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:
- Call
socket_.async_wait(udp::socket::wait_read, handler)to asynchronously wait until the socket has incoming data. - When notified that the socket is readable, use a custom
read_batch()function that invokesrecvmmsg(2)onsocket_.native_handle()to fetch up to $N$ datagrams directly into pre-allocated memory buffers. - Process the batch, then call
async_waitagain 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,
recvmmsgissues 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 theio_contextevent 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-registeringasync_wait(), minimizing reactor loop turns. - Cross-Platform Considerations:
recvmmsgis Linux-specific. For Windows systems handling extreme UDP throughput, look into Windows Registered I/O (RIO) viaRIORecvEx.