記録 · Entry

Backpressure: when the consumer is slower than the producer

An unbounded buffer does not fix a speed mismatch, it postpones the crash. Bounded queues, blocking sends and early rejection, and where each one belongs.

9 min read

BackendDistributed SystemsReliabilityPerformanceNode.js

Somewhere in every system there is a stage that produces work faster than the next stage can consume it. A request handler that accepts uploads faster than the disk writes them. A webhook receiver that takes events faster than the worker processes them. A log shipper reading lines faster than the network sends them.

The first thing everyone does is put a buffer between the two. The producer writes into it, the consumer reads out of it, and the mismatch disappears. For a while.

A buffer absorbs a difference in timing. It cannot absorb a difference in rate. If the producer averages a thousand a second and the consumer averages eight hundred, the buffer grows by two hundred a second, indefinitely, until it runs out of memory and the process dies. The buffer did not solve anything. It converted an obvious problem, requests being slow, into a hidden one, memory climbing over hours, and then into a worse one, the whole process gone at once with everything in the buffer lost.

Backpressure is the alternative: the consumer's speed propagates back to the producer, and the producer slows down, or stops, or is told no. It is the difference between a system that degrades and a system that falls over.

#TCP did this first

Every byte on the internet already moves under backpressure. A TCP receiver advertises a window, the number of bytes it is willing to accept. The sender may not send more than that without an acknowledgement. When the receiving application stops reading from the socket, its kernel buffer fills, the advertised window shrinks to zero, and the sender stops. Not slows: stops. When the application reads again, the window opens and the bytes flow.

Nothing is dropped and nothing grows without bound. The slow reader's slowness reaches all the way back to the sender's write() call, which blocks. That chain, from a slow reader to a blocked writer, with a bounded buffer in between, is the whole idea. Every application-level version of it is the same shape.

#The three things a full queue can do

Once the buffer between two stages is bounded, the producer will eventually find it full. There are exactly three responses, and the choice is the design decision.

Block. The producer waits until there is room. This is what a TCP write does, and what a bounded channel does in most languages. Nothing is lost. The cost is that the producer is now as slow as the consumer, and whatever is upstream of the producer is now waiting on it. The pressure moves upstream, which is the point. It should keep moving until it reaches something that can legitimately wait, usually a client, or something that can legitimately say no.

Drop. Throw the new item away, or throw the oldest away to make room. Nothing waits. This is right for data where the latest value supersedes the rest: metrics samples, position updates, a video frame. It is wrong for anything where every item matters, and the danger is that a drop policy chosen for one kind of data ends up applied to another.

Reject. Tell the producer no, immediately, with an error it can act on. An HTTP 503 or 429 with a Retry-After. A channel try_send that returns the item back. This is blocking without the waiting: the caller learns at once that there is no capacity and can decide what to do, which for a client usually means backing off and retrying, and for an internal caller might mean blocking after all.

Unbounded is not on the list, because unbounded is a decision to pick one of these three later, at the worst moment, without choosing.

#Say no as early as possible

Work that is going to be rejected should be rejected before anything has been spent on it. A request that is parsed, authenticated, validated, and then dropped because the worker queue is full has consumed CPU, a connection and a database round trip to produce a 503. The same 503 at the first line of the handler costs nothing.

The right place to check capacity is the front door: before reading the body, before touching the connection pool, before anything that is itself a scarce resource. A cheap counter of in-flight requests, compared against a limit, and an immediate 503 when it is over. This is load shedding, and it is backpressure at the edge, where the thing upstream is a client that can wait or retry and where saying no costs the least.

A server that never says no does not have infinite capacity. It has a queue it did not admit to, made of TCP accept backlogs, thread pools and socket buffers, and when that overflows the failure is a timeout rather than a 503, which the client cannot distinguish from a network fault, and so retries into the thing that is already over capacity.

#In Node: the return value everybody ignores

Node streams have backpressure built in, and most code defeats it in the first line.

for (const chunk of chunks) {
  writable.write(chunk)
}

write() returns a boolean. false means the internal buffer is past its highWaterMark and the caller should stop writing until the 'drain' event. Ignoring the return value does not lose data; it buffers it, without bound, in the process's memory, which is exactly the failure this post opened with. A loop like the one above pushing a large file into a slow socket will hold the entire file in memory.

for (const chunk of chunks) {
  if (!writable.write(chunk)) {
    await once(writable, 'drain')
  }
}

Or skip the bookkeeping entirely:

await pipeline(readable, transform, writable)

stream.pipeline and readable.pipe() handle the pause and resume for you, and pause the source when the destination is full. If a stream is involved and you are calling write() in a loop, you are almost certainly building the unbounded buffer by hand.

The same rule covers async iteration over a readable: for await (const chunk of readable) pulls one chunk at a time and does not read the next until the loop body finishes. Slow body, slow reads, and the pressure reaches the source.

#In Rust: the channel has a capacity for a reason

Tokio offers two mpsc channels, and the choice between them is the choice this post is about.

let (tx, rx) = tokio::sync::mpsc::channel::<Job>(256);

A bounded channel. tx.send(job).await completes at once while there is room and suspends the sending task when there is not. The producer is paused, not by anything it wrote, but by the shape of the channel. Two hundred and fifty-six jobs is the most that can ever be waiting. Memory is bounded. The consumer's speed is the producer's speed.

let (tx, rx) = tokio::sync::mpsc::unbounded_channel::<Job>();

tx.send(job) never waits and never fails while the receiver lives. It is the buffer that grows by two hundred a second. It has its place, in code paths that must not block for correctness reasons, and it should be treated as a decision that someone else will run out of memory later.

For the reject option, tx.try_send(job) returns Err(TrySendError::Full(job)) and hands the job back. That is the 503 in channel form.

The capacity number is worth a moment's thought. It should be large enough to absorb a burst the consumer can catch up on, and small enough that the latency it adds is acceptable: a queue of 256 jobs that take 10 milliseconds each is two and a half seconds of delay for the job at the back. Little's law applies to channels exactly as it does to connection pools: depth divided by throughput is the wait.

#Concurrency limits are backpressure too

A queue limits how much is waiting. A semaphore limits how much is running. Both are needed, and they are often confused.

let permits = Arc::new(Semaphore::new(32));

let permit = permits.clone().acquire_owned().await?;
tokio::spawn(async move {
    handle(job).await;
    drop(permit);
});

Thirty-two jobs in flight, and the thirty-third waits at acquire. Spawning a task per item with no limit is the unbounded queue in a different costume: every task is real memory and a real slot in the scheduler, and a burst of a hundred thousand items is a hundred thousand tasks.

The right limit is not a fixed number for very long. A downstream that is healthy can take fifty concurrent calls; the same downstream under load can take ten, and sending fifty makes it worse. Adaptive concurrency limiters treat the observed latency as the signal: when latency rises, the limit falls, and when it recovers, the limit climbs. Netflix's concurrency-limits library is the reference implementation, and the algorithm underneath, gradient descent on latency, is the same idea TCP uses to find the bandwidth of a path. A fixed limit is a fine start; an adaptive one is what you move to when the fixed number is wrong at different hours of the day.

#The pull-based consumer is already correct

Kafka consumers, most database cursors, and any API with a next() method share one property: the consumer asks for the next item when it wants one. Nothing is pushed into it. A slow consumer simply asks less often, and the backlog accumulates at the source, where it is durable and where lag can be measured, rather than in the consumer's memory, where it is neither.

This is why the classic mistake in a Kafka consumer is to read a batch and dispatch every message to a task without waiting: the consumer has converted itself from pull to push, and put an unbounded queue in its own heap. The fix is a bounded channel or a semaphore between the poll loop and the handlers, so the loop stops polling when the handlers are full. Kafka's pause() is there for exactly that.

#Deadlines are what stop a blocked producer waiting forever

Blocking is the polite response to a full queue, and it has one failure mode: the producer waits, and whatever is waiting on the producer waits, and the waiting is indefinite. A request that blocks on a full channel for fifty seconds is a request that should have failed at one.

Every wait for capacity should carry a deadline, and the deadline should be the remaining budget of the request that is waiting, not a constant. tokio::time::timeout(remaining, tx.send(job)) turns a block into a reject when the budget runs out, which is what a caller with a propagated deadline needs to hear.

#The one-line summary

Bound every buffer, decide in advance whether a full one blocks, drops or rejects, push that decision as far toward the edge as it will go, and give every wait a deadline. The system will run at the speed of its slowest stage either way; backpressure is how it does that without falling over.

#References

Related entries