Backpressure

Deep dive

How it works

In: items from a producer, a capacity (items or bytes) and a full-queue policy · Out: items delivered in order, plus a wait, drop or error signal to the producer · State: the queue or a credit counter, at most capacity entries · Cost: O(1) per item; memory ≤ capacity × item size.

1. Pick a bound in items or bytes. Little's law (L = λW, Little 1961) gives the wait: a full queue of C items served at rate μ delays the last item by C/μ.
2. Choose what a full queue does. Block/await: send parks the producer until a slot frees. Drop: discard the newest (keep history) or the oldest (keep fresh data). Reject: return an error so the caller can retry later or show a busy response (load shedding). Credit: the sender may not send until the receiver has granted room.
3. With credits the receiver publishes a limit; the sender tracks sent and sends only while sent < limit. The receiver raises the limit as it consumes, so its buffer never exceeds the window.
4. Chain it: each hop's bound must push back on the hop before it, up to the source (a client, a file reader, a Kafka partition). One unbounded hop hides the problem.
5. Choose the signal by source: a source that can wait (file, socket, database cursor) is blocked or paused; one that cannot (sensor, mouse, feed) is dropped or sampled.

How the real systems do it:

• Reactive Streams (JVM spec): a Subscriber must call request(n) to receive onNext, and total signals must never exceed total demand (rule 1.1). The spec derives the buffer bound P − B − N from requested minus received and buffered.
• HTTP/2 (RFC 9113 §5.2): credit-based, per stream and for the connection; only DATA frames count; WINDOW_UPDATE adds 1 to 2³¹−1 octets. Advertising 2³¹−1 and refilling on every receipt effectively disables control.
• QUIC (RFC 9000 §4): limits are absolute byte offsets, MAX_STREAM_DATA per stream and MAX_DATA for the connection; a sender violating them gets FLOW_CONTROL_ERROR. A blocked sender may send DATA_BLOCKED, but the receiver must not wait for it.
• TCP (RFC 9293): the receiver advertises a window in every segment (16-bit field, scaled by a window-scale option); at zero window the sender probes.
• Kafka consumers: pause(partitions) suspends fetching from those partitions without triggering a rebalance; resume restarts it; the pause is not preserved across a rebalance.
• Node.js streams: write() returns false once the buffer reaches highWaterMark; wait for 'drain' before writing again. Ignoring it makes Node buffer everything until memory runs out.
• Tokio: mpsc::channel(n)'s send().await waits for capacity; try_send fails at once when full or closed; reserve() waits first and returns a permit, so the message is not lost if the await is cancelled.
• Go: make(chan T, n) blocks the sender when the buffer is full; select with a default case turns a send into a non-blocking attempt.

Code

use std::collections::VecDeque;
use std::hint::black_box;
use std::sync::atomic::{AtomicUsize, Ordering::*};
use std::sync::{mpsc, Arc};
use std::thread;
use std::time::Instant;

/// Producer sends `n` items as fast as it can; consumer spends ~`work` loop iterations per item.
/// `bound = Some(c)` uses a bounded channel (send blocks when full), `None` an unbounded one.
/// Returns (max items in flight, elapsed seconds).
fn pipeline(bound: Option<usize>, n: usize, work: u64) -> (usize, f64) {
    let depth = Arc::new(AtomicUsize::new(0)); // sent but not yet received
    let max = Arc::new(AtomicUsize::new(0));
    let (tx, rx): (Box<dyn Fn(u64) + Send>, Box<dyn Fn() -> Option<u64> + Send>) = match bound {
        Some(c) => {
            let (t, r) = mpsc::sync_channel::<u64>(c);
            (Box::new(move |v| t.send(v).unwrap()), Box::new(move || r.recv().ok()))
        }
        None => {
            let (t, r) = mpsc::channel::<u64>();
            (Box::new(move |v| t.send(v).unwrap()), Box::new(move || r.recv().ok()))
        }
    };
    let t0 = Instant::now();
    let (d, m) = (depth.clone(), max.clone());
    let producer = thread::spawn(move || {
        for i in 0..n as u64 {
            let now = d.fetch_add(1, SeqCst) + 1; // count before send: includes the item a blocked producer holds, and one being processed
            m.fetch_max(now, SeqCst);
            tx(i); // blocks here when the bounded queue is full
        }
    });
    let mut sum = 0u64;
    while let Some(v) = rx() {
        depth.fetch_sub(1, SeqCst);
        for k in 0..work { sum = black_box(sum.wrapping_add(k ^ v)); } // slow consumer
    }
    producer.join().unwrap();
    (max.load(SeqCst), t0.elapsed().as_secs_f64())
}

/// QUIC/HTTP-2 style credit: the receiver publishes an absolute limit; the sender never passes it.
struct Link { sent: u64, limit: u64, consumed: u64, window: u64, buf: VecDeque<u64> }
impl Link {
    fn new(window: u64) -> Self { Link { sent: 0, limit: window, consumed: 0, window, buf: VecDeque::new() } }
    fn sender_tick(&mut self) -> bool { // sender: send one item if credit remains
        if self.sent < self.limit { self.buf.push_back(self.sent); self.sent += 1; true } else { false }
    }
    fn receiver_tick(&mut self) { // receiver: consume one item, top up credit once half the window is used
        if self.buf.pop_front().is_some() { self.consumed += 1; }
        if self.consumed + self.window >= self.limit + self.window / 2 { self.limit = self.consumed + self.window; }
    }
}

enum Policy { DropOldest, Disconnect }
/// Per-client send buffer for a fan-out server: never holds more than `cap` messages.
struct Client { q: VecDeque<u32>, cap: usize, policy: Policy, dropped: u32, alive: bool }
impl Client {
    fn push(&mut self, m: u32) {
        if !self.alive { return; }
        if self.q.len() == self.cap {
            match self.policy {
                Policy::DropOldest => { self.q.pop_front(); self.dropped += 1; }
                Policy::Disconnect => { self.alive = false; self.q.clear(); return; }
            }
        }
        self.q.push_back(m);
    }
}

fn main() {
    let (n, work) = (200_000, 2_000);
    let (d_b, t_b) = pipeline(Some(64), n, work);
    let (d_u, t_u) = pipeline(None, n, work);
    println!("bounded(64): max in flight {d_b}, {t_b:.2} s | unbounded: max in flight {d_u}, {t_u:.2} s");
    assert!(d_b <= 64 + 2); // 64 queued + 1 being processed + 1 held by the blocked producer
    assert!(d_u > 10 * d_b); // the unbounded queue just keeps growing

    let mut l = Link::new(8);
    let mut max_buf = 0;
    for t in 0..10_000 {
        l.sender_tick();
        if t % 3 == 0 { l.receiver_tick(); } // receiver is 3x slower than the sender
        assert!(l.sent <= l.limit); // credit invariant: never send beyond the granted limit
        assert!(l.buf.len() as u64 <= l.window); // so the receive buffer is bounded by the window
        max_buf = max_buf.max(l.buf.len());
    }
    assert!(l.sent > 3_000 && l.sent < 3_400); // throughput follows the receiver, not the sender
    println!("credit window 8: max buffered {max_buf}, sent {} of 10000 attempts", l.sent);

    let mut slow = Client { q: VecDeque::new(), cap: 4, policy: Policy::DropOldest, dropped: 0, alive: true };
    let mut gone = Client { q: VecDeque::new(), cap: 4, policy: Policy::Disconnect, dropped: 0, alive: true };
    for m in 0..100 {
        slow.push(m); gone.push(m);
        if m % 10 == 0 { slow.q.pop_front(); } // the slow client drains 1 in 10
        assert!(slow.q.len() <= 4 && gone.q.len() <= 4);
    }
    assert!(slow.dropped > 0 && slow.q.back() == Some(&99)); // newest data kept
    assert!(!gone.alive); // slow connection cut instead of buffering forever
    println!("client buffers: drop-oldest dropped {}, disconnect alive={}", slow.dropped, gone.alive);
}

Compile with rustc -O; the asserts check the queue bound, the credit invariant and the per-client buffer bound. Measured numbers are under Complexity and constants.

Complexity and constants

• Per item: O(1) for send, receive and credit update; a full queue's wait is bounded by C/μ.
• Memory: capacity × size_of::<T>() plus queue overhead; a credit window of W bytes bounds the receive buffer at W.
• Measured on this Mac (Apple M4 Pro), 200,000 items, consumer about 2 µs per item, 5 runs: sync_channel(64) peaked at 66 in flight (64 queued, 1 being processed, 1 held by the blocked producer) and finished in 0.54-0.73 s; the unbounded channel peaked at about 199,400 in flight (nearly everything) and finished in 0.42-0.47 s.
• Credit simulation, window 8, receiver 3× slower than the sender, measured on this Mac (Apple M4 Pro): peak 7 buffered, 3,340 of 10,000 send attempts accepted, so throughput equals the receiver's rate.
• Window size: to keep a link full the window must exceed bandwidth × delay (RFC 9113 §5.2.2); a smaller window caps throughput at window ÷ RTT.

Efficiency

• Cost: a bounded hand-off costs one lock or atomic step per item, plus a thread wake-up when producer or consumer parks. Measured on this Mac (Apple M4 Pro): the bounded run was about 25-30 % slower than the unbounded run in the test above (cause not profiled), but held peak memory at 66 items instead of about 199,000. The bound is the consumer's service rate; the queue only buffers jitter.
• Levers:
  • Larger window or capacity — fewer producer stalls, more memory held (RFC 9113 §5.2.2-5.2.3) — until it exceeds bandwidth × delay or the burst size.
  • Batched credit updates — fewer control frames; QUIC calls frequent small MAX_DATA updates undesirable (RFC 9000 §4.2) — when the receiver drains in bursts.
  • Reserve before producing — no wasted work on items that cannot be queued (Tokio reserve) — when producing is expensive.
  • Drop at the edge — a rejected item that was never parsed or enqueued costs least — overload with a replaceable source.
  • Autotuned window — grows with RTT and consumption rate (RFC 9000 §4.2) — long-lived connections.
• Naive → standard → state of the art:
  • Unbounded queue and a retry loop · memory and latency grow until a crash or timeout storm.
  • Bounded queue with blocking send, or try_send and a drop/reject policy · wait ≤ C/μ, memory ≤ C.
  • Credits chained through every hop, autotuned window, load shedding at the edge · bounded memory per hop, sender paced to the slowest stage.

Implementation notes

• Count in bytes when item sizes vary; a bound of 1,000 items can mean 1 KB or 1 GB.
• Drop-oldest suits state and telemetry (the newest value supersedes); drop-newest suits queued commands whose order matters; disconnect suits clients that stay slow.
• Count and export drops, rejects and time blocked; they are the overload signal.
• A blocked producer holds resources (a connection, a transaction); give send a timeout or deadline.
• Never block an event-loop thread on a full queue; await it, or try_send and reject.
• A receiver must grant more credit without waiting for the sender to ask (RFC 9000 §4.2), and must keep reading control frames (RFC 9113 §5.2.2) or WINDOW_UPDATE can deadlock behind data.

Pitfalls

• "Just make the buffer bigger": the backlog is still unbounded in time; by Little's law the wait grows with it.
• Dropping silently: lost data with no metric.
• A bound in one hop and an unbounded buffer in the next (a runtime or library channel) moves the backlog without removing it.
• A cancelled send().await inside Tokio select! loses the message; use reserve.
• A cycle of full queues (each stage both producer and consumer) can deadlock.
• Retrying a rejection without backoff re-creates the overload.

Variants and alternatives

In production

HTTP/2 and QUIC stacks (browsers, servers, proxies), Reactive Streams libraries on the JVM, Tokio and Go channels in services, Node.js streams for files, sockets and HTTP bodies, Kafka consumers that pause partitions, and fan-out servers that cap each client's outbound buffer (pub/sub fan-out, WebSockets). Actor mailboxes use the same bound (actor model); see also the bounded MPMC queue, caches and queues, QUIC and graceful draining.

Implementations

Sources

• John D. C. Little, A Proof for the Queuing Formula: L = λW, Operations Research 9(3), 1961, pp. 383-387. paper
• Reactive Streams specification (JVM), v1.0.4. spec · site
• M. Thomson, C. Benfield (eds.), HTTP/2, RFC 9113, 2022, §5.2 and §6.9. RFC
• J. Iyengar, M. Thomson (eds.), QUIC: A UDP-Based Multiplexed and Secure Transport, RFC 9000, 2021, §4. RFC
• W. Eddy (ed.), Transmission Control Protocol (TCP), RFC 9293, 2022, §3.8.6. RFC
• Tokio documentation, tokio::sync::mpsc. docs
• Node.js documentation, Stream. docs
• The Go Programming Language Specification, channel types and select statements. spec
• Apache Kafka documentation, KafkaConsumer (pause, resume). javadoc