09 - Backpressure and Flow Control: Managing the Case Where the Producer Outperforms the Consumer

The problem: the queue nobody watches

Imagine a producer that generates 100,000 events per second and a consumer that handles 50,000. The queue in between grows by 50,000 items per second; in an hour that is 50,000 × 3,600 = 180 million items waiting, each with its own object and its own memory cost. The process runs out of memory and falls over, but before that something less visible happens: every item waits longer and longer, and the latency the user sees grows without any error appearing in the logs. Backpressure is the mechanism by which the consumer tells upstream that it must slow down, so that the queue stays small and the wait predictable.

In this article the problem is measured first, then solved in four different ways, each with a program that was run: two Java queues compared, the JDK's Reactive Streams, the Reactor library and asyncio.Queue with Python's asynchronous generators. All the programs were run with OpenJDK 21.0.13, Reactor 3.7.1 (from a local Maven cache) and Python 3.14.4; timings change from one machine to another. For Go and Rust the mechanism is the same and is described in goroutines and channels and in Rust and Tokio.

Measuring the damage: unbounded queue and bounded queue

The first experiment is deliberately simple. A producer inserts 20 items every millisecond (about 20,000 per second), a consumer handles one per millisecond. The queue is first an unbounded LinkedBlockingQueue, then an ArrayBlockingQueue of 100 slots. Each item carries its own birth instant, so the consumer measures how long the last one extracted had waited.

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;

// The producer generates 20 items every millisecond (~20,000/s); the consumer handles one
// per millisecond (~1,000/s or fewer). What happens to the queue in between, over 3 seconds?
public class Code {
    static void esegui(String nome, BlockingQueue<Long> coda) throws Exception {
        AtomicLong prodotti = new AtomicLong(), consumati = new AtomicLong();
        long fine = System.nanoTime() + 3_000_000_000L;

        Thread produttore = new Thread(() -> {
            try {
                while (System.nanoTime() < fine) {
                    for (int i = 0; i < 20; i++) {
                        coda.put(System.nanoTime());          // put(): waits if the queue is full
                        prodotti.incrementAndGet();
                    }
                    Thread.sleep(1);
                }
            } catch (InterruptedException e) { }
        });
        long[] attesaUltimo = new long[1];
        Thread consumatore = new Thread(() -> {
            try {
                while (true) {
                    long nato = coda.take();
                    attesaUltimo[0] = System.nanoTime() - nato;   // how long it waited in the queue
                    consumati.incrementAndGet();
                    Thread.sleep(1);                              // slow processing
                }
            } catch (InterruptedException e) { }
        });
        produttore.setDaemon(true); consumatore.setDaemon(true);
        produttore.start(); consumatore.start();
        Thread.sleep(3_000);
        System.out.printf("%-22s produced %6d, consumed %5d, in queue %6d, wait of the last item ~%d ms%n",
            nome, prodotti.get(), consumati.get(), coda.size(), attesaUltimo[0] / 1_000_000);
    }

    public static void main(String[] args) throws Exception {
        esegui("unbounded queue", new LinkedBlockingQueue<>());
        esegui("queue bounded to 100", new ArrayBlockingQueue<>(100));
    }
}

// Output:
// unbounded queue        produced  51960, consumed  2564, in queue  49396, wait of the last item ~2842 ms
// queue bounded to 100   produced   2488, consumed  2388, in queue    100, wait of the last item ~132 ms

In the unbounded queue the producer inserted 51,960 items, the consumer handled 2,564, and 49,396 remain in the queue: the last one extracted had waited about 2.8 seconds, practically the whole duration of the test. In the queue of 100 the producer got 2,488 insertions in three seconds, just above the 2,388 consumed: put() forced it to keep the consumer's pace. The wait of the last item is 132 ms, and not by chance: with 100 items ahead and a consumer of about 800 per second you expect about 125 ms (100 divided by 800). The size of the queue is the maximum latency, and choosing it means choosing that latency.

This is also why a large buffer "just in case" is a wrong choice: it does not solve the imbalance, it hides it until memory runs out. If the load is really exceeding capacity, the only serious alternative to the bounded queue is to add consumers, not to lengthen the queue.

The four answers when the queue is full

A bounded queue that fills up forces a decision. There are four strategies, and the choice depends on what a lost item is worth and on what the producer can do.

• Wait (block): the producer stops until there is room. No loss, but the delay propagates upstream. For pipelines in which every item counts.
• Drop the new (drop newest): the incoming item is rejected. The producer does not slow down. For metrics, non-critical logs, interface events.
• Drop the oldest (drop oldest): room is made by removing the old one. For data where the last value matters, like a price or a position.
• Fail (error): an error is signalled at once. For operations where neither losing nor waiting is acceptable, and the caller must decide.

The first two answer the question "can I lose data?", the other two "which data do I lose?". The practical criterion is to start from the value of the item: if it is an order, you wait or fail; if it is a measurement that will be replaced in a second, you drop. Dropping silently is the common mistake: every drop must be counted and made visible with a metric, otherwise the system "works" while it loses data.

Reactive Streams: the demand travels upstream

With a blocking queue, the consumer never says how much it wants: it just drains. The Reactive Streams specifications, born for "asynchronous stream processing with non-blocking back pressure", reverse the direction: the Subscriber asks request(n) and the Publisher may not emit more than requested. Since JDK 9 the interfaces are in the standard library, in java.util.concurrent.Flow, in one-to-one correspondence with those of the specification. The listing uses SubmissionPublisher with a buffer of 16 slots and a slow subscriber, which takes 10 ms per item and asks for one at a time.

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

// java.util.concurrent.Flow and SubmissionPublisher: Reactive Streams in the JDK, no libraries.
public class Flusso {
    // A Subscriber that asks for ONE item at a time and takes 10 ms to handle it.
    static class Lento implements Flow.Subscriber<Integer> {
        final AtomicInteger ricevuti = new AtomicInteger();
        final CountDownLatch finito = new CountDownLatch(1);
        Flow.Subscription sub;
        public void onSubscribe(Flow.Subscription s) { sub = s; s.request(1); }
        public void onNext(Integer v) {
            try { Thread.sleep(10); } catch (InterruptedException e) { }
            ricevuti.incrementAndGet();
            sub.request(1);                       // "I am done, send another one"
        }
        public void onError(Throwable t) { t.printStackTrace(); }
        public void onComplete() { finito.countDown(); }
    }

    public static void main(String[] args) throws Exception {
        // 1. submit(): when the buffer (16 slots) is full, the producer STOPS
        Lento a = new Lento();
        long t0 = System.nanoTime();
        try (SubmissionPublisher<Integer> pub = new SubmissionPublisher<>(ForkJoinPool.commonPool(), 16)) {
            pub.subscribe(a);
            for (int i = 0; i < 100; i++) pub.submit(i);
            long tProduttore = (System.nanoTime() - t0) / 1_000_000;
            System.out.println("submit(): the producer finished after " + tProduttore + " ms");
        }
        a.finito.await();
        System.out.println("submit(): received " + a.ricevuti.get() + " of 100, total " + (System.nanoTime() - t0) / 1_000_000 + " ms");

        // 2. offer(): does not block, rejection is the producer's decision (here: drop and count)
        Lento b = new Lento();
        AtomicInteger scartati = new AtomicInteger();
        t0 = System.nanoTime();
        try (SubmissionPublisher<Integer> pub = new SubmissionPublisher<>(ForkJoinPool.commonPool(), 16)) {
            pub.subscribe(b);
            for (int i = 0; i < 100; i++) {
                pub.offer(i, (sub, item) -> { scartati.incrementAndGet(); return false; });  // false = do not retry
            }
            System.out.println("offer(): the producer finished after " + (System.nanoTime() - t0) / 1_000_000 + " ms");
        }
        b.finito.await();
        System.out.println("offer(): received " + b.ricevuti.get() + ", dropped " + scartati.get()
            + ", sum " + (b.ricevuti.get() + scartati.get()));
    }
}

// Output:
// submit(): the producer finished after 1015 ms
// submit(): received 100 of 100, total 1066 ms
// offer(): the producer finished after 10 ms
// offer(): received 16, dropped 84, sum 100

The Javadoc of SubmissionPublisher describes the two behaviors: submit blocks until there are resources (the simplest policy, the least responsive), while the offer methods may drop and offer a point to intercept the drop and retry. In the first run the producer took 1,015 ms to insert 100 items, because it had to wait for them to be handled, and the subscriber received all 100. In the second it finished in 10 ms but only 16 items were received: 84 were dropped and counted, and the sum comes to 100. The two runs are the cleanest example of the choice between "wait" and "drop the new".

Reactor: declaring the strategy in one line

Libraries like Project Reactor put the strategies inside the pipeline. The Reactor guide describes the real mechanism: the subscriber can work in unbounded mode, letting the source push at full speed, or use request to tell the source it is ready to handle at most n items. The listing tries three things: requesting in batches of 10, a buffer that drops the oldest and an onBackpressureDrop.

import org.reactivestreams.Subscription;
import reactor.core.publisher.*;
import reactor.core.scheduler.Schedulers;
import java.util.concurrent.atomic.AtomicInteger;

public class Reattivo {
    public static void main(String[] args) throws Exception {
        // 1. Pull: the consumer asks for 10 items at a time; the publisher emits no more.
        StringBuilder richieste = new StringBuilder();
        Flux.range(1, 30)
            .doOnRequest(n -> richieste.append(n).append(' '))
            .subscribe(new BaseSubscriber<Integer>() {
                @Override protected void hookOnSubscribe(Subscription s) { request(10); }
                @Override protected void hookOnNext(Integer v) {
                    if (v % 10 == 0) request(10);          // I have handled 10: I ask for the next 10
                }
            });
        System.out.println("request(n) received by the publisher: " + richieste.toString().trim());

        // 2. Fast source, slow consumer, buffer of 100 that DROPS THE OLDEST
        AtomicInteger consumati = new AtomicInteger();
        AtomicInteger scartati = new AtomicInteger();
        AtomicInteger ultimo = new AtomicInteger();
        Flux.range(1, 5_000)
            .onBackpressureBuffer(100, scartati_ -> scartati.incrementAndGet(),
                                  BufferOverflowStrategy.DROP_OLDEST)
            .publishOn(Schedulers.single(), 1)           // the consumer asks for 1 at a time
            .doOnNext(v -> { try { Thread.sleep(1); } catch (InterruptedException e) { } })
            .doOnNext(ultimo::set)
            .blockLast();
        // consumed = all those that went through the consumer
        System.out.println("DROP_OLDEST: dropped " + scartati.get() + " of 5000; last item received: " + ultimo.get());

        // 3. Same source with onBackpressureDrop: drops the NEW ones when the consumer has not requested
        AtomicInteger scartatiDrop = new AtomicInteger();
        AtomicInteger arrivati = new AtomicInteger();
        Flux.range(1, 5_000)
            .onBackpressureDrop(v -> scartatiDrop.incrementAndGet())
            .publishOn(Schedulers.single(), 1)
            .doOnNext(v -> { try { Thread.sleep(1); } catch (InterruptedException e) { } })
            .doOnNext(v -> arrivati.incrementAndGet())
            .blockLast();
        System.out.println("onBackpressureDrop: arrived " + arrivati.get() + ", dropped " + scartatiDrop.get()
            + ", sum " + (arrivati.get() + scartatiDrop.get()));
    }
}

// Output:
// request(n) received by the publisher: 10 10 10 10
// DROP_OLDEST: dropped 4895 of 5000; last item received: 5000
// onBackpressureDrop: arrived 2, dropped 4998, sum 5000

In the first part, the publisher receives four requests of 10 for thirty items: the first at subscription and the other three when the subscriber has handled 10. In the second, a buffer of 100 that drops the oldest let 4,895 items out of 5,000 fall, but delivered 5,000 last: it is the strategy for which "the most recent data is worth more". In the third, onBackpressureDrop let through only 2 items: the source is synchronous, it emits 5,000 in a few microseconds, and the consumer (which asks for one item at a time) requested very few in that span of time. The exact number depends on scheduling, but the sum (arrived + dropped) is always 5,000: every item is accounted for.

Python: bounded queues and asynchronous generators

In asyncio the "wait" answer is already in the queue. The documentation of asyncio.Queue says that, with maxsize greater than zero, await put() blocks when the queue reaches maxsize, and that with zero the queue is infinite. The listing produces a thousand items and consumes one every 2 ms, first with an unbounded queue and then with maxsize=10.

import asyncio
import time

async def prova(maxsize: int) -> None:
    coda: asyncio.Queue[int] = asyncio.Queue(maxsize=maxsize)   # 0 = unbounded
    picco = 0
    t0 = time.monotonic()
    produttore_finito = 0.0

    async def produttore():
        nonlocal produttore_finito
        for i in range(1000):
            await coda.put(i)            # with maxsize > 0 it suspends when the queue is full
        await coda.put(-1)               # sentinel
        produttore_finito = time.monotonic() - t0

    async def consumatore():
        nonlocal picco
        while True:
            picco = max(picco, coda.qsize())
            voce = await coda.get()
            if voce == -1:
                return
            await asyncio.sleep(0.002)   # slow processing

    async with asyncio.TaskGroup() as tg:
        tg.create_task(produttore())
        tg.create_task(consumatore())
    etichetta = "unbounded" if maxsize == 0 else f"maxsize={maxsize}"
    print(f"{etichetta:<14} queue at peak: {picco:4d}, producer finished after {produttore_finito:.1f} s, "
          f"all finished after {time.monotonic() - t0:.1f} s")

async def main():
    await prova(0)
    await prova(10)

asyncio.run(main())

# Output:
# unbounded      queue at peak: 1001, producer finished after 0.0 s, all finished after 2.3 s
# maxsize=10     queue at peak:   10, producer finished after 2.3 s, all finished after 2.3 s

With the unbounded queue the producer finished practically at once (0.0 s) and the queue reached a peak of 1,001 items: all the work is in memory before it is done. With maxsize=10 the peak is 10 and the producer finished after 2.3 s, together with the consumer: it walked at the pace of whoever consumes. The total time is the same in both cases (2.3 s), confirming that the bounded queue does not slow the system, it only slows whoever produces. The sentinel -1 closes the consumer, and the TaskGroup (see advanced asyncio) waits for both.

The second tool is the asynchronous generator: backpressure is implicit, because the generator suspends at every yield and resumes only when the consumer asks for the next value with async for. The listing prints both sides to show the alternating pace.

import asyncio

async def eventi():
    for i in range(4):
        print(f"  generator: producing {i}")
        yield i                          # it suspends here until the consumer asks for the next one

async def main():
    async for evento in eventi():
        print(f"consumer: received {evento}, processing it")
        await asyncio.sleep(0.01)

asyncio.run(main())

# Output:
#   generator: producing 0
# consumer: received 0, processing it
#   generator: producing 1
# consumer: received 1, processing it
#   generator: producing 2
# consumer: received 2, processing it
#   generator: producing 3
# consumer: received 3, processing it

The lines alternate exactly: the generator produces a value, the consumer processes it (including the 10 ms sleep), and only then does the generator produce the next. The model is "pull": there is no queue, so there is nothing to size. The limit is that a generator serves one consumer at a time and does not overlap production and processing; to do that you need a bounded queue in the middle again.

Go and Rust: the same principle, no examples here

In Go a buffered channel does the same thing as the bounded queue: the specification says a send waits when the buffer is full, and the buffer determines the maximum latency as in the first experiment. In Tokio, the documentation of mpsc::Sender::send says "sends a value, waiting until there is capacity"; an alternative that does not wait, try_send, returns an error if the channel is full, and it is the way to implement "drop the new" or "fail". I report no Go or Rust listings here, because the machine does not have the respective compilers and they would not have been run.

What you can do now

You can calculate how much an unbounded queue would grow and why its size coincides with the maximum latency, choose among the four answers according to the value of the item, and count what you drop. You can use a bounded queue, a SubmissionPublisher or an asynchronous generator as the case requires, and you have seen in numbers that a bounded queue does not remove capacity, it makes it visible.

The next and last article takes the same problems to the distributed scale: CAP theorem, consistency and CRDTs. For those who first want the opposite case, a lock that blocks everyone, see race conditions and deadlocks. How an event loop handles an event queue that fills up is in the article on the JavaScript event loop.

The counts (51,960, 2,564, 4,895, 1,001) and the timings (1,015 ms, 2.3 s) come from single runs on one machine; the order of magnitude and the "received + dropped" sum are stable, the specific values are not. The claims about Go and Tokio are taken from the sources above, not from code that was run.