ZMQ IPC Fan-out Benchmark

Message rate, data throughput, and delivery latency for a Julia PUB/SUB broadcaster

Kahli Burke

Abstract

A single publisher broadcasts to N subscribers over ZMQ ipc:// (Unix-domain sockets), and we measure three axes as payload size and subscriber count scale: message rate, data throughput (MB/s), and end-to-end delivery latency (recv − send percentiles). Latency is coordinated-omission-correct — messages are stamped with their intended send time under a fixed schedule, and a publisher-slip flag marks when offered load exceeded consumer capacity (so a latency number is either trustworthy or explicitly flagged as backlog). This is an in-box proxy: no network stack, subscribers are green-threaded tasks in one process. It characterises Julia's scheduler + ZMQ fan-out + GC under a broadcast load — it is not TCP connection-scaling, and not a fair cross-language comparison (ZMQ PUB copies per-subscriber inside the send loop).

Method

Each frame is [tag | seq | send-timestamp | padding]. A PUSH/PULL readiness handshake makes every subscriber ack before the timed run starts, defeating ZMQ's slow-joiner drop — so delivery is lossless and any shortfall is reported as a genuine drop. Subscribers are pooled across ZMQ contexts (500 subs/context) to stay under the ZMQ_MAX_SOCKETS = 1023 cap.

Two regimes are reported separately:

Saturation (unpaced): the publisher sends as fast as possible → measures peak throughput. Latency here is pure queue residency and is not reported as delivery cost.
Paced (fixed msgs/s): the publisher holds a schedule → latency is trustworthy as long as the slip flag stays green.

Sweep A — data throughput vs payload size

Fixing a modest fan-out (8 subscribers) and running unpaced (saturation), we grow the frame from 64 B to 1 MB. Two things trade off: the message rate a system can push (limited by per-message overhead — syscalls, framing, scheduler wakeups) and the raw byte throughput (limited by memory bandwidth and the per-subscriber copy). Small frames are dominated by the former, large frames by the latter, and the crossover is visible in the two charts below.

## Sweep A — data throughput vs payload size (8 subscribers, saturation)
## Data throughput here is single-publisher-bound (one thread does the serial
## fan-out), so this is a single-thread send ceiling, not an all-core number.
bytelbl(n) = n < 1024 ? "$n B" : n < 2^20 ? "$(n÷1024) KB" : "$(n÷2^20) MB"

function measure_payload(payload; n_subs=8, tries=6)
    frame  = max(ZMQFanoutBench.HDR, payload)
    n_msgs = clamp(div(1_000_000_000, n_subs*frame), 100, 20_000)
    lasterr = nothing
    for t in 1:tries
        slate_progress(NaN; msg="$(bytelbl(payload)): run $t/$tries …")
        try
            r = run_fanout(; n_subs, n_msgs, payload, rate=0.0, io=IOBuffer())
            r.dropped == 0 && return r          # first clean run wins
        catch e
            lasterr = e                          # readiness/transient — retry
        end
        GC.gc(true); sleep(0.4)
    end
    lasterr === nothing ? error("payload $payload never ran clean") : throw(lasterr)
end

measure_payload (generic function with 1 method)

payloadsA = [64, 4096, 65536, 262144, 1048576]
dfA = DataFrame(payload_B=Int[], n_msgs=Int[], msg_rate=Float64[],
                mbps=Float64[], per_stream_mbps=Float64[], p50_ms=Float64[], drop=Int[])
for (i, payload) in enumerate(payloadsA)
    slate_progress((i-1)/length(payloadsA); msg="measuring $(bytelbl(payload)) …")
    r = measure_payload(payload)
    push!(dfA, (payload, r.n_msgs, r.deliv_per_s, r.mbps, r.mbps/8, r.p50, r.dropped))
    slate_progress(i/length(payloadsA); msg="$(bytelbl(payload)) → $(round(Int, r.mbps)) MB/s")
    GC.gc(true)
end
dfA;

payload_B	n_msgs	msg_rate	mbps	per_stream_mbps	p50_ms
64 B	20,000	2,975,729	190	24	6.0
4.0 KB	20,000	267,676	1096	137	277.7
64.0 KB	1,907	35,586	2332	292	200.1
256.0 KB	476	11,373	2981	373	174.3
1.0 MB	119	2,684	2815	352	148.1

Sweep B — the production/consumption knee (500 subscribers)

Holding 500 subscribers, we ramp the paced publish rate through the knee and compare two transports for the same fan-out: ZMQ ipc:// (the real path — Unix sockets, wire framing) and a pure-Julia Channel baseline (one bounded channel per subscriber; no ZMQ, no IPC, no serialization). Below the knee, demanded (rate × 500) and achieved deliveries/s track; above it the consumers can't absorb the fan-out, achieved flatlines, and latency becomes backlog. The baseline isolates Julia's raw scheduler ceiling from the transport tax — and the two trade off differently.

## 500 subscribers, 256 B frames — rate sweep concentrated around the knee (~3.4k msg/s)
## Two transports: ZMQ ipc:// vs a pure-Julia Channel baseline (no ZMQ, no IPC, no serialization).
ratesB = [1000, 1500, 2000, 2500, 3000, 3400, 3800, 4200, 5000, 6000]
nsubsB = 500; nmsgsB = 6000

# ZMQ readiness can trip on a cold-start straggler at 500 subs — retry until clean.
function zmq_run(rate)
    for _ in 1:6
        try
            rr = run_fanout(; n_subs=nsubsB, n_msgs=nmsgsB, payload=256, rate=Float64(rate), io=IOBuffer())
            rr.dropped == 0 && return rr
        catch; end
        GC.gc(true); sleep(0.4)
    end
    return run_fanout(; n_subs=nsubsB, n_msgs=nmsgsB, payload=256, rate=Float64(rate), io=IOBuffer())
end

zmq_run (generic function with 1 method)

rowsB = NamedTuple[]
for (i, rate) in enumerate(ratesB)
    slate_progress((i-1)/length(ratesB); msg="ZMQ @ $(rate) msg/s ($i/$(length(ratesB))) …")
    rz = zmq_run(rate)
    slate_progress((i-0.5)/length(ratesB); msg="Channel @ $(rate) msg/s ($i/$(length(ratesB))) …")
    rc = run_fanout_channels(; n_subs=nsubsB, n_msgs=nmsgsB, rate=Float64(rate))
    demanded = rate * nsubsB
    for (t, r) in (("ZMQ ipc", rz), ("Channel", rc))
        push!(rowsB, (rate=rate, transport=t, demanded=demanded, achieved=r.deliv_per_s,
                      keepup=r.deliv_per_s >= 0.97*demanded, slip_ms=r.max_slip_ns/1e6,
                      p50_ms=r.p50, p99_ms=r.p99, p999_ms=r.p999, drop=r.dropped))
    end
    slate_progress(i/length(ratesB); msg="$(rate) msg/s: ZMQ $(round(rz.deliv_per_s/1e6;digits=2))M vs Chan $(round(rc.deliv_per_s/1e6;digits=2))M deliv/s")
    GC.gc(true)
end
dfB = DataFrame(rowsB);

rate	transport	demanded	achieved	keepup	slip_ms	p50_ms	p99_ms	p999_ms
1,000	Channel	500,000	492,287	true	11	0	4	9
1,500	Channel	750,000	733,511	true	3	0	0	2
2,000	Channel	1,000,000	970,220	true	8	0	4	8
2,500	Channel	1,250,000	1,227,240	true	49	14	47	49
3,000	Channel	1,500,000	1,171,208	false	582	276	579	582
3,400	Channel	1,700,000	1,176,661	false	694	237	685	693
3,800	Channel	1,900,000	1,246,018	false	743	446	732	742
4,200	Channel	2,100,000	1,341,835	false	760	444	760	772
5,000	Channel	2,500,000	1,218,811	false	1171	507	1158	1169
6,000	Channel	3,000,000	1,172,173	false	1469	717	1452	1467
1,000	ZMQ ipc	500,000	497,698	true	27	21	140	251
1,500	ZMQ ipc	750,000	740,399	true	51	27	197	353
2,000	ZMQ ipc	1,000,000	977,923	true	53	33	265	422
2,500	ZMQ ipc	1,250,000	1,221,036	true	79	58	357	558
3,000	ZMQ ipc	1,500,000	1,426,947	false	129	108	605	784
3,400	ZMQ ipc	1,700,000	1,617,431	false	88	80	632	780
3,800	ZMQ ipc	1,900,000	1,600,236	false	251	190	871	1181
4,200	ZMQ ipc	2,100,000	1,771,106	false	217	223	869	1101
5,000	ZMQ ipc	2,500,000	1,809,983	false	334	324	1311	1562
6,000	ZMQ ipc	3,000,000	1,713,236	false	387	546	1113	1299

zmq = dfB[dfB.transport .== "ZMQ ipc", :]
ch  = dfB[dfB.transport .== "Channel", :]
fig = Figure(size=(820, 440))
ax = Axis(fig[1, 1]; yscale=log10, xlabel="publish rate (msg/s)", ylabel="p99 delivery latency (ms)",
          title="Tail latency (p99) vs offered load (500 subscribers)")
scatterlines!(ax, zmq.rate, zmq.p99_ms; markersize=11, linewidth=2.5, label="ZMQ ipc://")
scatterlines!(ax, ch.rate,  ch.p99_ms;  markersize=11, linewidth=2.5, label="Julia Channel (baseline)")
axislegend(ax; position=:rb)
fig

$output$

Sweep C — Request/Reply (the interactive path)

Fan-out is the broadcast path; this is the interactive one — every send, auth, or history fetch is a request that waits for a reply. N DEALER clients issue closed-loop requests (each waits for its reply before sending the next) to a single ROUTER server that echoes them back; the echo carries the send timestamp, so we measure true round-trip latency. Because closed-loop clients self-throttle, sweeping the client count traces the classic latency-vs-throughput curve: throughput climbs with concurrency until the server saturates, then round-trip latency climbs at (roughly) fixed throughput as requests queue.

## Concurrency sweep: vary client count, closed-loop, 64 B requests.
clientsC = [1, 2, 5, 10, 25, 50, 100, 200, 400]

# Total round-trips per point ≈ 40k (bounded runtime); retry on the rare drop.
function reqrep_point(nc; tries=4)
    n_reqs = clamp(div(40_000, nc), 100, 400)
    local r
    for t in 1:tries
        slate_progress(NaN; msg="$(nc) clients: run $t/$tries …")
        try
            r = run_reqrep(; n_clients=nc, n_reqs, payload=64, rate=0.0)
            r.dropped == 0 && return r
        catch; end
        GC.gc(true); sleep(0.3)
    end
    return r
end

reqrep_point (generic function with 1 method)

dfC = DataFrame(n_clients=Int[], n_reqs=Int[], req_per_s=Float64[],
                p50_ms=Float64[], p99_ms=Float64[], p999_ms=Float64[], drop=Int[])
for (i, nc) in enumerate(clientsC)
    slate_progress((i-1)/length(clientsC); msg="$(nc) clients …")
    r = reqrep_point(nc)
    push!(dfC, (nc, r.n_reqs, r.req_per_s, r.p50, r.p99, r.p999, r.dropped))
    slate_progress(i/length(clientsC); msg="$(nc) clients → $(round(Int,r.req_per_s)) req/s, p50 $(round(r.p50;digits=1))ms")
    GC.gc(true)
end
dfC;

n_clients	n_reqs	req_per_s	p50_ms	p99_ms	p999_ms
1	400	7,583	0.13	0.24	0.34
2	400	12,182	0.15	0.35	1.04
5	400	20,251	0.24	0.42	1.13
10	400	24,425	0.40	0.58	0.75
25	400	27,056	0.92	1.20	1.43
50	400	25,582	1.91	3.15	4.29
100	400	24,864	3.91	6.70	8.07
200	200	23,708	8.04	15.17	21.66
400	100	21,208	17.99	33.10	39.71

Sweep D — multi-publisher scaling (is the send ceiling per-thread?)

The payload sweep is bottlenecked on one publisher thread doing a serial fan-out send — which is why the CPU never pegs. So: does the byte-throughput ceiling scale if we add publisher threads? Here we run M independent single-publisher fan-out groups concurrently (each its own publisher + subscribers over its own endpoints) and measure aggregate delivered throughput over the shared window. If aggregate scales with M, the limit was per-core; if it plateaus, it's a shared bound — memory bandwidth and the per-subscriber copy, not the scheduler.

## M independent single-publisher groups, saturation, 64 KB frames, 8 subs each.
pubsD = [1, 2, 3, 4, 6, 8, 12]

function mp_point(np; tries=3)
    local r
    for t in 1:tries
        slate_progress(NaN; msg="$(np) publishers: run $t/$tries …")
        r = run_fanout_mp(; n_pubs=np, n_subs=8, n_msgs=800, payload=65536)
        r.dropped == 0 && return r
        GC.gc(true); sleep(0.3)
    end
    return r
end

mp_point (generic function with 1 method)

dfD = DataFrame(n_pubs=Int[], agg_mbps=Float64[], per_pub_mbps=Float64[], drop=Int[])
for (i, np) in enumerate(pubsD)
    slate_progress((i-1)/length(pubsD); msg="$(np) publishers …")
    r = mp_point(np)
    push!(dfD, (np, r.agg_mbps, r.per_pub_mbps, r.dropped))
    slate_progress(i/length(pubsD); msg="$(np) pub → $(round(Int,r.agg_mbps)) MB/s aggregate")
    GC.gc(true)
end
dfD;

n_pubs	agg_mbps	per_pub_mbps
1	1199	1199
2	1471	736
3	1785	595
4	2053	513
6	2377	396
8	2670	334
12	2849	237

Sweep E — fan-in (the ingestion / write path)

The mirror of fan-out: many producers stream to one consumer (N PUSH → 1 PULL). The single consumer is the bottleneck, so this measures how much one ingestion point can absorb. Sweeping the producer count at saturation, throughput climbs until the consumer saturates, then flattens — more producers can't push a single consumer any faster. (Producer tasks are cooperative @async, the correct ZMQ.jl model; ZMQ's own I/O threads still do the transfer.)

## N PUSH producers → 1 PULL consumer, saturation, 256 B messages.
producersE = [1, 2, 4, 8, 16, 32, 64]

function fanin_point(np; tries=3)
    local r
    for t in 1:tries
        slate_progress(NaN; msg="$(np) producers: run $t/$tries …")
        r = run_fanin(; n_producers=np, n_msgs=2000, payload=256, rate=0.0)
        r.dropped == 0 && return r
        GC.gc(true); sleep(0.3)
    end
    return r
end

fanin_point (generic function with 1 method)

dfE = DataFrame(n_producers=Int[], ingest_per_s=Float64[], mbps=Float64[], p50_ms=Float64[], drop=Int[])
for (i, np) in enumerate(producersE)
    slate_progress((i-1)/length(producersE); msg="$(np) producers …")
    r = fanin_point(np)
    push!(dfE, (np, r.ingest_per_s, r.mbps, r.p50, r.dropped))
    slate_progress(i/length(producersE); msg="$(np) prod → $(round(Int,r.ingest_per_s)) msg/s")
    GC.gc(true)
end
dfE;

Sweep F — topic-filtered pub/sub (rooms / channels)

The realistic version of broadcast: each subscriber listens to one of K topics (a chat room), and the publisher tags every message with a topic. ZMQ matches a byte-prefix and filters publisher-side (v3+), so a subscriber receives only its topic's ~n_msgs/K messages — the publisher never ships a message to a subscriber that didn't subscribe. Fixing 100 subscribers and sweeping the topic count: as rooms multiply, each subscriber's delivered volume falls as 1/K and total fan-out shrinks, while delivery stays lossless and low-latency — selective routing is cheap.

## 100 subscribers, sweep topic count, publisher paced at 5k msg/s, 256 B.
topicsF = [1, 2, 5, 10, 25, 50, 100]

function topic_point(k; tries=3)
    local r
    for t in 1:tries
        slate_progress(NaN; msg="$(k) topics: run $t/$tries …")
        r = run_topic(; n_subs=100, n_topics=k, n_msgs=5000, payload=256, rate=5000.0)
        r.delivered == r.expected && return r
        GC.gc(true); sleep(0.3)
    end
    return r
end

topic_point (generic function with 1 method)

dfF = DataFrame(n_topics=Int[], delivered=Int[], per_sub=Float64[], p50_ms=Float64[], p99_ms=Float64[])
for (i, k) in enumerate(topicsF)
    slate_progress((i-1)/length(topicsF); msg="$(k) topics …")
    r = topic_point(k)
    push!(dfF, (k, r.delivered, r.per_sub_avg, r.p50, r.p99))
    slate_progress(i/length(topicsF); msg="$(k) topics → $(round(Int,r.per_sub_avg))/sub, p99 $(round(r.p99;digits=1))ms")
    GC.gc(true)
end
dfF;

Sweep G — load-balanced work queue

The task-queue pattern: one producer, N workers, and each message goes to exactly one worker (PUSH round-robins) — the opposite of fan-out's each-to-all. This is how you distribute jobs across a worker pool. Sweeping the worker count at saturation, we measure distribution throughput and fairness — how evenly the round-robin splits the load (coefficient of variation across per-worker counts; 0 = perfectly even).

## 1 PUSH producer → N PULL workers (each msg to one worker), saturation, 256 B.
workersG = [1, 2, 4, 8, 16, 32]

function workq_point(nw; tries=3)
    local r
    for t in 1:tries
        slate_progress(NaN; msg="$(nw) workers: run $t/$tries …")
        r = run_workqueue(; n_workers=nw, n_msgs=20000, payload=256, rate=0.0)
        r.dropped == 0 && return r
        GC.gc(true); sleep(0.3)
    end
    return r
end

workq_point (generic function with 1 method)

dfG = DataFrame(n_workers=Int[], throughput=Float64[], fairness_cv=Float64[],
                min_count=Int[], max_count=Int[], p50_ms=Float64[])
for (i, nw) in enumerate(workersG)
    slate_progress((i-1)/length(workersG); msg="$(nw) workers …")
    r = workq_point(nw)
    push!(dfG, (nw, r.throughput, r.fairness_cv, r.min_count, r.max_count, r.p50))
    slate_progress(i/length(workersG); msg="$(nw) workers → $(round(Int,r.throughput)) msg/s, cv=$(round(r.fairness_cv;digits=3))")
    GC.gc(true)
end
dfG;

Takeaways

(Figures computed from this run. The saturation numbers vary with machine load — read them as order-of-magnitude ceilings, not constants. All patterns are lossless in these runs.)

Two throughput regimes (payload). Small frames are message-rate-bound — 64 B reaches ~3.0 M msg/s but only ~190 MB/s. Large frames are bandwidth-bound — throughput plateaus around ~2.8 GB/s aggregate (~352 MB/s per subscriber stream) at 1 MB frames. It is all driven by one publisher thread doing a serial fan-out send, so it never saturates more than a core.
The send ceiling is partly per-thread, but bandwidth-bound overall. Running many concurrent publisher groups lifts aggregate throughput to only ~2.4× the single-publisher number, plateauing at a shared ~2.8 GB/s — memory bandwidth + the per-subscriber copy, not the scheduler.
The fan-out knee (ZMQ vs in-process). At 500 subscribers × 256 B, ZMQ ipc:// keeps up to 2500 msg/s and tops out at ~1.81 M deliveries/s (p99 ~140→1113 ms across the knee). The pure-Julia Channel baseline is far lower-latency below the knee (p50 ~0.23 ms, p99 4 ms) but saturates earlier (2500 msg/s, ~1.34 M deliveries/s) and its tail explodes to ~1.5 s. In-process wins on latency; ZMQ's offloaded fan-out wins on peak throughput.
Request/Reply (interactive path). A single round-trip over ipc is ~0.13 ms; one ROUTER server saturates at ~27056 req/s (~25 clients), then added concurrency only grows round-trip latency (Little's Law) — p50 ~18.0 ms at 400 clients while throughput holds.
Ingestion (fan-in). A single PULL consumer absorbs up to ~3.6 M msg/s (256 B), reached with just a couple of producers — consumer-bound, so more producers don't help.
Topic routing (rooms/channels). ZMQ filters publisher-side: 100 subscribers split across K topics each receive exactly 1/K of the stream (down to 50/sub at 100 topics), lossless, and latency stays low — selective routing costs nothing extra.
Work queue (load balancing). PUSH round-robin delivers each message to exactly one worker with perfect fairness (CV = 0) at every worker count; throughput (~2.8 M msg/s) is bounded by the single producer, not the workers.
The Julia-specific caution. Fan-out's 500 green-threaded subscribers show elevated tails (p99 ~140 ms at 1 k msg/s) from scheduler + GC overhead. And ZMQ here must be driven from cooperative @async tasks, never Threads.@spawn — multithreaded socket I/O corrupts/crashes it. That thread-discipline is the real cost of ZMQ-in-Julia for a messaging backend.
Scope. In-box IPC proxy — no network stack, no TLS. Not a cross-language comparison, not TCP connection-scaling. It characterises Julia's scheduler + ZMQ + GC under each messaging shape.