Skip to main content

read_ahead

read_ahead bounds how many deliveries a subscription reads ahead of its handler, 1000 by default; past it the sender waits.

A subscription reads its socket on a task of its own and hands the deliveries to its handler through a buffer. read_ahead sets the size of that buffer, counted in deliveries. When the buffer is full, the subscription stops reading. The socket buffers fill in turn, and the peer's send waits until the handler catches up. So a slow handler slows its sender down, and the memory of this process stays bounded. On PUB/SUB the waiting sender is the PUB socket, held back by its slowest matching subscriber, as Backpressure draws.

The setting belongs to the broker: ZmqQueue::read_ahead, ZmqFanout::read_ahead and ZmqRpc::read_ahead take a NonZeroUsize. Every subscription mounted on the broker reads with that bound, and a clone of the broker keeps its own value.

The in-process transport hands deliveries over without a bound, so a test sees the bound only over sockets, in a live test.

Running in production sets the bound from the environment.

Example​

Each frame takes the renderer seconds, so it reads four frames ahead and leaves the rest with the sender.

examples/zmq_subscriptions_read_ahead.rs
#[ruststream::app]
fn app() -> impl App {
    // Each frame takes seconds to render, so a short read-ahead holds the rest at the sender.
    let frames =
        ZmqQueue::new(ZmqEndpoint::bind("tcp://0.0.0.0:5555")).read_ahead(nonzero!(4_usize));
    RustStream::new(AppInfo::new("renderer", "0.1.0")).with_broker(frames, |b| {
        b.include(render);
    })
}