Build a log shipping pipeline with PUSH and PULL

You will build a small log shipper over TCP on localhost. Three apps push their log lines into one collector, which stamps each line and pushes it on to a writer stage. Then a stalled collector shows what the high-water marks and a send timeout do when the next stage can't keep up.

rzmq = "0"

Set up the project#

Start a new binary crate.

cargo new logship
cd logship

Add rzmq and Tokio. rzmq runs on the Tokio runtime, so you need both. Your [dependencies] section should look like this:

[dependencies]
rzmq = "0"
tokio = { version = "1", features = ["full"] }

The default features cover everything here. Every socket talks TCP on 127.0.0.1, so each piece could run as its own process. They all live in one process as Tokio tasks so you can run the whole thing at once. The programs use Tokio's single-threaded current_thread runtime, which keeps the output in the same order from run to run.

Collect lines from several apps#

The task ventilator in the earlier tutorial used PUSH/PULL to fan work out: one PUSH that binds, several PULL workers that connect. The same pair also works the other way round. Here one PULL socket is the stable end and binds. Each app connects its own PUSH socket to it. That is fan-in: many producers feed one consumer and none of the producers needs to know about the others.

A PULL socket only receives and a PUSH socket only sends, so there is no reply to wait for. An app hands a line to its socket and carries on.

Replace src/main.rs with this:

use rzmq::{Context, Msg, SocketType, ZmqError};

const COLLECTOR: &str = "tcp://127.0.0.1:5640";
const APPS: [&str; 3] = ["api", "auth", "billing"];
const LINES_PER_APP: usize = 3;

#[tokio::main(flavor = "current_thread")]
async fn main() -> Result<(), ZmqError> {
    let ctx = Context::new()?;

    let collector = ctx.socket(SocketType::Pull)?;
    collector.bind(COLLECTOR).await?;

    let mut apps = Vec::new();
    for app in APPS {
        let shipper = ctx.socket(SocketType::Push)?;
        shipper.connect(COLLECTOR).await?;
        apps.push(tokio::spawn(async move {
            for n in 1..=LINES_PER_APP {
                let line = format!("{app}: request {n} done");
                shipper.send(Msg::from_vec(line.into_bytes())).await?;
            }
            Ok::<(), ZmqError>(())
        }));
    }

    for _ in 0..APPS.len() * LINES_PER_APP {
        let msg = collector.recv().await?;
        println!("{}", String::from_utf8_lossy(msg.data().unwrap_or_default()));
    }

    for app in apps {
        app.await.expect("app task panicked")?;
    }
    ctx.term().await?;
    Ok(())
}

Each app gets its own PUSH socket and runs in its own Tokio task. The collector knows that three apps send three lines each, so it reads nine messages and stops. main then waits for the app tasks and calls ctx.term(), which closes every socket made from the context.

Run it:

$ cargo run
api: request 1 done
api: request 2 done
api: request 3 done
auth: request 1 done
auth: request 2 done
auth: request 3 done
billing: request 1 done
billing: request 2 done
billing: request 3 done

The lines from one app always arrive in the order that app sent them. How lines from different apps mix depends on when each app sends. On the single-threaded runtime each app task sends its three lines in one go, so they arrive grouped by app.

Add a writer stage#

A real collector does something with each line and hands it on. Pipelines are what PUSH/PULL is for: each stage reads from a PULL socket, does its work and writes to a PUSH socket that the next stage reads from. Here the collector becomes an aggregator that gives every line a sequence number and pushes it to a writer.

The collector can no longer count on a fixed number of lines, so the apps say when they are done. Each app sends an END message after its last line. The aggregator counts the END messages and, once all three apps are finished, sends one END of its own down to the writer. The writer prints lines until it sees that END. A PUSH socket keeps each sender's messages in order, so an app's END always arrives after its last line.

The apps now start before the aggregator binds. That is the missing-peer case: a PUSH socket with nothing connected to it waits in send until a peer is there. Meanwhile the connect keeps retrying in the background until the aggregator's bind shows up. Nothing is lost and no app needs to know whether the collector is up yet.

Replace src/main.rs with the full program:

use rzmq::{Context, Msg, Socket, SocketType, ZmqError};

const COLLECTOR: &str = "tcp://127.0.0.1:5640";
const WRITER: &str = "tcp://127.0.0.1:5641";
const APPS: [&str; 3] = ["api", "auth", "billing"];
const LINES_PER_APP: usize = 3;
const END: &[u8] = b"END";

fn text(msg: &Msg) -> String {
    String::from_utf8_lossy(msg.data().unwrap_or_default()).into_owned()
}

async fn app(name: &'static str, shipper: Socket) -> Result<(), ZmqError> {
    for n in 1..=LINES_PER_APP {
        let line = format!("{name}: request {n} done");
        shipper.send(Msg::from_vec(line.into_bytes())).await?;
    }
    shipper.send(Msg::from_static(END)).await
}

async fn aggregator(inbox: Socket, outbox: Socket) -> Result<(), ZmqError> {
    let mut seq = 0;
    let mut finished = 0;
    while finished < APPS.len() {
        let msg = inbox.recv().await?;
        if msg.data() == Some(END) {
            finished += 1;
            continue;
        }
        seq += 1;
        let stamped = format!("{seq:04} {}", text(&msg));
        outbox.send(Msg::from_vec(stamped.into_bytes())).await?;
    }
    outbox.send(Msg::from_static(END)).await
}

async fn writer(inbox: Socket) -> Result<usize, ZmqError> {
    let mut written = 0;
    loop {
        let msg = inbox.recv().await?;
        if msg.data() == Some(END) {
            return Ok(written);
        }
        println!("{}", text(&msg));
        written += 1;
    }
}

#[tokio::main(flavor = "current_thread")]
async fn main() -> Result<(), ZmqError> {
    let ctx = Context::new()?;

    let mut apps = Vec::new();
    for name in APPS {
        let shipper = ctx.socket(SocketType::Push)?;
        shipper.connect(COLLECTOR).await?;
        apps.push(tokio::spawn(app(name, shipper)));
    }

    let inbox = ctx.socket(SocketType::Pull)?;
    inbox.bind(COLLECTOR).await?;
    let outbox = ctx.socket(SocketType::Push)?;
    outbox.bind(WRITER).await?;
    let aggregating = tokio::spawn(aggregator(inbox, outbox));

    let writer_inbox = ctx.socket(SocketType::Pull)?;
    writer_inbox.connect(WRITER).await?;
    let written = writer(writer_inbox).await?;
    println!("writer: {written} lines");

    aggregating.await.expect("aggregator panicked")?;
    for app in apps {
        app.await.expect("app panicked")?;
    }
    ctx.term().await?;
    Ok(())
}

The aggregator binds both of its sockets. It sits in the middle of the pipeline and is the piece least likely to move, so the apps and the writer connect to it. writer runs on the main task and returns the number of lines it wrote.

Run it:

$ cargo run
0001 auth: request 1 done
0002 auth: request 2 done
0003 auth: request 3 done
0004 api: request 1 done
0005 api: request 2 done
0006 api: request 3 done
0007 billing: request 1 done
0008 billing: request 2 done
0009 billing: request 3 done
writer: 9 lines

The apps have to reconnect once the aggregator binds. The order they get through that can change between runs, so the apps may appear in a different order. The sequence numbers always run from 0001 to 0009, each app's three lines stay in order and the writer always reports nine lines.

Stall the collector#

So far every stage keeps up. When one doesn't, messages queue up. Two socket options put a limit on that queue. SNDHWM is the high-water mark for the sending side: how many messages a PUSH socket holds for a peer that hasn't taken them yet. RCVHWM is the same for the receiving side: how many a PULL socket holds that your code hasn't read with recv. When the queue is full, send on the PUSH socket waits for space. That is the backpressure in this pattern. A slow consumer slows its producers down instead of making them buffer without end. The socket doesn't drop any message.

Waiting is right for the aggregator, which has nothing better to do. An app shouldn't stop serving requests because its log collector is stuck. SNDTIMEO sets how long send may wait, in milliseconds. When the time runs out, send returns ZmqError::Timeout and the message was not sent. The app can then decide what to do. Here it drops the rest of its burst.

This step uses one app talking straight to a collector, so there is only one queue to watch. The collector is stalled: it reads nothing until the app tells it how many lines got through. The app tries to ship a burst of 100,000 lines.

Replace src/main.rs:

use rzmq::socket::{RCVHWM, SNDHWM, SNDTIMEO};
use rzmq::{Context, Msg, Socket, SocketType, ZmqError};
use tokio::sync::oneshot;

const COLLECTOR: &str = "tcp://127.0.0.1:5642";
const HWM: i32 = 100;
const BURST: usize = 100_000;

async fn stalled_collector(inbox: Socket, resume: oneshot::Receiver<usize>) -> Result<(), ZmqError> {
    let shipped = resume.await.expect("app went away");
    for _ in 0..shipped {
        inbox.recv().await?;
    }
    println!("collector: caught up on {shipped} lines");
    Ok(())
}

#[tokio::main(flavor = "current_thread")]
async fn main() -> Result<(), ZmqError> {
    let ctx = Context::new()?;

    let inbox = ctx.socket(SocketType::Pull)?;
    inbox.set_option(RCVHWM, HWM).await?;
    inbox.bind(COLLECTOR).await?;
    let (resume, resumed) = oneshot::channel();
    let collecting = tokio::spawn(stalled_collector(inbox, resumed));

    let shipper = ctx.socket(SocketType::Push)?;
    shipper.set_option(SNDHWM, HWM).await?;
    shipper.set_option(SNDTIMEO, 100i32).await?;
    shipper.connect(COLLECTOR).await?;

    let mut shipped = 0;
    for n in 1..=BURST {
        let line = format!("api: request {n} done");
        match shipper.send(Msg::from_vec(line.into_bytes())).await {
            Ok(()) => shipped += 1,
            Err(ZmqError::Timeout) => break,
            Err(e) => return Err(e),
        }
    }
    println!("app: shipped {shipped} lines, dropped {}", BURST - shipped);

    resume.send(shipped).expect("collector went away");
    collecting.await.expect("collector panicked")?;
    ctx.term().await?;
    Ok(())
}

Both options take an i32 and are set before bind and connect. The high-water marks go on both ends, SNDHWM on the PUSH and RCVHWM on the PULL. The oneshot channel stands in for whatever unsticks a real collector. It also tells the collector how many lines to read.

Run it:

$ cargo run
app: shipped 22800 lines, dropped 77200
collector: caught up on 22800 lines

Your numbers will likely differ. Messages also sit in the operating system's TCP buffers between the two sockets, so the app gets further than the two high-water marks alone would allow. How much further depends on the machine and the size of each line. What stays fixed is that shipped plus dropped is always 100,000 and the collector gets every line that was shipped. Raise HWM and more lines get through before the timeout; lower it and fewer do.

Without SNDTIMEO the app's first send into a full queue would wait until the collector started reading again. That is exactly what you want for a pipeline stage like the aggregator, where a slow writer should slow down everything upstream of it.

Shut down cleanly#

In the pipeline every task knows when it is finished, because the END messages travel through every stage. Shutdown is waiting for the tasks in order and then terminating the context:

let writer_inbox = ctx.socket(SocketType::Pull)?;
writer_inbox.connect(WRITER).await?;
let written = writer(writer_inbox).await?;
println!("writer: {written} lines");

aggregating.await.expect("aggregator panicked")?;
for app in apps {
    app.await.expect("app panicked")?;
}
ctx.term().await?;

The writer returns after the aggregator's END, which the aggregator only sends after the END from every app. By the time main calls Context::term(), every message has been read and no task still holds a socket it is using. term() closes every socket made from the context and waits for their background tasks to finish. Messages still queued at that point are handled according to each socket's LINGER option. Any error along the way is passed up with ? instead of being ignored.

The END marker is one message per producer. The stage in the middle has to count them before passing its own on, since a PULL socket has no idea how many producers are connected to it.

Where to go next#

Every stage in this tutorial is a PULL that reads and a PUSH that writes, so you can add stages or run several copies of one. Several writers connected to the aggregator's PUSH would share the lines round-robin, which is the fan-out shape from the ventilator tutorial. Other options work the same way as the ones here. RCVTIMEO puts a time limit on recv and returns ZmqError::Timeout when it runs out. monitor_default() reports ConnectionCongested and ConnectionUncongested events when a peer's send queue hits its high-water mark and drains again.

The rzmq usage guide on GitHub covers socket options, monitoring, security and the io_uring backend.

The full API reference is on docs.rs.