Tokio mpsc: bounded send vs try_send vs timeout
The first time I shipped a bounded mpsc to production, I picked capacity 8 because the example in the docs picked capacity 8. Two weeks later a downstream consumer hiccuped for ninety seconds and our ingest service quietly stopped accepting webhooks — every producer task was parked inside send().await, waiting politely for slots that were not coming back. The fix took twenty minutes. Understanding why I had built a distributed deadlock out of three lines of Tokio took a weekend, a notebook, and a load generator that finally let me see what each send variant actually does when the channel is full.
This article is that weekend, distilled into a runnable tokio-mpsc-lab project you can cargo new and follow commit by commit. We will build a producer/consumer harness in Rust on top of tokio and tracing, wire a bounded channel at capacity 8, then implement and benchmark all three send paths — send().await, try_send, and send_timeout — against an identical workload, emitting CSV metrics for throughput, p99 latency, drop rate, and producer-stall time. By the end you will have numbers, not opinions, about which path belongs at which call-site. The rule that fell out of the lab and now lives above my desk: a bounded channel is a contract about what your producer does when the consumer is slow, and send().await is the only variant that signs the contract in your producer's blood.
If you write Rust services that move messages between async tasks — request fan-out, work queues, telemetry pipelines, anything that touches tokio::sync::mpsc — this walkthrough will leave you able to pick a send strategy per call-site with a real reason, combine try_send with a fallback queue without losing data silently, and shut the whole thing down gracefully when the receiver finally closes.
Step 1: Bootstrapping the Producer/Consumer Lab with Tokio + Tracing
The whole point of this article is to compare three ways of pushing
messages onto a tokio::sync::mpsc channel — send().await,
try_send, and send_timeout — under identical workloads. Before we
can compare anything we need a rig that holds the channel shape, the
producer, the consumer, and the metrics in one place, so that later
steps only swap the send call and leave everything else untouched.
This step ships that rig. We create a fresh tokio-mpsc-lab crate,
wire in the smallest tokio feature set we need, and define a
HarnessConfig / HarnessReport pair plus a run_baseline driver
that joins a producer task and a consumer task. We also bolt on
tracing early — channel debugging is miserable without spans, and
adding it later means rewriting every helper.
Setup
Create a fresh binary-free library crate so the harness can be unit
tested without dragging a main.rs along. The repo layout after this
step is:
tokio-mpsc-lab/
├── Cargo.toml
├── Cargo.lock
└── src/
└── lib.rs
The Cargo.toml pins tokio's multi-thread runtime, macros, the sync
primitives, and the timer wheel. We will need time from step 4
onward for send_timeout, but enabling it now keeps the feature set
stable across the article. tracing and tracing-subscriber go in as
regular dependencies because the test helpers also want to emit logs.
[package]
name = "tokio-mpsc-lab"
version = "0.1.0"
edition = "2021"
description = "Lab for comparing tokio mpsc send / try_send / send_timeout under backpressure."
license = "MIT OR Apache-2.0"
[dependencies]
tokio = { version = "1.40", features = ["rt-multi-thread", "macros", "sync", "time"] }
tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["env-filter", "fmt"] }
[dev-dependencies]
tokio = { version = "1.40", features = ["rt-multi-thread", "macros", "sync", "time", "test-util"] }
The dev-dependency block re-imports tokio with test-util so that the
async test attribute and the future-pacing helpers are available
without bleeding into the public surface.
Implementation
The crate's public surface is small on purpose. A Message carries an
ordered sequence number plus an opaque payload — enough for the
consumer to assert that nothing was reordered, and enough payload to
make queueing visible without dominating runtime cost.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Message {
pub seq: u64,
pub payload: Vec<u8>,
}
impl Message {
pub fn new(seq: u64, payload: Vec<u8>) -> Self {
Self { seq, payload }
}
}
We model the channel strategy as an enum even though step 1 only has
one variant. This is deliberate — when steps 3 and 4 add TrySend and
SendTimeout, callers will already be matching on this type, so the
diff stays small. HarnessConfig then bundles every dimension we want
to sweep: queue capacity, total messages, payload size, and the
strategy itself.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Strategy {
Send,
}
#[derive(Debug, Clone)]
pub struct HarnessConfig {
pub capacity: usize,
pub total_messages: u64,
pub payload_size: usize,
pub strategy: Strategy,
}
impl Default for HarnessConfig {
fn default() -> Self {
Self {
capacity: 8,
total_messages: 64,
payload_size: 16,
strategy: Strategy::Send,
}
}
}
The default capacity: 8 is small on purpose. Bounded channels are
only interesting when the producer can outrun the consumer, and a
buffer of eight messages with a 16-byte payload guarantees the
producer will park at least once on a slow consumer.
Tracing initialisation is gated behind a Once so that running every
test in the same process does not register half a dozen subscribers.
Tests opt in by calling init_tracing() at the top — the function
respects RUST_LOG but falls back to tokio_mpsc_lab=info so the
output stays readable.
static TRACING_INIT: Once = Once::new();
pub fn init_tracing() {
TRACING_INIT.call_once(|| {
let filter = EnvFilter::try_from_default_env()
.unwrap_or_else(|_| EnvFilter::new("tokio_mpsc_lab=info"));
let _ = tracing_subscriber::fmt()
.with_env_filter(filter)
.with_test_writer()
.try_init();
});
}
The producer is the simplest possible loop over 0..total_messages.
It builds a Message, awaits the bounded send, and bails out the
instant the receiver closes the channel. We count successful sends
separately from the loop index so step 3 can report partial delivery
when try_send starts dropping work.
#[instrument(skip(tx), fields(total = total_messages, payload_size))]
pub async fn producer(
tx: mpsc::Sender<Message>,
total_messages: u64,
payload_size: usize,
) -> u64 {
let mut sent = 0u64;
for seq in 0..total_messages {
let msg = Message::new(seq, vec![0u8; payload_size]);
if tx.send(msg).await.is_err() {
info!(at = seq, "consumer closed channel; stopping producer");
break;
}
sent += 1;
}
info!(sent, "producer finished");
sent
}
The consumer is symmetric. It pulls from the receiver until recv
returns None, asserts that the sequence numbers arrive in order
under debug_assert_eq!, and returns the total count. The order
check is cheap insurance — if a future strategy ever reorders work,
the assertion in debug builds will scream.
#[instrument(skip(rx))]
pub async fn consumer(mut rx: mpsc::Receiver<Message>) -> u64 {
let mut received = 0u64;
while let Some(msg) = rx.recv().await {
debug_assert_eq!(msg.seq, received);
received += 1;
}
info!(received, "consumer drained channel");
received
}
Finally run_baseline glues the two halves together. It builds the
bounded channel from the config, spawns both tasks, awaits them, and
stamps the elapsed wall-clock onto the report. Joining producer first
is fine because tokio's mpsc closes the channel when the last sender
drops, which naturally terminates the consumer.
#[instrument(skip(config), fields(
capacity = config.capacity,
total = config.total_messages,
strategy = ?config.strategy,
))]
pub async fn run_baseline(config: HarnessConfig) -> HarnessReport {
let (tx, rx) = mpsc::channel::<Message>(config.capacity);
let started = Instant::now();
let producer_task: JoinHandle<u64> =
tokio::spawn(producer(tx, config.total_messages, config.payload_size));
let consumer_task: JoinHandle<u64> = tokio::spawn(consumer(rx));
let produced = producer_task.await.expect("producer task panicked");
let consumed = consumer_task.await.expect("consumer task panicked");
HarnessReport {
produced,
consumed,
elapsed: started.elapsed(),
}
}
Four unit tests pin the invariants we will rely on for the rest of the article. The first runs the default workload and asserts that every produced message is consumed. The second pushes capacity down to one — the most adversarial bounded queue — and confirms that all 32 messages still round-trip. The third drops the receiver before the producer starts and asserts that no messages land. The fourth simply checks that the default config is what we documented.
Verification
cargo test
Finished `test` profile [unoptimized + debuginfo] target(s) in 0.72s
Running unittests src/lib.rs (target/debug/deps/tokio_mpsc_lab-faf64ef3387c85e5)
running 4 tests
test tests::default_config_uses_capacity_eight ... ok
test tests::producer_stops_when_consumer_drops_receiver ... ok
test tests::baseline_harness_delivers_every_message ... ok
test tests::capacity_one_still_drains_all_messages ... ok
test result: ok. 4 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.02s
Doc-tests tokio_mpsc_lab
running 0 tests
test result: ok. 0 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.00s
All four tests pass. The harness delivers every message under the default config, survives the capacity-one stress test, and bails cleanly when the receiver disappears.
What we built
A minimal but real producer/consumer rig sitting on top of
tokio::sync::mpsc::channel. The shape — HarnessConfig →
run_baseline → HarnessReport — is the same shape every later
strategy will reuse, so the comparison stays apples-to-apples.
A Strategy enum with one variant. Adding TrySend and
SendTimeout later becomes a routine match-arm extension instead of
a refactor, and any caller that already pattern-matched on the enum
will get a compile error to remind them to handle the new arm.
A tracing setup that is process-safe across tests. Every async
function carries an #[instrument] span, so when the producer parks
on a full channel in step 2 the wakeup latency will already be
visible in the log output — no extra plumbing needed.
A regression net of four tests covering the happy path, the worst-case capacity, the receiver-drop edge case, and the default config. These are the tests the next steps will keep green as the strategy enum grows.
Repository
The state of the code after this step: 2e658cc
Step 2: Per-Send Latency Capture with Spans and a LatencyStats Sink
Step 1 left us with a producer/consumer rig that knows how many messages
made it across, but nothing about how long each send took. That number
is the whole point of the comparison — when we plug try_send and
send_timeout in later, the only honest way to argue about backpressure
behaviour is to look at the per-send latency distribution under
identical workloads.
This step bolts a latency-capture layer onto the baseline send().await
path. We add a LatencyStats sink, change the producer to stamp
Instant::now() around every call and emit a dedicated send span per
attempt, and surface the samples on HarnessReport. Two new tests pin
the invariants we care about: a slow consumer must inflate latency, and
a fast consumer must not.
Setup
No new dependencies are needed — tokio, tracing, and
tracing-subscriber already ship from step 1. All edits land in
src/lib.rs. The new public surface is two structs (LatencyStats,
ProducerOutcome) and one new field on HarnessReport
(send_latencies). We also import a few more items from tracing to
build per-send spans:
use tracing::{info, info_span, instrument, trace, Instrument};
The Instrument trait is the one that lets us attach an arbitrary span
to a future via .instrument(span), which is the cleanest way to keep
each send attempt under its own span without rewriting the producer
as a state machine.
Implementation
The first new type is the sink itself. LatencyStats is intentionally
a thin wrapper around Vec<Duration> — we want raw samples available
for later histogram work, plus three convenience reductions
(count, max, min, mean) so unit tests can assert on aggregate
shape without re-implementing the math.
#[derive(Debug, Clone, Default)]
pub struct LatencyStats {
samples: Vec<Duration>,
}
impl LatencyStats {
fn record(&mut self, latency: Duration) {
self.samples.push(latency);
}
pub fn count(&self) -> usize { self.samples.len() }
pub fn samples(&self) -> &[Duration] { &self.samples }
pub fn max(&self) -> Option<Duration> { self.samples.iter().copied().max() }
pub fn min(&self) -> Option<Duration> { self.samples.iter().copied().min() }
pub fn mean(&self) -> Option<Duration> {
if self.samples.is_empty() {
return None;
}
let total: Duration = self.samples.iter().copied().sum();
Some(total / self.samples.len() as u32)
}
}
record is crate-private on purpose. Only the producer is allowed to
push samples; consumers of LatencyStats see an immutable view. Mean
returns Option<Duration> because an empty sample set has no meaningful
mean, and 0 would be a misleading default later when we tabulate
results across strategies.
Next we split the producer's return value into a dedicated
ProducerOutcome. Returning a bare u64 worked when "sent count" was
the only thing worth knowing, but step 2 needs to ship the latency
samples back from the producer task, and bundling them in a struct
keeps the join site readable.
#[derive(Debug, Clone)]
pub struct ProducerOutcome {
pub sent: u64,
pub latencies: LatencyStats,
}
#[derive(Debug, Clone)]
pub struct HarnessReport {
pub produced: u64,
pub consumed: u64,
pub elapsed: Duration,
pub send_latencies: LatencyStats,
}
The producer loop is where the real work happens. For every iteration
we build a per-send span carrying the sequence number and payload size,
attach that span to the send future via .instrument(send_span.clone()),
stamp the wall-clock before and after the await, and trace the latency
inside the span before deciding whether to record it.
#[instrument(skip(tx), fields(total = total_messages, payload_size))]
pub async fn producer(
tx: mpsc::Sender<Message>,
total_messages: u64,
payload_size: usize,
) -> ProducerOutcome {
let mut sent = 0u64;
let mut latencies = LatencyStats::default();
for seq in 0..total_messages {
let msg = Message::new(seq, vec![0u8; payload_size]);
let send_span = info_span!("send", seq, payload = payload_size);
let started = Instant::now();
let send_result = tx.send(msg).instrument(send_span.clone()).await;
let latency = started.elapsed();
send_span.in_scope(|| {
trace!(?latency, "send completed");
});
if send_result.is_err() {
info!(at = seq, "consumer closed channel; stopping producer");
break;
}
latencies.record(latency);
sent += 1;
}
info!(sent, samples = latencies.count(), "producer finished");
ProducerOutcome { sent, latencies }
}
Three subtle decisions live in this block. First, the latency is
measured around the await, not around tx.send(msg) alone — that is
deliberate, because the await is where the producer parks on a full
queue, which is the exact phenomenon we want to surface. Second, we
record the sample only on success, so a closed-channel error never
contaminates the distribution. Third, the per-send span is cloned so
both the future and the post-await trace! can share it, which keeps
the latency event nested under the same span as the actual send.
run_baseline only needs the smallest possible diff: pull the latency
sink off ProducerOutcome and hand it to HarnessReport. Nothing else
in the join shape changes, which is exactly what step 1's Strategy
enum was designed to allow.
pub async fn run_baseline(config: HarnessConfig) -> HarnessReport {
let (tx, rx) = mpsc::channel::<Message>(config.capacity);
let started = Instant::now();
let producer_task: JoinHandle<ProducerOutcome> =
tokio::spawn(producer(tx, config.total_messages, config.payload_size));
let consumer_task: JoinHandle<u64> = tokio::spawn(consumer(rx));
let outcome = producer_task.await.expect("producer task panicked");
let consumed = consumer_task.await.expect("consumer task panicked");
HarnessReport {
produced: outcome.sent,
consumed,
elapsed: started.elapsed(),
send_latencies: outcome.latencies,
}
}
The existing tests get a one-line tightening — they now assert that the
number of latency samples equals the number of successful sends, which
catches the easy bug where record(..) accidentally moves above the
error check. The receiver-drop test additionally asserts
outcome.latencies.count() == 0, locking in the rule that failed sends
never pollute the distribution.
Two brand-new tests cover the actual signal we care about. The first parks the producer behind a deliberately slow consumer (capacity 1, 5ms of sleep per recv) and asserts that the max sample is at least 1ms — proving the latency channel is wired to the right thing. The second runs a fast consumer over a roomy capacity-16 channel and asserts the mean stays under 10ms, ruling out a noisy baseline.
#[tokio::test]
async fn slow_consumer_inflates_send_latency() {
init_tracing();
let (tx, mut rx) = mpsc::channel::<Message>(1);
let producer_task = tokio::spawn(producer(tx, 8, 0));
let consumer_task = tokio::spawn(async move {
let mut received = 0u64;
while let Some(_msg) = rx.recv().await {
tokio::time::sleep(Duration::from_millis(5)).await;
received += 1;
}
received
});
let outcome = producer_task.await.expect("producer panicked");
let received = consumer_task.await.expect("consumer panicked");
assert_eq!(outcome.sent, 8);
assert_eq!(received, 8);
let max = outcome
.latencies
.max()
.expect("expected at least one latency sample");
assert!(max >= Duration::from_millis(1));
}
The lower bound of 1ms is intentionally loose. We are not asserting a specific scheduling delay — we are asserting that backpressure shows up in the samples at all. Tightening the bound to, say, 4ms would couple the test to the consumer's sleep duration and create flake risk on loaded CI runners.
Verification
cargo test
Finished `test` profile [unoptimized + debuginfo] target(s) in 0.81s
Running unittests src/lib.rs (target/debug/deps/tokio_mpsc_lab-faf64ef3387c85e5)
running 6 tests
test tests::default_config_uses_capacity_eight ... ok
test tests::producer_stops_when_consumer_drops_receiver ... ok
test tests::fast_consumer_keeps_send_latency_small ... ok
test tests::baseline_harness_delivers_every_message ... ok
test tests::capacity_one_still_drains_all_messages ... ok
test tests::slow_consumer_inflates_send_latency ... ok
test result: ok. 6 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.06s
Doc-tests tokio_mpsc_lab
running 0 tests
test result: ok. 0 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.00s
All six tests pass. The four step-1 invariants still hold, the slow consumer test confirms backpressure shows up in the latency samples, and the fast consumer test pins down a quiet baseline so future regressions are obvious.
What we built
A LatencyStats sink that owns the raw Vec<Duration> of per-send
samples and exposes count, max, min, and mean. The samples
themselves stay accessible via samples() so later steps can plug a
percentile reducer or a histogram in without changing the producer.
A producer that emits one info_span!("send", seq, payload) per
iteration, instruments the send future with that span, and records a
latency sample only when the send succeeds. Failed sends never enter
the distribution, which keeps the comparison across strategies honest
once try_send starts rejecting messages in step 3.
An updated HarnessReport that carries send_latencies alongside
produced, consumed, and elapsed. Every downstream report — for
Send, TrySend, SendTimeout — will read the same shape, so the
final comparison table is one struct away.
Two new regression tests that pin the signal direction itself: latency
must rise under a deliberately slow consumer, and it must stay small
under a roomy fast consumer. With both edges nailed down, step 3 can
swap in try_send confident that any new behaviour shows up as a real
change in the samples, not as a measurement artefact.
Repository
The state of the code after this step: 062ad88
Step 3: Turning send().await Parks Into a First-Class Metric with poll_fn
Step 2 left us with a LatencyStats channel that shows how long each
send().await blocks, but it cannot tell us why it blocked. A 5ms
sample could mean the producer parked once on a full queue and slept
through one consumer cycle, or it could mean the runtime was busy and
the future was repolled three times before the channel had room. Those
two stories have very different implications for tuning capacity, and
the only way to tell them apart is to count Poll::Pending returns
directly.
This step adds that count. We introduce a ParkStats sink alongside
LatencyStats, wrap the existing tx.send(msg) future in a poll_fn
that increments a counter every time the underlying future yields, and
surface the per-send park count on HarnessReport. Three new tests pin
the semantics: a slow consumer must park the producer repeatedly, an
over-sized buffer must keep every send on the fast path, and even a
single-permit channel must park at least once when the consumer runs in
a separate task.
Setup
No new crates are needed. tokio, tracing, and tracing-subscriber
already cover everything the producer touches, and std::future::poll_fn
has been stable since Rust 1.64 — well below the 2021-edition baseline
the project already pins. All edits land in src/lib.rs.
The public surface grows by one type (ParkStats) and one field on the
existing ProducerOutcome and HarnessReport structs (park_stats).
We also pull two items out of std::future so the new helper has the
poll plumbing it needs:
use std::future::{poll_fn, Future};
The Future import is there because we will pin a typed future and
call .as_mut().poll(cx) on it by hand. poll_fn is the wrapper that
turns a closure-of-Poll into a future without forcing us to declare a
named future type for one helper.
Implementation
ParkStats is a small aggregate sink. It records, per successful send,
how many times the send future returned Pending before completing,
and rolls those samples into four reductions the tests and later steps
will care about: fast-path sends, parked sends, total parks, and the
raw per-send sequence.
#[derive(Debug, Clone, Default)]
pub struct ParkStats {
fast_path_sends: u64,
parked_sends: u64,
total_parks: u64,
per_send_parks: Vec<u64>,
}
impl ParkStats {
fn record(&mut self, parks: u64) {
self.per_send_parks.push(parks);
self.total_parks += parks;
if parks == 0 {
self.fast_path_sends += 1;
} else {
self.parked_sends += 1;
}
}
}
record is crate-private — only the producer is allowed to push
samples — and the public reductions return owned u64 values rather
than references, so callers cannot accidentally hold a borrow across an
.await. The binary split between fast_path_sends (parks == 0) and
parked_sends (parks ≥ 1) gives us a yes/no answer to "did this send
ever wait?" without throwing away the raw counts.
The next piece is the helper that actually drives the send future and
counts its Pending returns. We pin the instrumented send future onto
the stack, then poll it from inside a poll_fn closure, incrementing a
local counter every time the inner poll returns Pending. When the
future eventually returns Ready, the helper unpacks its three outputs
together: the send result, the elapsed wall-clock, and the park count.
async fn send_with_park_count(
tx: &mpsc::Sender<Message>,
msg: Message,
send_span: tracing::Span,
) -> (Result<(), mpsc::error::SendError<Message>>, Duration, u64) {
let send_fut = tx.send(msg).instrument(send_span);
tokio::pin!(send_fut);
let started = Instant::now();
let mut parks: u64 = 0;
let result = poll_fn(|cx| {
let polled = send_fut.as_mut().poll(cx);
if polled.is_pending() {
parks += 1;
}
polled
})
.await;
(result, started.elapsed(), parks)
}
Two design choices live in this helper. First, we count Pending
rather than wakeups directly — every Pending return is paired with a
later wakeup that drove the next poll, so the count is the same number
but cheaper to observe. Second, the helper does not own the span; it
takes one in and attaches it to the inner future, so the existing
per-send tracing structure from step 2 survives untouched.
The producer loop swaps the inline tx.send(msg).await for a call to
the new helper and feeds the returned parks count into ParkStats.
Recording is gated behind the same success check as the latency sample,
so a closed-channel error never contaminates either distribution.
let (send_result, latency, parks) =
send_with_park_count(&tx, msg, send_span.clone()).await;
send_span.in_scope(|| {
trace!(?latency, parks, "send completed");
});
if send_result.is_err() {
info!(at = seq, "consumer closed channel; stopping producer");
break;
}
latencies.record(latency);
park_stats.record(parks);
sent += 1;
run_baseline only needs to plumb the new field through. It reads
outcome.park_stats off the joined producer outcome and stamps it onto
HarnessReport next to send_latencies. That keeps the report shape
uniform — every future strategy will produce the same five fields, so
the cross-strategy comparison stays one struct away.
Three new tests pin the signal. The first runs a deliberately slow
consumer (capacity=1, 5ms sleep per recv) and asserts that at least
six of the eight sends park; the second runs a fat-buffered fast
consumer and asserts that every send hits the fast path with zero
parks; the third runs a capacity=1 workload through run_baseline
and asserts at least one park, proving the harness itself exercises
backpressure rather than accidentally serialising everything onto one
task.
#[tokio::test]
async fn slow_consumer_parks_producer_repeatedly() {
init_tracing();
let (tx, mut rx) = mpsc::channel::<Message>(1);
let producer_task = tokio::spawn(producer(tx, 8, 0));
let consumer_task = tokio::spawn(async move {
let mut received = 0u64;
while let Some(_msg) = rx.recv().await {
tokio::time::sleep(Duration::from_millis(5)).await;
received += 1;
}
received
});
let outcome = producer_task.await.expect("producer panicked");
let received = consumer_task.await.expect("consumer panicked");
assert_eq!(outcome.sent, 8);
assert_eq!(received, 8);
let parked = outcome.park_stats.parked_sends();
assert!(parked >= 6, "expected at least 6 parked sends (got {parked})");
}
The lower bound of six (rather than eight) is intentional. The first
two sends can each grab the single permit immediately, before the
consumer has even been scheduled, so they legitimately complete on the
fast path. Asserting >= 6 rather than >= 8 captures the
backpressure signal without coupling the test to scheduler ordering
quirks on a loaded CI runner.
Verification
cargo test
Finished `test` profile [unoptimized + debuginfo] target(s) in 0.19s
Running unittests src/lib.rs (target/debug/deps/tokio_mpsc_lab-faf64ef3387c85e5)
running 9 tests
test tests::default_config_uses_capacity_eight ... ok
test tests::producer_stops_when_consumer_drops_receiver ... ok
test tests::capacity_one_forces_at_least_one_park ... ok
test tests::fast_consumer_keeps_producer_on_fast_path ... ok
test tests::fast_consumer_keeps_send_latency_small ... ok
test tests::capacity_one_still_drains_all_messages ... ok
test tests::baseline_harness_delivers_every_message ... ok
test tests::slow_consumer_inflates_send_latency ... ok
test tests::slow_consumer_parks_producer_repeatedly ... ok
test result: ok. 9 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.22s
Doc-tests tokio_mpsc_lab
running 0 tests
test result: ok. 0 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.00s
All nine tests pass. The six invariants from steps 1 and 2 still hold,
and the three new wakeup-aware tests confirm that send().await parks
exactly where the channel runs out of permits and stays on the fast
path everywhere else.
What we built
A ParkStats sink that owns per-send Poll::Pending counts and
exposes the four reductions later steps care about: fast_path_sends,
parked_sends, total_parks, and max_parks_per_send. The sink lives
next to LatencyStats on ProducerOutcome, so every successful send
contributes exactly one latency sample and exactly one park sample.
A send_with_park_count helper that wraps the existing instrumented
send future in a poll_fn. Because the helper polls the same future
the producer was already awaiting, it observes the runtime's wakeup
behaviour without forking tokio's internals or paying any allocation
cost beyond a single counter on the stack.
An updated HarnessReport that now carries both distributions —
send_latencies and park_stats — alongside produced, consumed,
and elapsed. The next two steps will swap try_send and
send_timeout into the same producer shape and emit a report with
exactly the same fields, so the final cross-strategy comparison
collapses to reading the same struct three times.
Three new tests that pin the backpressure signal in both directions.
The slow-consumer test proves that send().await parks at least six of
eight sends when the channel is permit-starved, the fat-buffer test
proves that every send hits the fast path when capacity comfortably
exceeds workload, and the capacity=1 harness test proves the rig
itself exercises real parking rather than accidentally serialising
producer and consumer.
Repository
The state of the code after this step: dbd8dcf
Step 4: Plugging in try_send with a Drop-Newest Path and a Drop-Oldest Ring Buffer
Step 3 left us with a send().await baseline that knows exactly how
long each send waited and how many Poll::Pending returns the future
emitted before completing. That tells us everything about parking,
but it tells us nothing about a producer that refuses to park at all.
Most real backpressure stories want a third option: when the queue is
full, do not block — make a policy decision and keep moving.
This step ships that option. We grow the Strategy enum with
TrySend and TrySendDropOldest, add a TrySendStats sink that
counts TrySendError::Full separately from TrySendError::Closed,
and wire two new producer entry points into the existing
run_baseline switch. The non-blocking variants reuse the
LatencyStats and ParkStats channels from earlier steps so all
three strategies will land on the same scoreboard once the comparison
step arrives.
Setup
No new crates. tokio::sync::mpsc::error::TrySendError lives in the
sync feature we already enabled in step 1, and the producer-side
ring buffer uses std::collections::VecDeque from the standard
library. All edits land in src/lib.rs. Two new imports cover the new
machinery:
use std::collections::VecDeque;
use tokio::sync::mpsc::error::TrySendError;
The public surface grows by two Strategy variants
(TrySend, TrySendDropOldest), one config field
(drop_oldest_buffer), one stats struct (TrySendStats), and one
optional field on the existing ProducerOutcome and HarnessReport
(try_stats: Option<TrySendStats>). The Option is deliberate —
Strategy::Send has no try-send observations to report, and an
Option is cleaner than zero-filled defaults that lie about whether
the strategy ran.
Implementation
TrySendStats is the new sink. It counts Full and Closed
observations separately, splits drops by reason (dropped_full for
the drop-newest path, dropped_oldest for the ring-buffer eviction),
and tracks the high-water mark of the producer-side buffer. Every
counter is pub because, unlike the latency and park sinks, this
struct is pure summary data and there is no internal invariant to
guard.
#[derive(Debug, Clone, Default)]
pub struct TrySendStats {
pub full_observations: u64,
pub closed_observations: u64,
pub dropped_full: u64,
pub dropped_oldest: u64,
pub max_buffer_depth: usize,
}
The separation between full_observations and dropped_full matters
once the ring-buffer path joins the picture. In the simple TrySend
producer they are identical — every Full means a drop. In the
drop-oldest producer the same message can re-park the buffer many
times before draining, so full_observations becomes the larger
"transient pressure" number while dropped_full stays zero (the
drop-oldest path never throws new messages away — it evicts old
ones, accounted under dropped_oldest).
The first new producer is the drop-newest variant. Its loop is one
synchronous call to tx.try_send(msg) followed by a classification
helper that decides whether to count the message as sent, dropped, or
the loop-terminating Closed signal. A cooperative yield_now
between iterations stops the producer from monopolising a
single-threaded runtime.
#[instrument(skip(tx), fields(total = total_messages, payload_size, strategy = "try_send"))]
pub async fn producer_try_send(
tx: mpsc::Sender<Message>,
total_messages: u64,
payload_size: usize,
) -> ProducerOutcome {
let mut sent = 0u64;
let mut latencies = LatencyStats::default();
let mut park_stats = ParkStats::default();
let mut try_stats = TrySendStats::default();
for seq in 0..total_messages {
let msg = Message::new(seq, vec![0u8; payload_size]);
let started = Instant::now();
let outcome = tx.try_send(msg);
if !record_try_send_outcome(
outcome,
started,
seq,
&mut sent,
&mut latencies,
&mut park_stats,
&mut try_stats,
) {
break;
}
tokio::task::yield_now().await;
}
info!(
sent,
dropped = try_stats.dropped_full,
full = try_stats.full_observations,
"try_send producer finished",
);
ProducerOutcome {
sent,
latencies,
park_stats,
try_stats: Some(try_stats),
}
}
The classification helper is a small free function rather than a
method on TrySendStats because it touches four sinks at once. Each
arm of the match is a different kind of event, and pulling them out
of the loop body keeps the producer readable while honouring the
codebase rule against deeply nested control flow.
fn record_try_send_outcome(
outcome: Result<(), TrySendError<Message>>,
started: Instant,
seq: u64,
sent: &mut u64,
latencies: &mut LatencyStats,
park_stats: &mut ParkStats,
try_stats: &mut TrySendStats,
) -> bool {
match outcome {
Ok(()) => {
*sent += 1;
latencies.record(started.elapsed());
park_stats.record(0);
true
}
Err(TrySendError::Full(_)) => {
try_stats.full_observations += 1;
try_stats.dropped_full += 1;
trace!(seq, "try_send saw Full; dropping new message");
true
}
Err(TrySendError::Closed(_)) => {
try_stats.closed_observations += 1;
info!(at = seq, "try_send saw Closed; stopping producer");
false
}
}
}
Three details are worth pinning. First, the helper still records a
latency sample on success — a synchronous try_send is not free, and
having the same LatencyStats channel in scope lets the cross-strategy
comparison treat fast-path latency as a fair number. Second, every
successful try_send records park_stats.record(0) because
try_send is fully synchronous and never registers a waker; reusing
the same sink keeps fast_path_sends + parked_sends == sent as an
invariant across all three strategies. Third, Closed returns false
to break the producer loop; Full returns true so the loop keeps
running and we keep observing pressure.
The second new producer is the drop-oldest fallback. It keeps a small
VecDeque<Message> between the producer and the channel: every fresh
message is pushed onto the back, the front is evicted whenever the
buffer overflows, and the loop drains as much of the buffer as
try_send will accept on each iteration.
#[instrument(skip(tx), fields(total = total_messages, payload_size, strategy = "drop_oldest", buf = buffer_capacity))]
pub async fn producer_try_send_drop_oldest(
tx: mpsc::Sender<Message>,
total_messages: u64,
payload_size: usize,
buffer_capacity: usize,
) -> ProducerOutcome {
let mut buffer: VecDeque<Message> = VecDeque::with_capacity(buffer_capacity.max(1));
let mut sent = 0u64;
let mut latencies = LatencyStats::default();
let mut park_stats = ParkStats::default();
let mut try_stats = TrySendStats::default();
let mut closed = false;
for seq in 0..total_messages {
let msg = Message::new(seq, vec![0u8; payload_size]);
enqueue_drop_oldest(&mut buffer, msg, buffer_capacity, &mut try_stats);
let flush = flush_buffer_try_send(
&tx,
&mut buffer,
&mut sent,
&mut latencies,
&mut park_stats,
&mut try_stats,
);
if matches!(flush, FlushOutcome::Closed) {
closed = true;
break;
}
tokio::task::yield_now().await;
}
if !closed {
drain_buffer_awaiting(
&tx,
&mut buffer,
&mut sent,
&mut latencies,
&mut park_stats,
&mut try_stats,
)
.await;
}
ProducerOutcome { sent, latencies, park_stats, try_stats: Some(try_stats) }
}
The loop body delegates to three small helpers — enqueue_drop_oldest,
flush_buffer_try_send, and drain_buffer_awaiting — so the producer
itself stays under the nesting budget. The post-loop drain switches
from try_send back to send().await so anything still resident in
the buffer when the input stream ends gets a chance to land before the
channel closes.
fn enqueue_drop_oldest(
buffer: &mut VecDeque<Message>,
msg: Message,
buffer_capacity: usize,
try_stats: &mut TrySendStats,
) {
buffer.push_back(msg);
if buffer.len() > buffer_capacity {
let dropped = buffer.pop_front();
if dropped.is_some() {
try_stats.dropped_oldest += 1;
}
}
if buffer.len() > try_stats.max_buffer_depth {
try_stats.max_buffer_depth = buffer.len();
}
}
enqueue_drop_oldest is the eviction policy in three lines. We always
append the new message first, then check the overflow condition, then
update the buffer high-water mark. Doing the append before the evict
is what makes this a drop-oldest policy rather than a reject-newest
one — even when the buffer is already full, the newest message wins
the slot and the oldest one is paid for.
fn flush_buffer_try_send(
tx: &mpsc::Sender<Message>,
buffer: &mut VecDeque<Message>,
sent: &mut u64,
latencies: &mut LatencyStats,
park_stats: &mut ParkStats,
try_stats: &mut TrySendStats,
) -> FlushOutcome {
while let Some(front) = buffer.pop_front() {
let started = Instant::now();
match tx.try_send(front) {
Ok(()) => {
*sent += 1;
latencies.record(started.elapsed());
park_stats.record(0);
}
Err(TrySendError::Full(returned)) => {
buffer.push_front(returned);
try_stats.full_observations += 1;
return FlushOutcome::Continue;
}
Err(TrySendError::Closed(_)) => {
try_stats.closed_observations += 1;
return FlushOutcome::Closed;
}
}
}
FlushOutcome::Continue
}
flush_buffer_try_send is where the Full(returned) payload of
TrySendError finally earns its keep. The error variant hands the
message back to the caller — owning the failed payload is exactly what
lets us push it back onto the front of the buffer and retry on the
next iteration, without cloning. full_observations increments here
but dropped_full does not, so the cross-strategy comparison can tell
"the channel was full" from "we threw a message away".
run_baseline swaps from a one-arm spawn into a match over the
strategy and selects the right producer. The consumer task, the
elapsed-time stamp, and the report assembly are unchanged, which is
exactly the payoff the step-1 Strategy enum was set up to deliver.
let producer_task: JoinHandle<ProducerOutcome> = match config.strategy {
Strategy::Send => tokio::spawn(producer(tx, config.total_messages, config.payload_size)),
Strategy::TrySend => tokio::spawn(producer_try_send(
tx, config.total_messages, config.payload_size,
)),
Strategy::TrySendDropOldest => tokio::spawn(producer_try_send_drop_oldest(
tx, config.total_messages, config.payload_size, config.drop_oldest_buffer,
)),
};
Six new tests pin the semantics. Two cover the drop-newest path: an
ample-capacity run that must deliver every message with zero drops,
and a small-capacity run with no concurrent consumer that must record
at least one dropped_full. One pins the Full-versus-Closed
distinction by dropping the receiver before the first send and
asserting closed_observations == 1 while dropped_full == 0. The
remaining three pin the drop-oldest path: an eviction test against a
slow consumer, an ample-capacity test where eviction must never
trigger, and a closed-receiver test that proves the fallback loop
exits instead of spinning.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn drop_oldest_fallback_evicts_earliest_messages() {
init_tracing();
let (tx, mut rx) = mpsc::channel::<Message>(2);
let producer_task = tokio::spawn(producer_try_send_drop_oldest(tx, 40, 0, 2));
let consumer_task = tokio::spawn(async move {
let mut seen: Vec<u64> = Vec::new();
while let Some(msg) = rx.recv().await {
seen.push(msg.seq);
tokio::time::sleep(Duration::from_millis(3)).await;
}
seen
});
let outcome = producer_task.await.expect("producer panicked");
let seen = consumer_task.await.expect("consumer panicked");
let stats = outcome.try_stats.as_ref().expect("try_stats must be populated");
assert_eq!(outcome.sent + stats.dropped_oldest, 40);
assert!(stats.dropped_oldest >= 1);
assert_eq!(*seen.last().expect("at least one delivery"), 39);
}
The last assertion is the load-bearing one. Because the producer ends
with a send().await drain pass, whatever message is sitting at the
back of the buffer when the input stream ends is guaranteed to land
on the consumer. Pinning seen.last() == 39 proves the drop-oldest
policy never sacrifices the freshest message, which is the entire
selling point of choosing this eviction policy over drop-newest.
Verification
cargo test
Finished `test` profile [unoptimized + debuginfo] target(s) in 0.23s
Running unittests src/lib.rs (target/debug/deps/tokio_mpsc_lab-faf64ef3387c85e5)
running 15 tests
test tests::default_config_uses_capacity_eight ... ok
test tests::producer_stops_when_consumer_drops_receiver ... ok
test tests::drop_oldest_propagates_closed_signal ... ok
test tests::capacity_one_forces_at_least_one_park ... ok
test tests::try_send_distinguishes_closed_from_full ... ok
test tests::try_send_delivers_all_when_buffer_is_ample ... ok
test tests::drop_oldest_with_fast_consumer_delivers_all ... ok
test tests::fast_consumer_keeps_send_latency_small ... ok
test tests::try_send_drops_messages_when_consumer_cannot_keep_up ... ok
test tests::fast_consumer_keeps_producer_on_fast_path ... ok
test tests::baseline_harness_delivers_every_message ... ok
test tests::capacity_one_still_drains_all_messages ... ok
test tests::drop_oldest_fallback_evicts_earliest_messages ... ok
test tests::slow_consumer_parks_producer_repeatedly ... ok
test tests::slow_consumer_inflates_send_latency ... ok
test result: ok. 15 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.19s
Doc-tests tokio_mpsc_lab
running 0 tests
test result: ok. 0 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.00s
All fifteen tests pass. The nine step-1-through-3 invariants still
hold, and the six new tests confirm that try_send reports Full
versus Closed correctly and that the drop-oldest fallback evicts the
expected end of the queue.
What we built
A TrySendStats sink that records full_observations,
closed_observations, dropped_full, dropped_oldest, and
max_buffer_depth. The split between observations and drops gives
the cross-strategy report a single place to read both "how much
pressure did we see?" and "how much did we throw away?" without
conflating the two.
A producer_try_send entry point that does one synchronous
tx.try_send(msg) per message, classifies the result into the right
sink via a small free function, and yields between iterations so a
single-threaded runtime cannot starve the consumer. The latency and
park sinks from earlier steps stay populated, which means a future
comparison can plot fast-path latency across Send, TrySend, and
the drop-oldest variant on one axis.
A producer_try_send_drop_oldest entry point that runs the
drop-oldest eviction policy entirely in producer-side state. The
small VecDeque ring buffer absorbs transient pressure, the
flush_buffer_try_send helper retries the front of the buffer on the
next iteration when the channel is full, and a final send().await
drain phase guarantees the freshest message reaches the consumer no
matter how aggressive the eviction was earlier.
A widened run_baseline that dispatches on Strategy and produces
the same HarnessReport shape for all three variants. The next step
can plug send_timeout into the same slot, attach its own bounded
wait stats, and read the cross-strategy comparison off three reports
that already share four out of five fields.
Repository
The state of the code after this step: aeedba4
Step 5: Bounding the Producer Wait with send_timeout — Telling Elapsed from Closed
Step 4 left us with two extremes on the same harness: Strategy::Send
parks until a permit appears, and Strategy::TrySend refuses to wait
at all. The middle of that spectrum — wait, but not forever — is the
shape most production producers actually want, and tokio exposes it
directly through mpsc::Sender::send_timeout.
This step ships Strategy::SendTimeout end-to-end. We grow the
HarnessConfig with a per-call send_timeout_deadline, add a
TimeoutStats sink that distinguishes SendTimeoutError::Timeout
("the deadline expired") from SendTimeoutError::Closed ("no receiver
will ever read this"), and wire a new producer_send_timeout into the
existing run_baseline switch. The latency and park sinks keep
working unchanged, which means the next step can drop all four
strategies onto the same scoreboard with no extra plumbing.
Setup
No new crates. tokio::sync::mpsc::error::SendTimeoutError lives in
the sync feature already turned on in step 1, and send_timeout
itself needs the time feature that step 1 enabled for Duration
and Instant. All edits land in src/lib.rs; the only import change
folds the new error type into the existing error:: bring-in:
use tokio::sync::mpsc::error::{SendTimeoutError, TrySendError};
The public surface grows by one Strategy variant (SendTimeout),
one config field (send_timeout_deadline: Duration, defaulting to
50 ms), one stats struct (TimeoutStats), and one optional field on
ProducerOutcome / HarnessReport (timeout_stats: Option<TimeoutStats>).
The Option follows the same pattern step 4 set for try_stats: a
strategy that doesn't run this code path has no honest data to report,
and None is a cleaner signal than a zero-filled struct that lies
about whether the producer ever called send_timeout.
Implementation
TimeoutStats is the new sink, and it earns its keep by keeping
observation counters separate from drop counters in the same
spirit as TrySendStats from step 4. elapsed_observations is "how
many times did we hit the deadline?", dropped_elapsed is "how many
messages did we throw away as a result?", and the two stay in
lockstep for the simple producer — but the split lets future
strategies (a retry-once policy, for example) report pressure
separately from loss.
#[derive(Debug, Clone, Default)]
pub struct TimeoutStats {
pub elapsed_observations: u64,
pub closed_observations: u64,
pub dropped_elapsed: u64,
pub deadline: Duration,
pub max_wait: Duration,
}
deadline is echoed back into the stats struct so a downstream report
can render the configured budget next to the observed waits without
re-plumbing the HarnessConfig. max_wait records the longest
successful wait — the practical tuning signal for this strategy.
When max_wait creeps up toward deadline, the budget is on the
tight side and more elapsed events are about to appear; when it
sits far below, the deadline is loose enough that the strategy is
quietly degenerating into plain send().await.
The producer entry point is short because all the bookkeeping lives
in two helpers. The outer loop builds a Message, hands it to a
park-counting wrapper around send_timeout, and then routes the
outcome through a classifier that knows how to update every sink in
one place.
#[instrument(
skip(tx),
fields(
total = total_messages,
payload_size,
strategy = "send_timeout",
deadline_ms = deadline.as_millis() as u64,
),
)]
pub async fn producer_send_timeout(
tx: mpsc::Sender<Message>,
total_messages: u64,
payload_size: usize,
deadline: Duration,
) -> ProducerOutcome {
let mut sent = 0u64;
let mut latencies = LatencyStats::default();
let mut park_stats = ParkStats::default();
let mut timeout_stats = TimeoutStats { deadline, ..Default::default() };
for seq in 0..total_messages {
let msg = Message::new(seq, vec![0u8; payload_size]);
let send_span = info_span!("send_timeout", seq, payload = payload_size);
let (outcome, waited, parks) =
send_timeout_with_park_count(&tx, msg, deadline, send_span.clone()).await;
send_span.in_scope(|| trace!(?waited, parks, "send_timeout completed"));
let metrics = SendMetrics { waited, parks };
if !record_send_timeout_outcome(
outcome, metrics, seq,
&mut sent, &mut latencies, &mut park_stats, &mut timeout_stats,
) {
break;
}
}
ProducerOutcome {
sent, latencies, park_stats,
try_stats: None,
timeout_stats: Some(timeout_stats),
}
}
The park-counting wrapper is the same trick step 3 used for
send().await: pin the future, drive it through poll_fn, and
increment a counter on every Poll::Pending return. The wrapper
exists because a send_timeout future can park for two distinct
reasons — waiting for a channel permit, or waiting for the timer
waker — and the producer should treat both as parks for the purpose
of the cross-strategy scoreboard.
async fn send_timeout_with_park_count(
tx: &mpsc::Sender<Message>,
msg: Message,
deadline: Duration,
send_span: tracing::Span,
) -> (Result<(), SendTimeoutError<Message>>, Duration, u64) {
let send_fut = tx.send_timeout(msg, deadline).instrument(send_span);
tokio::pin!(send_fut);
let started = Instant::now();
let mut parks: u64 = 0;
let result = poll_fn(|cx| {
let polled = send_fut.as_mut().poll(cx);
if polled.is_pending() {
parks += 1;
}
polled
})
.await;
(result, started.elapsed(), parks)
}
Returning the elapsed Duration alongside the result is what keeps
max_wait honest. The classifier records the same waited value as
both a latency sample on success and as the max_wait candidate, so
the scoreboard's latency series and the timeout sink's tail-wait
signal stay derived from the same measurement rather than two
slightly-off clocks.
The classifier itself is a small free function so the loop body stays
within the codebase rule against nested control flow. Each arm of the
match handles a different kind of event, and returning a bool
keeps the loop-termination decision at the call site:
fn record_send_timeout_outcome(
outcome: Result<(), SendTimeoutError<Message>>,
metrics: SendMetrics,
seq: u64,
sent: &mut u64,
latencies: &mut LatencyStats,
park_stats: &mut ParkStats,
timeout_stats: &mut TimeoutStats,
) -> bool {
let SendMetrics { waited, parks } = metrics;
match outcome {
Ok(()) => {
*sent += 1;
latencies.record(waited);
park_stats.record(parks);
if waited > timeout_stats.max_wait {
timeout_stats.max_wait = waited;
}
true
}
Err(SendTimeoutError::Timeout(_)) => {
timeout_stats.elapsed_observations += 1;
timeout_stats.dropped_elapsed += 1;
trace!(seq, ?waited, "send_timeout deadline exceeded; dropping");
true
}
Err(SendTimeoutError::Closed(_)) => {
timeout_stats.closed_observations += 1;
info!(at = seq, "send_timeout saw Closed; stopping producer");
false
}
}
}
The Timeout arm returns true because a single missed deadline says
nothing about the next message — the consumer may still drain a slot
before the next call. The Closed arm returns false because once
the receiver is gone every subsequent send_timeout is guaranteed to
return Closed immediately, and a producer that kept polling would
just burn CPU re-deriving the same answer. This is exactly the reason
to prefer send_timeout over a hand-rolled tokio::time::timeout
around send: the latter collapses both failure modes into a single
Elapsed error and forces the caller to inspect the channel state to
tell them apart.
Wiring the new strategy into run_baseline is a one-arm extension to
the step-4 match. The consumer task, the elapsed-time stamp, and the
report assembly are all unchanged — the only new line in the
HarnessReport is the timeout_stats: outcome.timeout_stats carry:
let producer_task: JoinHandle<ProducerOutcome> = match config.strategy {
Strategy::Send => tokio::spawn(producer(tx, config.total_messages, config.payload_size)),
Strategy::TrySend => tokio::spawn(producer_try_send(
tx, config.total_messages, config.payload_size,
)),
Strategy::TrySendDropOldest => tokio::spawn(producer_try_send_drop_oldest(
tx, config.total_messages, config.payload_size, config.drop_oldest_buffer,
)),
Strategy::SendTimeout => tokio::spawn(producer_send_timeout(
tx, config.total_messages, config.payload_size, config.send_timeout_deadline,
)),
};
Five new tests pin the semantics. send_timeout_delivers_all_when_buffer_is_ample
asserts that an ample-capacity run records zero timeouts and echoes
the configured deadline through TimeoutStats. send_timeout_drops_messages_when_consumer_misses_deadline
sets a 5 ms deadline against a 20 ms-per-recv consumer and asserts
elapsed_observations >= 1 while closed_observations == 0, plus
the bookkeeping invariant sent + dropped_elapsed == total.
#[tokio::test]
async fn send_timeout_distinguishes_closed_from_elapsed() {
init_tracing();
let (tx, rx) = mpsc::channel::<Message>(2);
drop(rx);
let outcome = producer_send_timeout(tx, 10, 0, Duration::from_millis(50)).await;
let stats = outcome.timeout_stats.as_ref().expect("populated");
assert_eq!(outcome.sent, 0);
assert_eq!(stats.elapsed_observations, 0);
assert_eq!(stats.closed_observations, 1);
assert_eq!(outcome.latencies.count(), 0);
}
That third test is the load-bearing one for the article's central
claim. It drops the receiver before the first send, runs the
producer with a generous 50 ms deadline, and proves that the very
first call returns Closed rather than waiting out the deadline and
reporting Timeout. A hand-rolled tokio::time::timeout wrapper
would fail that assertion — it would see Ok(Err(SendError)) only
after the inner future resolved, and the cheapest path to the same
behaviour requires inspecting tx.is_closed() separately.
send_timeout_records_wait_time_for_successful_parks exercises the
positive-wait path against a capacity-1 channel with a slow consumer
and asserts that max_wait stays bounded by the configured deadline
while parked_sends() >= 1. send_timeout_deadline_tuning_changes_drop_rate
runs the same workload at 1 ms and 200 ms deadlines and asserts the
monotonicity property: a longer deadline must never drop more
messages than a shorter one.
Verification
cargo test
Finished `test` profile [unoptimized + debuginfo] target(s) in 1.33s
Running unittests src/lib.rs (target/debug/deps/tokio_mpsc_lab-faf64ef3387c85e5)
running 20 tests
test tests::default_config_uses_capacity_eight ... ok
test tests::drop_oldest_propagates_closed_signal ... ok
test tests::producer_stops_when_consumer_drops_receiver ... ok
test tests::send_timeout_distinguishes_closed_from_elapsed ... ok
test tests::send_timeout_delivers_all_when_buffer_is_ample ... ok
test tests::drop_oldest_with_fast_consumer_delivers_all ... ok
test tests::capacity_one_still_drains_all_messages ... ok
test tests::capacity_one_forces_at_least_one_park ... ok
test tests::baseline_harness_delivers_every_message ... ok
test tests::fast_consumer_keeps_producer_on_fast_path ... ok
test tests::fast_consumer_keeps_send_latency_small ... ok
test tests::try_send_distinguishes_closed_from_full ... ok
test tests::try_send_drops_messages_when_consumer_cannot_keep_up ... ok
test tests::try_send_delivers_all_when_buffer_is_ample ... ok
test tests::drop_oldest_fallback_evicts_earliest_messages ... ok
test tests::send_timeout_records_wait_time_for_successful_parks ... ok
test tests::slow_consumer_parks_producer_repeatedly ... ok
test tests::slow_consumer_inflates_send_latency ... ok
test tests::send_timeout_drops_messages_when_consumer_misses_deadline ... ok
test tests::send_timeout_deadline_tuning_changes_drop_rate ... ok
test result: ok. 20 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.85s
Doc-tests tokio_mpsc_lab
running 0 tests
test result: ok. 0 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.00s
All twenty tests pass. The fifteen step-1-through-4 invariants still
hold, and the five new tests confirm that send_timeout reports
Timeout versus Closed correctly, bounds successful waits below
the configured deadline, and responds monotonically to deadline
tuning.
What we built
A TimeoutStats sink that records elapsed_observations,
closed_observations, dropped_elapsed, deadline, and max_wait.
The split between observation and drop counters mirrors
TrySendStats from step 4, so the cross-strategy report has a
uniform shape: every non-blocking strategy reports pressure and loss
on independent axes, and the configured budget travels with the
results.
A producer_send_timeout entry point that wraps mpsc::Sender::send_timeout
in the same park-counting poll_fn harness step 3 built for
send().await. The wrapper hands back the elapsed wait and the
Pending count alongside the result, so the same measurement feeds the
latency sink, the park sink, and the new max_wait tuning signal —
no clock skew between series.
A classifier that turns the three SendTimeoutError outcomes into
the right sink updates: Ok(()) records a latency sample and a park
count, Timeout increments both elapsed_observations and
dropped_elapsed while keeping the loop running, and Closed
records the observation and exits. This is the structural reason to
prefer send_timeout over tokio::time::timeout(send) — the caller
learns why the wait ended without re-deriving channel state from a
nested Result.
A widened run_baseline that dispatches on a four-variant Strategy
enum and produces the same HarnessReport shape for every variant.
The next step can plot Send, TrySend, TrySendDropOldest, and
SendTimeout on a single comparison chart without any further
plumbing — every report already carries the same LatencyStats,
ParkStats, and the right Option-tagged strategy-specific sink.
Repository
The state of the code after this step: a53a68e
Step 6: Building the Load-Generator Binary — One Workload, Three Strategies, CSV Output
Step 5 left us with four Strategy variants and a HarnessReport
that already carries every counter we need: latency samples, park
counts, an Option<TrySendStats>, and an Option<TimeoutStats>. What
the project lacked was a single entry point that exercises all three
headline strategies through the same workload and writes the numbers
somewhere a downstream chart can consume them.
This step ships that entry point. We add a cargo binary at
src/bin/load_generator.rs, a loadgen module that owns the CSV
schema, a LatencyStats::percentile helper so the row can report p99
without external crates, and nine new tests pinning the format. The
narrative payoff is direct: every comparison the article makes from
here on cites a CSV row this binary produced.
Setup
No new dependencies. The binary uses only std::env, std::str::FromStr,
and std::time::Duration for argument parsing, plus the existing
tokio_mpsc_lab library API. We are deliberately not pulling in clap
or argh for four integer knobs — the cost in build time and binary
size dwarfs the convenience win.
Three files change. src/lib.rs grows a pub mod loadgen; declaration
and a LatencyStats::percentile method. src/loadgen.rs is new and
holds CSV_COLUMNS, csv_header, csv_row, strategy_label, and
the LOAD_GENERATOR_STRATEGIES constant. src/bin/load_generator.rs
is new and is the binary entry point — cargo run --bin load_generator
becomes the canonical way to produce a CSV.
The Cargo.toml from step 1 already enables the rt-multi-thread,
macros, sync, and time features of tokio, so #[tokio::main],
mpsc::channel, and the existing send_timeout deadline all keep
working unchanged. The package was created with cargo new --lib in
step 1, and adding a binary under src/bin/ is the conventional way
to ship one extra entry point without restructuring into a workspace.
Implementation
The first chunk is the percentile helper. The CSV row reports p99
alongside mean and max, and a custom estimator keeps the library
free of an extra dependency:
pub fn percentile(&self, pct: f64) -> Option<Duration> {
if self.samples.is_empty() {
return None;
}
let mut sorted = self.samples.clone();
sorted.sort();
let clamped = pct.clamp(0.0, 1.0);
let last = sorted.len() - 1;
let idx = (clamped * last as f64).round() as usize;
Some(sorted[idx.min(last)])
}
Sorting a clone per call is intentional. The load generator asks for a
percentile exactly once per run, after all sends have finished, so a
running quantile sketch would carry pointless complexity for the scale
the article cares about. The clamp + min(last) belt-and-braces
keep the index inside bounds even when callers pass 1.0 and the
rounded product equals len, which is the kind of off-by-one that
otherwise surfaces only on quiet workloads with exactly one sample.
The CSV schema is the second chunk. We declare the columns once as a
&[&str] and derive both the header line and the row format from it,
so adding a column means editing one place rather than three:
pub const CSV_COLUMNS: &[&str] = &[
"strategy", "capacity", "total_messages", "payload_size",
"deadline_ms", "produced", "consumed", "elapsed_us",
"mean_latency_us", "p99_latency_us", "max_latency_us",
"fast_path_sends", "parked_sends", "total_parks",
"full_observations", "closed_observations",
"dropped_full", "dropped_oldest", "dropped_elapsed",
"max_wait_us",
];
Twenty columns. The first five are config echoes — every comparison
chart needs to know which knob produced the row, and embedding the
knob in the row is cheaper than carrying a sidecar manifest. The next
six are latency and throughput. The remaining nine are
strategy-specific counters, with 0 for any strategy that does not
exercise that code path. try_send_drop_oldest is intentionally
excluded from the headline scoreboard because the article frames the
comparison as send versus try_send versus send_timeout; the
fourth variant remains testable but lives outside the binary's loop.
The row builder is a single function that knows how to project a
HarnessReport onto the schema. Strategy-specific stats live behind
Options on the report, so each lookup uses map_or(0, ...) to fall
back to a zero for strategies that did not populate that sink:
let try_view = report.try_stats.as_ref();
let full_observations = try_view.map_or(0, |t| t.full_observations);
let closed_from_try = try_view.map_or(0, |t| t.closed_observations);
let dropped_full = try_view.map_or(0, |t| t.dropped_full);
let dropped_oldest = try_view.map_or(0, |t| t.dropped_oldest);
let timeout_view = report.timeout_stats.as_ref();
let closed_from_timeout = timeout_view.map_or(0, |t| t.closed_observations);
let dropped_elapsed = timeout_view.map_or(0, |t| t.dropped_elapsed);
let max_wait_us = timeout_view.map_or(0u128, |t| t.max_wait.as_micros());
let closed_observations = closed_from_try + closed_from_timeout;
Merging the two closed_observations sources into one column is the
non-obvious choice here. Both TrySendError::Closed and
SendTimeoutError::Closed describe the same underlying event — the
receiver is gone — so unifying them keeps the per-row interpretation
honest: any non-zero value in that column means the consumer died at
least once during the run, regardless of which strategy detected it
first.
The binary itself is fourteen lines of actual logic. Workload knobs
come from environment variables through a parse_env<T: FromStr>
helper, defaults are inlined, and the loop prints exactly one header
line and one row per strategy:
#[tokio::main]
async fn main() {
init_tracing();
let capacity: usize = parse_env("LOADGEN_CAPACITY", 8);
let total_messages: u64 = parse_env("LOADGEN_TOTAL", 256);
let payload_size: usize = parse_env("LOADGEN_PAYLOAD", 64);
let deadline_ms: u64 = parse_env("LOADGEN_DEADLINE_MS", 10);
println!("{}", csv_header());
for strategy in LOAD_GENERATOR_STRATEGIES {
let config = HarnessConfig {
capacity,
total_messages,
payload_size,
strategy,
drop_oldest_buffer: capacity,
send_timeout_deadline: Duration::from_millis(deadline_ms),
};
let report = run_baseline(config.clone()).await;
println!("{}", csv_row(strategy_label(strategy), &config, &report));
}
}
Iterating over the LOAD_GENERATOR_STRATEGIES constant — rather than
hand-listing the three variants in the binary — lets a single test in
the library verify that the binary's loop matches the article's
framing. Misordering the variants would silently break the chart
captions, and the constant-plus-test pair makes that desync impossible
without a corresponding test failure.
Nine new tests pin the format. Three are pure unit tests:
csv_header_lists_all_columns_in_order asserts column count, the
sentinel first column (strategy) and the sentinel last column
(max_wait_us); strategy_labels_are_stable locks the four label
strings against accidental renames; and load_generator_strategies_match_article_framing
asserts the constant equals [Send, TrySend, SendTimeout] in that
exact order. The remaining six are async integration tests that run
the real harness and assert per-strategy invariants — for instance
that produced + dropped_full == total_messages for try_send and
produced + dropped_elapsed == total_messages for send_timeout.
#[tokio::test]
async fn load_generator_workload_runs_all_three_strategies() {
init_tracing();
let base = HarnessConfig {
capacity: 4,
total_messages: 32,
payload_size: 0,
strategy: Strategy::Send,
drop_oldest_buffer: 4,
send_timeout_deadline: Duration::from_millis(5),
};
let mut rows = Vec::new();
for strategy in LOAD_GENERATOR_STRATEGIES {
let mut config = base.clone();
config.strategy = strategy;
let report = run_baseline(config.clone()).await;
rows.push(csv_row(strategy_label(strategy), &config, &report));
}
assert!(rows[0].starts_with("send,"));
assert!(rows[1].starts_with("try_send,"));
assert!(rows[2].starts_with("send_timeout,"));
}
Two more tests cover the percentile helper: one asserts the boundary
behaviour (percentile(0.0) is the min, percentile(1.0) is the max,
percentile(0.5) sits between the second and fourth elements of a
five-sample set), and one asserts that an empty sample set yields
None rather than panicking. The empty-set case matters because the
send strategy with total_messages = 0 would otherwise produce a
divide-by-zero in any naive implementation.
Verification
cargo test
Finished `test` profile [unoptimized + debuginfo] target(s) in 0.28s
Running unittests src/lib.rs (target/debug/deps/tokio_mpsc_lab-faf64ef3387c85e5)
running 29 tests
test loadgen::tests::load_generator_strategies_match_article_framing ... ok
test loadgen::tests::csv_header_lists_all_columns_in_order ... ok
test loadgen::tests::strategy_labels_are_stable ... ok
test tests::default_config_uses_capacity_eight ... ok
test tests::drop_oldest_propagates_closed_signal ... ok
test loadgen::tests::csv_row_for_try_send_strategy_emits_full_observation_column ... ok
test tests::capacity_one_forces_at_least_one_park ... ok
test loadgen::tests::csv_row_for_send_strategy_zeros_non_blocking_counters ... ok
test tests::percentile_on_empty_samples_is_none ... ok
test tests::percentile_returns_a_sample_in_sorted_position ... ok
test loadgen::tests::csv_row_for_send_timeout_includes_deadline_ms ... ok
test tests::baseline_harness_delivers_every_message ... ok
test tests::producer_stops_when_consumer_drops_receiver ... ok
test tests::capacity_one_still_drains_all_messages ... ok
test tests::send_timeout_distinguishes_closed_from_elapsed ... ok
test tests::drop_oldest_with_fast_consumer_delivers_all ... ok
test tests::send_timeout_delivers_all_when_buffer_is_ample ... ok
test tests::fast_consumer_keeps_send_latency_small ... ok
test tests::fast_consumer_keeps_producer_on_fast_path ... ok
test tests::try_send_distinguishes_closed_from_full ... ok
test tests::try_send_drops_messages_when_consumer_cannot_keep_up ... ok
test tests::try_send_delivers_all_when_buffer_is_ample ... ok
test loadgen::tests::load_generator_workload_runs_all_three_strategies ... ok
test tests::drop_oldest_fallback_evicts_earliest_messages ... ok
test tests::send_timeout_records_wait_time_for_successful_parks ... ok
test tests::slow_consumer_inflates_send_latency ... ok
test tests::slow_consumer_parks_producer_repeatedly ... ok
test tests::send_timeout_drops_messages_when_consumer_misses_deadline ... ok
test tests::send_timeout_deadline_tuning_changes_drop_rate ... ok
test result: ok. 29 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.82s
All twenty-nine tests pass — the twenty step-5 invariants, two new
percentile tests, and seven new loadgen tests. Running the binary
itself produces a header line plus three rows on stdout (tracing
spans go to stderr, so a pipe to a file captures only the CSV):
cargo run --quiet --bin load_generator 2>/dev/null
strategy,capacity,total_messages,payload_size,deadline_ms,produced,consumed,elapsed_us,mean_latency_us,p99_latency_us,max_latency_us,fast_path_sends,parked_sends,total_parks,full_observations,closed_observations,dropped_full,dropped_oldest,dropped_elapsed,max_wait_us
send,8,256,64,10,256,256,10080,8,56,116,225,31,31,0,0,0,0,0,0
try_send,8,256,64,10,252,252,4471,1,2,25,252,0,0,4,0,4,0,0,0
send_timeout,8,256,64,10,256,256,7295,7,43,58,225,31,31,0,0,0,0,0,58
The shape of the output already tells the story the rest of the
article will tell at scale. send delivers everything but takes the
longest wall-clock (10080 µs) and shows 31 parks. try_send is the
fastest end-to-end (4471 µs) and the lowest-latency per send, but it
drops four messages outright. send_timeout delivers all 256
messages, parks 31 times like send, but caps its longest successful
wait at max_wait_us = 58 — well under the 10 ms deadline — and
finishes in 7295 µs, in between the other two.
What we built
A loadgen module that owns the comparison schema. Twenty CSV
columns named once, a csv_header() and csv_row() pair derived from
that list, and a strategy_label() that locks the four user-visible
strings against accidental renames. Anything downstream that reads
the CSV — a notebook, a chart, a polars script — depends on this
single source of truth.
A LatencyStats::percentile helper that handles the 0.0, 1.0, and
empty-set edge cases explicitly. The article cites p99 in every
strategy comparison from here on, so the helper has to be honest about
its bounds even on tiny sample sets — and the two new unit tests pin
exactly that behaviour.
A load_generator cargo binary that drives Strategy::Send,
Strategy::TrySend, and Strategy::SendTimeout through one identical
HarnessConfig. Knobs come from LOADGEN_CAPACITY, LOADGEN_TOTAL,
LOADGEN_PAYLOAD, and LOADGEN_DEADLINE_MS environment variables,
so a benchmark run is reproducible as a single shell line and the
binary stays dependency-free.
Nine new tests covering the schema and the loop: column order, label
stability, the workload constant matching the article's framing, and
per-strategy invariants on produced + dropped == total for both
non-blocking strategies. The step-7 chart will read this CSV and walk
readers through the trade-offs — but the numbers it cites are now
generated by code the test suite locks down, not by ad hoc
measurement.
Repository
The state of the code after this step: abd913f
Step 7: Putting All Three Strategies on One Scoreboard — Throughput, p99 Latency, Drop Rate, and Producer-Stall Time Side by Side
Step 6 gave us a load_generator binary that emits a twenty-column CSV
with every counter the library tracks. That output is honest but
overwhelming: a reader scanning it has to remember which columns are
strategy-specific zeros, which represent the same underlying event under
different names, and which trade off against each other. The article
needs a narrower lens — four numbers, one row per strategy, ranked so
the trade-off is immediate.
This step ships that lens. We add a comparison module that collapses
the headline counters into four normalised metrics — throughput, p99
latency, drop rate, producer-stall time — and a compare binary that
renders them as a Markdown table. Eight new tests pin the normalisation
rules so the scoreboard never silently lies about a strategy that does
not populate a particular counter.
Setup
No new dependencies. The module sits next to loadgen.rs and reuses
run_baseline, HarnessConfig, HarnessReport, Strategy,
LatencyStats::percentile, and LOAD_GENERATOR_STRATEGIES from earlier
steps. The ASCII table renderer is a thirty-line String-builder — a
formatting crate would dwarf the actual logic.
Three files change. src/lib.rs grows a single pub mod comparison;
declaration so the module is visible on the public API. src/comparison.rs
is new and holds ComparisonRow, Comparison, the COMPARISON_COLUMNS
constant, the table renderer, and eight #[tokio::test] cases.
src/bin/compare.rs is new and is the user-visible entry point —
cargo run --bin compare is the canonical way to print the table.
The Cargo.toml from step 1 already provides everything we need: the
tokio multi-threaded runtime drives from_workload, tracing records
the per-strategy harness runs at INFO, and std::time::Duration
arithmetic carries the stall accounting. The compare binary reads the
same LOADGEN_CAPACITY, LOADGEN_TOTAL, LOADGEN_PAYLOAD, and
LOADGEN_DEADLINE_MS environment variables as load_generator, so a
reader does not have to learn a second control surface.
Implementation
The first chunk is the row struct. ComparisonRow is one row of the
scoreboard, but instead of mirroring the twenty raw CSV columns it
exposes only the normalised four plus the labels and totals needed to
read them in context:
#[derive(Debug, Clone)]
pub struct ComparisonRow {
pub strategy: Strategy,
pub label: &'static str,
pub produced: u64,
pub consumed: u64,
pub dropped: u64,
pub total_messages: u64,
pub elapsed: Duration,
pub throughput_msgs_per_sec: f64,
pub drop_rate: f64,
pub mean_latency: Duration,
pub p99_latency: Duration,
pub producer_stall: Duration,
}
The struct deliberately carries both produced and dropped even
though produced + dropped == total_messages for both non-blocking
strategies. Keeping the redundant accounting visible means a reader who
distrusts the drop_rate column can spot-check it against the integers
in the same row — the scoreboard never asks them to take a derived
number on faith.
The second chunk is the drop unifier. Three different counters mean
"the source generated a message that never reached the consumer":
TrySendError::Full, the drop-oldest eviction path, and
SendTimeoutError::Elapsed. The scoreboard treats them as one quantity
so that strategies can be compared on the cost the application pays:
fn dropped_messages(report: &HarnessReport) -> u64 {
let try_drops = report
.try_stats
.as_ref()
.map(|t| t.dropped_full + t.dropped_oldest)
.unwrap_or(0);
let timeout_drops = report
.timeout_stats
.as_ref()
.map(|t| t.dropped_elapsed)
.unwrap_or(0);
try_drops + timeout_drops
}
Strategy::Send populates neither try_stats nor timeout_stats, so
the unwrap_or(0) branches handle it without a strategy match — the
scoreboard does not need to know which strategy wrote the report, only
how to read the optional sinks. A future fifth strategy that hangs its
own stats struct off HarnessReport can plug in here by extending this
helper, not by editing the renderer.
The third chunk is the stall accounting, which is the only metric that
required a design decision rather than a pure projection. Every
successful send already records its wait time in send_latencies, so
their sum is the measured time the producer was blocked. But
send_timeout Elapsed observations also burn approximately one full
deadline before the producer gives up and discards the message — that
time never appears as a send_latencies sample because the send did
not succeed:
fn producer_stall_time(report: &HarnessReport) -> Duration {
let measured: Duration = report.send_latencies.samples().iter().copied().sum();
let timeout_overhead = report
.timeout_stats
.as_ref()
.map(|t| {
let count: u32 = t.elapsed_observations.try_into().unwrap_or(u32::MAX);
t.deadline.saturating_mul(count)
})
.unwrap_or_default();
measured + timeout_overhead
}
The saturating_mul matters. A pathological run with a huge deadline
and many elapsed observations could overflow Duration's nanos field,
and a panic in the scoreboard renderer is the worst kind of bug — it
takes the whole comparison down because of an arithmetic edge case in
one row. Saturating to the max representable Duration keeps the
output readable: a row whose stall time pegs at the cap is obviously
the loser regardless of the exact number.
The fourth chunk is Comparison::from_workload. It clones the base
config, swaps in each Strategy from the existing
LOAD_GENERATOR_STRATEGIES constant, runs the harness, and packs the
report into a row. The strategy order in the rendered table is
inherited from that constant — [Send, TrySend, SendTimeout] — and the
existing load_generator_strategies_match_article_framing test in
loadgen already locks that ordering down:
impl Comparison {
pub async fn from_workload(base: HarnessConfig) -> Self {
let mut rows = Vec::with_capacity(LOAD_GENERATOR_STRATEGIES.len());
for strategy in LOAD_GENERATOR_STRATEGIES {
let mut config = base.clone();
config.strategy = strategy;
let report = run_baseline(config.clone()).await;
rows.push(ComparisonRow::from_report(&config, &report));
}
Self { rows }
}
}
The final chunk is the renderer. render_table builds an ASCII table
keyed by the COMPARISON_COLUMNS constant, computing per-column widths
from header + body so that an arbitrarily large producer_stall_us
never breaks the alignment. The output is GitHub-Markdown-compatible —
the article can paste it inline elsewhere and it stays a real table:
pub const COMPARISON_COLUMNS: &[&str] = &[
"strategy",
"produced",
"consumed",
"dropped",
"drop_rate",
"throughput_msgs_s",
"mean_latency_us",
"p99_latency_us",
"producer_stall_us",
];
Eight new tests pin the contract. comparison_runs_all_three_strategies_in_article_order
asserts the row count and that rows[0]..rows[2] carry the labels
send, try_send, send_timeout in that exact order.
send_strategy_drops_zero_and_throughput_is_positive enforces the
defining invariant of blocking send — no drops, ever — and rules out a
NaN throughput from a zero-duration run. try_send_drop_rate_matches_unified_drop_counter
checks the unifier against the raw TrySendStats sum to catch any
future drift between the two views, and
try_send_drop_classification_separates_full_from_closed then
double-locks the same sink against any later refactor that quietly
collapses Full and Closed into one bucket.
Two tests cover the stall accounting:
try_send_producer_stall_is_tiny_compared_to_total_run asserts that
non-blocking sends never cumulatively exceed 50 ms of stall (a sanity
ceiling — a real non-blocking strategy must stay orders of magnitude
under that), and send_timeout_stall_includes_per_timeout_deadline_overhead
constructs a workload guaranteed to trigger Elapsed observations and
verifies that the reported stall includes deadline × elapsed_count on
top of the measured per-send waits. The remaining two pin the
edge-case contracts: drop_rate_zero_when_total_messages_zero rules
out NaN from a zero-message run, and render_table_is_well_formed_and_contains_every_header
verifies that every line of the rendered table has the same pipe count
so the alignment can never silently drift.
Verification
cargo test
Finished `test` profile [unoptimized + debuginfo] target(s) in 22.67s
Running unittests src/lib.rs (target/debug/deps/tokio_mpsc_lab-faf64ef3387c85e5)
running 37 tests
test comparison::tests::drop_rate_zero_when_total_messages_zero ... ok
test loadgen::tests::csv_header_lists_all_columns_in_order ... ok
test loadgen::tests::load_generator_strategies_match_article_framing ... ok
test comparison::tests::try_send_drop_classification_separates_full_from_closed ... ok
test loadgen::tests::csv_row_for_try_send_strategy_emits_full_observation_column ... ok
test comparison::tests::try_send_drop_rate_matches_unified_drop_counter ... ok
test comparison::tests::try_send_producer_stall_is_tiny_compared_to_total_run ... ok
test loadgen::tests::csv_row_for_send_strategy_zeros_non_blocking_counters ... ok
test comparison::tests::send_strategy_drops_zero_and_throughput_is_positive ... ok
test loadgen::tests::csv_row_for_send_timeout_includes_deadline_ms ... ok
test comparison::tests::render_table_is_well_formed_and_contains_every_header ... ok
test loadgen::tests::strategy_labels_are_stable ... ok
test tests::baseline_harness_delivers_every_message ... ok
test tests::default_config_uses_capacity_eight ... ok
test tests::percentile_on_empty_samples_is_none ... ok
test tests::drop_oldest_propagates_closed_signal ... ok
test tests::percentile_returns_a_sample_in_sorted_position ... ok
test tests::producer_stops_when_consumer_drops_receiver ... ok
test comparison::tests::comparison_runs_all_three_strategies_in_article_order ... ok
test tests::capacity_one_still_drains_all_messages ... ok
test tests::capacity_one_forces_at_least_one_park ... ok
test tests::send_timeout_distinguishes_closed_from_elapsed ... ok
test tests::drop_oldest_with_fast_consumer_delivers_all ... ok
test tests::fast_consumer_keeps_send_latency_small ... ok
test tests::send_timeout_delivers_all_when_buffer_is_ample ... ok
test tests::fast_consumer_keeps_producer_on_fast_path ... ok
test loadgen::tests::load_generator_workload_runs_all_three_strategies ... ok
test tests::try_send_distinguishes_closed_from_full ... ok
test tests::try_send_drops_messages_when_consumer_cannot_keep_up ... ok
test tests::try_send_delivers_all_when_buffer_is_ample ... ok
test tests::drop_oldest_fallback_evicts_earliest_messages ... ok
test tests::send_timeout_records_wait_time_for_successful_parks ... ok
test tests::slow_consumer_parks_producer_repeatedly ... ok
test tests::slow_consumer_inflates_send_latency ... ok
test comparison::tests::send_timeout_stall_includes_per_timeout_deadline_overhead ... ok
test tests::send_timeout_drops_messages_when_consumer_misses_deadline ... ok
test tests::send_timeout_deadline_tuning_changes_drop_rate ... ok
test result: ok. 37 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.64s
All thirty-seven tests pass — the twenty-nine carried over from step 6
plus the eight new comparison::tests. Running the compare binary
itself produces a one-table summary on stdout (tracing spans go to
stderr so a pipe to a file captures only the scoreboard):
cargo run --quiet --bin compare 2>/dev/null
| strategy | produced | consumed | dropped | drop_rate | throughput_msgs_s | mean_latency_us | p99_latency_us | producer_stall_us |
|--------------|----------|----------|---------|-----------|-------------------|-----------------|----------------|-------------------|
| send | 64 | 64 | 0 | 0.000 | 21571.5 | 8 | 35 | 534 |
| try_send | 59 | 59 | 5 | 0.078 | 42423.2 | 1 | 3 | 93 |
| send_timeout | 64 | 64 | 0 | 0.000 | 25629.5 | 11 | 54 | 746 |
The shape of the table is the article's thesis distilled to three
rows. send delivers every message and pays for it: 534 µs of
cumulative producer stall and a 35 µs p99 wait. try_send is the
throughput champion at 42423 msgs/s with a 3 µs p99 — but at the cost
of a 7.8% drop rate, five messages the application emitted and the
consumer never saw. send_timeout sits in the middle: zero drops like
send, marginally faster throughput than send, a slightly higher
p99 latency, and the highest stall total because Elapsed accounting
pads it for the time spent waiting for permits that never came.
What we built
A comparison module that owns the four-metric scoreboard. The
ComparisonRow struct exposes throughput, p99 latency, drop rate, and
producer-stall time directly while keeping the raw produced /
dropped integers visible, so a sceptical reader can spot-check any
derived number against the underlying counts in the same row.
A drop-rate unifier that collapses three different mechanisms — Full, drop-oldest, Elapsed — into one "messages the application paid for and never saw" quantity. That collapse is what lets the table compare strategies on equal footing: the consumer does not care why a message was lost, only whether it was.
A producer-stall accounting that adds per-Elapsed deadline overhead on
top of the measured per-send wait times. Without it, send_timeout
would look free in any workload that sheds load — the scoreboard would
silently undercount the deadline burned on every Elapsed observation
and the table would lie about the strategy's true cost.
A compare cargo binary plus eight new tests pinning the contract:
row order, drop unification (with a paired classification check that
keeps Full and Closed distinct), stall accounting, table alignment,
and the two NaN-edge cases. The article's headline framing —
send maximises delivery, try_send maximises throughput, send_timeout
caps the worst case — is now backed by a single command whose output
the test suite locks against drift.
Repository
The state of the code after this step: 254a7ec
Step 8: Packaging the Production Playbook — Decision Rubric, Hybrid Try-Send, and Watch-Driven Shutdown
Step 7 produced an honest scoreboard. It settles whiteboard arguments about throughput, drop rate, p99 latency, and producer-stall time — but it remains a measurement artefact. A team adopting the crate needs three different deliverables: a default strategy they can reach for at design time, a sender shape that survives bursts without blocking, and a shutdown path that does not leak messages or panic on a closed receiver.
This final step packages those deliverables into a new production
module. recommend_strategy is a three-rule waterfall over a boolean
CallSiteProfile. HybridSender pairs try_send with a bounded
producer-side fallback queue. producer_with_shutdown races
Sender::reserve against a watch::Receiver<bool> inside a biased
tokio::select!. Fifteen new tests pin each contract so adopters can
copy the patterns without first having to re-derive them.
Setup
No new dependencies on the public surface — tokio::sync::watch is
already behind the sync feature flag enabled in step 1. Internally
the module reaches for std::collections::VecDeque for the fallback
queue and tracing::instrument for the shutdown task span, both
already present in the tree. The hybrid sender and the shutdown
producer reuse Message, LatencyStats, and Strategy from the
library root.
Two files change. src/lib.rs grows a single pub mod production;
declaration so the patterns sit on the public API alongside
comparison and loadgen. src/production.rs is new and holds
CallSiteProfile, recommend_strategy, HybridSender,
HybridPushOutcome, producer_with_shutdown, and ShutdownOutcome
— plus the fifteen #[test] and #[tokio::test] cases that lock
each piece down.
The module deliberately exports the smallest surface a copy-paste
adopter could need. CallSiteProfile is three bool fields because a
real designer can answer them at the whiteboard; a richer enum would
push the discriminating logic onto the caller. HybridSender exposes
raw counters (fast_path_hits, drained_total, fallback_evictions)
instead of a derived ratio, because the interesting operational alert
is "fallback evicting at all", which is binary, not statistical.
Implementation
The first chunk is the decision rubric. recommend_strategy is a
three-rule waterfall over the boolean profile — backpressure capacity
beats drop tolerance, drop tolerance beats deadline enforcement, and
deadline enforcement is the fallback when neither earlier rule fires:
pub fn recommend_strategy(profile: CallSiteProfile) -> Strategy {
if !profile.latency_sensitive && profile.can_backpressure {
return Strategy::Send;
}
if profile.drop_acceptable {
return Strategy::TrySend;
}
Strategy::SendTimeout
}
The ordering matters. A batch ingest that can throttle its own upstream
should block — that is how end-to-end rate coupling happens — even
if dropping is technically tolerable. A telemetry emitter on the
request path picks try_send over send_timeout because the deadline
machinery costs more than the data is worth. SendTimeout lands at
the bottom as the conservative default: it caps the worst case without
sacrificing any message that arrives inside the budget.
The second chunk is the hybrid sender shape. HybridSender owns the
mpsc::Sender, a VecDeque<Message> fallback, a configured cap, and
four counters. Every push is synchronous — the contract is "never
await on the hot path" — and walks three steps in order: drain the
backlog into the channel, attempt a direct send, spill into the
fallback:
pub fn push(&mut self, msg: Message) -> HybridPushOutcome {
if self.closed {
return HybridPushOutcome::Closed;
}
self.drain();
if self.closed {
return HybridPushOutcome::Closed;
}
if self.fallback.is_empty() {
return self.attempt_direct(msg);
}
self.spill(msg)
}
Draining before the direct attempt preserves FIFO ordering — a message that spilled three pushes ago lands ahead of the one the caller is offering right now. This is the invariant most naive "buffer + flush" hybrids get wrong; if you flush after the new message there is no upper bound on how stale a fallback entry can get before the consumer sees it.
The third chunk is the eviction policy. spill pushes onto the back
of the VecDeque and then checks the cap; if exceeded, it pops the
front (the oldest spilled message) and increments
fallback_evictions. This mirrors Strategy::TrySendDropOldest from
step 4 — the application loses the oldest unsent message, not the
freshest one the caller just produced:
fn spill(&mut self, msg: Message) -> HybridPushOutcome {
self.fallback.push_back(msg);
if self.fallback.len() > self.fallback_cap {
self.fallback.pop_front();
self.fallback_evictions += 1;
return HybridPushOutcome::Evicted;
}
HybridPushOutcome::Buffered
}
The fourth chunk is the shutdown producer. producer_with_shutdown
loops over total_messages and on each iteration races
shutdown.changed() against tx.reserve() inside a tokio::select!
with biased. reserve is cancellation-safe — dropping the permit
releases the channel slot, and the Message we constructed locally
never went anywhere, so a fired shutdown loses zero work:
async fn race_send_against_shutdown(
tx: &mpsc::Sender<Message>,
shutdown: &mut watch::Receiver<bool>,
msg: Message,
) -> StepOutcome {
let started = Instant::now();
tokio::select! {
biased;
_ = shutdown.changed() => StepOutcome::Shutdown,
permit = tx.reserve() => deliver_or_close(permit, msg, started),
}
}
biased is the subtle bit. Without it tokio::select! polls the two
futures in random order, which means a shutdown that arrived while a
permit was also ready could lose the race and still send one more
message. biased makes the shutdown branch unconditionally preempt
the send branch, so the moment the watch flips no further messages
land on the wire.
The fifth chunk is the receiver-gone handling. tx.reserve() returns
Err(_) when the receiver has been dropped, and the producer treats
that as a clean exit — not an error — because "the other half is
done" is a normal lifecycle event for an mpsc pair. The outcome flag
closed_gracefully distinguishes it from the shutdown-fired case via
shutdown_observed, so an operator inspecting the ShutdownOutcome
can tell which side initiated the wind-down:
fn deliver_or_close(
permit: Result<mpsc::Permit<'_, Message>, mpsc::error::SendError<()>>,
msg: Message,
started: Instant,
) -> StepOutcome {
match permit {
Ok(p) => {
p.send(msg);
StepOutcome::Sent(started.elapsed())
}
Err(_) => StepOutcome::ReceiverGone,
}
}
Fifteen tests cover the three patterns. The rubric gets five —
rubric_picks_send_for_backpressure_capable_batch_path,
rubric_picks_try_send_for_latency_sensitive_telemetry,
rubric_picks_try_send_when_backpressure_unavailable_and_drops_ok,
rubric_picks_send_timeout_for_latency_sensitive_lossless_path, and
rubric_prefers_send_over_drop_when_caller_can_backpressure. Each
asserts a single profile maps to a single strategy with a one-line
justification in the message; together they exhaustively cover the
three-rule waterfall and the priority ordering between rules.
HybridSender gets six tests pinning each push outcome and the flush
path. hybrid_sender_fast_path_when_channel_has_room enforces that an
empty channel takes the direct path with no fallback involvement.
hybrid_sender_spills_into_fallback_when_channel_full proves the
spill triggers exactly when the channel saturates.
hybrid_sender_drains_fallback_after_consumer_clears_channel is the
FIFO test — after the consumer drains, the next push must flush
backlog before the new message.
hybrid_sender_evicts_oldest_when_fallback_overflows asserts the
drop-oldest semantics on cap excess.
hybrid_sender_becomes_closed_when_receiver_dropped and
hybrid_flush_blocking_drains_backlog_into_channel cover the closed
short-circuit and the async drain used at shutdown.
producer_with_shutdown gets four tests covering the four shutdown
shapes: an upfront-set watch yields zero sends and a flag-set outcome;
a dropped receiver records closed_gracefully without
shutdown_observed; a mid-run shutdown stops the producer before
total_messages are sent and the consumer never sees more than the
producer reported; and a clean run with no shutdown delivers every
message and leaves shutdown_observed false.
Verification
cargo test
Finished `test` profile [unoptimized + debuginfo] target(s) in 18.42s
Running unittests src/lib.rs (target/debug/deps/tokio_mpsc_lab-faf64ef3387c85e5)
running 52 tests
test comparison::tests::drop_rate_zero_when_total_messages_zero ... ok
test comparison::tests::render_table_is_well_formed_and_contains_every_header ... ok
test comparison::tests::send_strategy_drops_zero_and_throughput_is_positive ... ok
test comparison::tests::send_timeout_stall_includes_per_timeout_deadline_overhead ... ok
test comparison::tests::try_send_drop_classification_separates_full_from_closed ... ok
test comparison::tests::try_send_drop_rate_matches_unified_drop_counter ... ok
test comparison::tests::try_send_producer_stall_is_tiny_compared_to_total_run ... ok
test comparison::tests::comparison_runs_all_three_strategies_in_article_order ... ok
test loadgen::tests::csv_header_lists_all_columns_in_order ... ok
test loadgen::tests::csv_row_for_send_strategy_zeros_non_blocking_counters ... ok
test loadgen::tests::csv_row_for_send_timeout_includes_deadline_ms ... ok
test loadgen::tests::csv_row_for_try_send_strategy_emits_full_observation_column ... ok
test loadgen::tests::load_generator_strategies_match_article_framing ... ok
test loadgen::tests::load_generator_workload_runs_all_three_strategies ... ok
test loadgen::tests::strategy_labels_are_stable ... ok
test production::tests::rubric_picks_send_for_backpressure_capable_batch_path ... ok
test production::tests::rubric_picks_send_timeout_for_latency_sensitive_lossless_path ... ok
test production::tests::rubric_picks_try_send_for_latency_sensitive_telemetry ... ok
test production::tests::rubric_picks_try_send_when_backpressure_unavailable_and_drops_ok ... ok
test production::tests::rubric_prefers_send_over_drop_when_caller_can_backpressure ... ok
test production::tests::hybrid_sender_becomes_closed_when_receiver_dropped ... ok
test production::tests::hybrid_sender_drains_fallback_after_consumer_clears_channel ... ok
test production::tests::hybrid_sender_evicts_oldest_when_fallback_overflows ... ok
test production::tests::hybrid_sender_fast_path_when_channel_has_room ... ok
test production::tests::hybrid_sender_spills_into_fallback_when_channel_full ... ok
test production::tests::hybrid_flush_blocking_drains_backlog_into_channel ... ok
test production::tests::dropped_receiver_treated_as_graceful_exit ... ok
test production::tests::no_shutdown_signal_delivers_every_message ... ok
test production::tests::shutdown_signal_set_upfront_yields_zero_sends ... ok
test production::tests::shutdown_during_run_stops_producer_without_panic ... ok
test tests::baseline_harness_delivers_every_message ... ok
test tests::capacity_one_forces_at_least_one_park ... ok
test tests::capacity_one_still_drains_all_messages ... ok
test tests::default_config_uses_capacity_eight ... ok
test tests::drop_oldest_fallback_evicts_earliest_messages ... ok
test tests::drop_oldest_propagates_closed_signal ... ok
test tests::drop_oldest_with_fast_consumer_delivers_all ... ok
test tests::fast_consumer_keeps_producer_on_fast_path ... ok
test tests::fast_consumer_keeps_send_latency_small ... ok
test tests::percentile_on_empty_samples_is_none ... ok
test tests::percentile_returns_a_sample_in_sorted_position ... ok
test tests::producer_stops_when_consumer_drops_receiver ... ok
test tests::send_timeout_deadline_tuning_changes_drop_rate ... ok
test tests::send_timeout_delivers_all_when_buffer_is_ample ... ok
test tests::send_timeout_distinguishes_closed_from_elapsed ... ok
test tests::send_timeout_drops_messages_when_consumer_misses_deadline ... ok
test tests::send_timeout_records_wait_time_for_successful_parks ... ok
test tests::slow_consumer_inflates_send_latency ... ok
test tests::slow_consumer_parks_producer_repeatedly ... ok
test tests::try_send_delivers_all_when_buffer_is_ample ... ok
test tests::try_send_distinguishes_closed_from_full ... ok
test tests::try_send_drops_messages_when_consumer_cannot_keep_up ... ok
test result: ok. 52 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.71s
All fifty-two tests pass — the thirty-seven carried forward from step
7 plus the fifteen new production::tests. The breakdown matches the
three deliverables: five rubric cases, six HybridSender cases, and
four producer_with_shutdown cases.
What we built
A recommend_strategy rubric that maps a three-flag CallSiteProfile
to one of the three article strategies. The waterfall is short enough
to commit to memory: backpressure-capable callers block,
drop-tolerant callers try_send, and everyone else gets
send_timeout. The five rubric tests pin both the individual mappings
and the priority ordering — including the non-obvious case where a
caller that could drop should still block if it also has upstream
to throttle.
A HybridSender that pairs try_send with a bounded producer-side
fallback queue. The hot path never awaits; bursts spill into the
fallback, drain opportunistically on later pushes, and only evict
once the cap is exceeded — and even then they evict the oldest
spilled message, never the freshest one the caller just produced.
Four counters expose how often each path fires so the operational
alert is "fallback evicting at all", not a noisy ratio.
A producer_with_shutdown task that races Sender::reserve against a
watch::Receiver<bool> inside a biased tokio::select!. A flipped
shutdown flag preempts the in-flight send, a dropped receiver
surfaces as a clean exit rather than an error, and the
ShutdownOutcome flags shutdown_observed and closed_gracefully
separately so an operator can tell which half initiated the
wind-down. Cancellation-safety of reserve means a preempted permit
never leaks a slot and never silently loses the in-flight Message.
Together the three patterns turn the measurement crate from steps 1–7 into something a service team can actually adopt. The fifty-two-test suite is the contract; the eight-step article is the rationale; the companion repo is the working reference. A reader who landed on the scoreboard in step 7 wondering "but what do I do with this?" now has a copy-pastable answer for each of the three call sites they are likely to face.
Repository
The state of the code after this step: 6efc858
Repository
Full source at https://github.com/vytharion/tokio-mpsc-bounded-send-vs-trysend-vs-timeout.
Walk the lessons by stepping through the git commits in the repo — each major step has its own commit you can git checkout and rerun.