09 - Backpressure e Flow Control: Gestire il Caso in cui il Producer Supera il Consumer

Il problema: la coda che nessuno guarda

Immagina un produttore che genera 100.000 eventi al secondo e un consumatore che ne smaltisce 50.000. La coda in mezzo cresce di 50.000 elementi al secondo; in un'ora sono 50.000 × 3.600 = 180 milioni di elementi in attesa, ciascuno con il suo oggetto e il suo costo di memoria. Il processo finisce la memoria e cade, ma prima succede una cosa meno visibile: ogni elemento aspetta sempre di più, e la latenza che l'utente vede cresce senza che nessun errore appaia nei log. La backpressure (contropressione) è il meccanismo con cui il consumatore comunica a monte che deve rallentare, così che la coda resti piccola e l'attesa prevedibile.

In questo articolo il problema viene misurato prima, poi risolto in quattro modi diversi, ciascuno con un programma eseguito: due code Java a confronto, i Reactive Streams della JDK, la libreria Reactor e asyncio.Queue con i generatori asincroni di Python. Tutti i programmi sono stati eseguiti con OpenJDK 21.0.13, Reactor 3.7.1 (da una cache Maven locale) e Python 3.14.4; i tempi cambiano da una macchina all'altra. Per Go e Rust il meccanismo è lo stesso ed è descritto in goroutine e channel e in Rust e Tokio.

Misurare il danno: coda illimitata e coda limitata

Il primo esperimento è volutamente semplice. Un produttore inserisce 20 elementi ogni millisecondo (circa 20.000 al secondo), un consumatore ne smaltisce uno per millisecondo. La coda è prima una LinkedBlockingQueue senza limite, poi una ArrayBlockingQueue da 100 posti. Ogni elemento porta il proprio istante di nascita, così il consumatore misura quanto ha atteso l'ultimo estratto.

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

// Il produttore genera 20 elementi ogni millisecondo (~20.000/s); il consumatore ne smaltisce uno
// per millisecondo (~1.000/s o meno). Cosa succede alla coda nel mezzo, per 3 secondi?
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(): attende se la coda e' piena
                        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;   // quanto ha atteso in coda
                    consumati.incrementAndGet();
                    Thread.sleep(1);                              // elaborazione lenta
                }
            } catch (InterruptedException e) { }
        });
        produttore.setDaemon(true); consumatore.setDaemon(true);
        produttore.start(); consumatore.start();
        Thread.sleep(3_000);
        System.out.printf("%-22s prodotti %6d, consumati %5d, in coda %6d, attesa dell'ultimo elemento ~%d ms%n",
            nome, prodotti.get(), consumati.get(), coda.size(), attesaUltimo[0] / 1_000_000);
    }

    public static void main(String[] args) throws Exception {
        esegui("coda illimitata", new LinkedBlockingQueue<>());
        esegui("coda limitata a 100", new ArrayBlockingQueue<>(100));
    }
}

// Output:
// coda illimitata        prodotti  51960, consumati  2564, in coda  49396, attesa dell'ultimo elemento ~2842 ms
// coda limitata a 100    prodotti   2488, consumati  2388, in coda    100, attesa dell'ultimo elemento ~132 ms

Nella coda illimitata il produttore ha inserito 51.960 elementi, il consumatore ne ha smaltiti 2.564, e in coda ne restano 49.396: l'ultimo estratto aveva atteso circa 2,8 secondi, praticamente l'intera durata della prova. Nella coda da 100 il produttore ha ottenuto 2.488 inserimenti in tre secondi, appena sopra i 2.388 consumati: put() lo ha costretto ad andare al passo del consumatore. L'attesa dell'ultimo elemento è 132 ms, e non per caso: con 100 elementi davanti e un consumatore da circa 800 al secondo ci si aspetta circa 125 ms (100 diviso 800). La dimensione della coda è la latenza massima, e sceglierla significa scegliere quella latenza.

Questo è anche il motivo per cui un buffer grande «per sicurezza» è una scelta sbagliata: non risolve lo squilibrio, lo nasconde finché la memoria non finisce. Se il carico sta davvero superando la capacità, l'unica alternativa seria alla coda limitata è aggiungere consumatori, non allungare la coda.

Le quattro risposte quando la coda è piena

Una coda limitata che si riempie obbliga a decidere. Le strategie sono quattro, e la scelta dipende da che cosa vale un elemento perso e da che cosa può fare il produttore.

• Attendere (block): il produttore si ferma finché c'è posto. Nessuna perdita, ma il ritardo si propaga a monte. Per pipeline in cui ogni elemento conta.
• Scartare il nuovo (drop newest): l'elemento in arrivo viene rifiutato. Il produttore non rallenta. Per metriche, log non critici, eventi di interfaccia.
• Scartare il più vecchio (drop oldest): si fa posto togliendo il vecchio. Per dati in cui conta l'ultimo valore, come un prezzo o una posizione.
• Fallire (error): si segnala un errore subito. Per operazioni in cui non è accettabile né perdere né aspettare, e il chiamante deve decidere.

Le prime due rispondono alla domanda «posso perdere dati?», le altre due a «quale dato perdo?». Il criterio pratico è partire dal valore dell'elemento: se è un ordine, si attende o si fallisce; se è una misura che sarà sostituita tra un secondo, si scarta. Scartare in silenzio è l'errore comune: ogni scarto va contato e reso visibile con una metrica, altrimenti il sistema «funziona» mentre perde dati.

Reactive Streams: la domanda viaggia a monte

Con una coda bloccante, il consumatore non dice mai quanto vuole: si limita a svuotare. Le specifiche Reactive Streams, nate per lo «stream processing asincrono con contropressione non bloccante», invertono la direzione: il Subscriber chiede request(n) e il Publisher non può emettere più di quanto richiesto. Dal JDK 9 le interfacce sono nella libreria standard, in java.util.concurrent.Flow, in corrispondenza uno a uno con quelle della specifica. Il listato usa SubmissionPublisher con un buffer da 16 posti e un sottoscrittore lento, che ci mette 10 ms per elemento e ne chiede uno alla volta.

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

// java.util.concurrent.Flow e SubmissionPublisher: Reactive Streams nella JDK, senza librerie.
public class Flusso {
    // Un Subscriber che chiede UN elemento alla volta e ci mette 10 ms a smaltirlo.
    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);                       // "ho finito, mandane un altro"
        }
        public void onError(Throwable t) { t.printStackTrace(); }
        public void onComplete() { finito.countDown(); }
    }

    public static void main(String[] args) throws Exception {
        // 1. submit(): quando il buffer (16 posti) e' pieno, il produttore SI FERMA
        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(): il produttore ha finito dopo " + tProduttore + " ms");
        }
        a.finito.await();
        System.out.println("submit(): ricevuti " + a.ricevuti.get() + " su 100, totale " + (System.nanoTime() - t0) / 1_000_000 + " ms");

        // 2. offer(): non blocca, il rifiuto e' una decisione del produttore (qui: scarta e conta)
        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 = non riprovare
            }
            System.out.println("offer(): il produttore ha finito dopo " + (System.nanoTime() - t0) / 1_000_000 + " ms");
        }
        b.finito.await();
        System.out.println("offer(): ricevuti " + b.ricevuti.get() + ", scartati " + scartati.get()
            + ", somma " + (b.ricevuti.get() + scartati.get()));
    }
}

// Output:
// submit(): il produttore ha finito dopo 1015 ms
// submit(): ricevuti 100 su 100, totale 1066 ms
// offer(): il produttore ha finito dopo 10 ms
// offer(): ricevuti 16, scartati 84, somma 100

Il Javadoc di SubmissionPublisher descrive i due comportamenti: submit blocca finché ci sono risorse (la politica più semplice, la meno reattiva), mentre i metodi offer possono scartare e offrono un punto per intercettare lo scarto e riprovare. Nella prima prova il produttore ha impiegato 1.015 ms per inserire 100 elementi, perché ha dovuto aspettare lo smaltimento, e il sottoscrittore ha ricevuto tutti i 100. Nella seconda ha finito in 10 ms ma solo 16 elementi sono stati ricevuti: 84 sono stati scartati e contati, e la somma torna 100. Le due prove sono l'esempio più pulito della scelta tra «attendere» e «scartare il nuovo».

Reactor: dichiarare la strategia in una riga

Le librerie come Project Reactor mettono le strategie dentro la pipeline. La guida di Reactor descrive il meccanismo reale: il sottoscrittore può lavorare in modalità illimitata, lasciando che la sorgente spinga alla massima velocità, oppure usare request per dire alla sorgente che è pronto a gestire al più n elementi. Il listato prova tre cose: la richiesta a lotti da 10, un buffer che scarta il più vecchio e un 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: il consumatore chiede 10 elementi alla volta; l'editore non ne emette di piu'.
        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);          // ne ho smaltiti 10: chiedo i prossimi 10
                }
            });
        System.out.println("request(n) ricevuti dall'editore: " + richieste.toString().trim());

        // 2. Sorgente veloce, consumatore lento, buffer di 100 che SCARTA IL PIU' VECCHIO
        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)           // il consumatore chiede 1 alla volta
            .doOnNext(v -> { try { Thread.sleep(1); } catch (InterruptedException e) { } })
            .doOnNext(ultimo::set)
            .blockLast();
        // consumati = tutti quelli che hanno attraversato il consumatore
        System.out.println("DROP_OLDEST: scartati " + scartati.get() + " su 5000; ultimo elemento ricevuto: " + ultimo.get());

        // 3. Stessa sorgente con onBackpressureDrop: scarta i NUOVI quando il consumatore non ha richiesto
        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: arrivati " + arrivati.get() + ", scartati " + scartatiDrop.get()
            + ", somma " + (arrivati.get() + scartatiDrop.get()));
    }
}

// Output:
// request(n) ricevuti dall'editore: 10 10 10 10
// DROP_OLDEST: scartati 4895 su 5000; ultimo elemento ricevuto: 5000
// onBackpressureDrop: arrivati 2, scartati 4998, somma 5000

Nella prima parte, l'editore riceve quattro richieste da 10 per trenta elementi: la prima all'iscrizione e le altre tre quando il sottoscrittore ne ha smaltiti 10. Nella seconda, un buffer da 100 che scarta il più vecchio ha lasciato cadere 4.895 elementi su 5.000, ma ha consegnato per ultimo il 5.000: è la strategia per cui «il dato più recente vale di più». Nella terza, onBackpressureDrop ha lasciato passare solo 2 elementi: la sorgente è sincrona, ne emette 5.000 in pochi microsecondi, e il consumatore (che chiede un elemento alla volta) ne ha richiesti pochissimi in quel lasso di tempo. Il numero esatto dipende dallo scheduling, ma la somma (arrivati + scartati) è sempre 5.000: ogni elemento è contato.

Python: code limitate e generatori asincroni

In asyncio la risposta «attendere» è già nella coda. La documentazione di asyncio.Queue dice che, con maxsize maggiore di zero, await put() blocca quando la coda raggiunge maxsize, e che con zero la coda è infinita. Il listato produce mille elementi e ne consuma uno ogni 2 ms, prima con coda illimitata e poi con maxsize=10.

import asyncio
import time

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

    async def produttore():
        nonlocal produttore_finito
        for i in range(1000):
            await coda.put(i)            # con maxsize > 0 si sospende quando la coda e' piena
        await coda.put(-1)               # sentinella
        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)   # elaborazione lenta

    async with asyncio.TaskGroup() as tg:
        tg.create_task(produttore())
        tg.create_task(consumatore())
    etichetta = "illimitata" if maxsize == 0 else f"maxsize={maxsize}"
    print(f"{etichetta:<14} coda al picco: {picco:4d}, produttore finito dopo {produttore_finito:.1f} s, "
          f"tutto finito dopo {time.monotonic() - t0:.1f} s")

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

asyncio.run(main())

# Output:
# illimitata     coda al picco: 1001, produttore finito dopo 0.0 s, tutto finito dopo 2.3 s
# maxsize=10     coda al picco:   10, produttore finito dopo 2.3 s, tutto finito dopo 2.3 s

Con la coda illimitata il produttore ha finito praticamente subito (0,0 s) e la coda ha raggiunto un picco di 1.001 elementi: tutto il lavoro è in memoria prima di essere fatto. Con maxsize=10 il picco è 10 e il produttore ha finito dopo 2,3 s, insieme al consumatore: ha camminato al passo di chi consuma. Il tempo totale è lo stesso nei due casi (2,3 s), a conferma che la coda limitata non rallenta il sistema, rallenta solo chi produce. Il sentinella -1 chiude il consumatore, e il TaskGroup (vedi asyncio avanzato) attende entrambi.

Il secondo strumento è il generatore asincrono: la backpressure è implicita, perché il generatore si sospende a ogni yield e riprende solo quando il consumatore chiede il valore successivo con async for. Il listato stampa i due lati per far vedere il passo alternato.

import asyncio

async def eventi():
    for i in range(4):
        print(f"  generatore: produco {i}")
        yield i                          # qui si sospende finche' il consumatore chiede il prossimo

async def main():
    async for evento in eventi():
        print(f"consumatore: ricevo {evento}, lo elaboro")
        await asyncio.sleep(0.01)

asyncio.run(main())

# Output:
#   generatore: produco 0
# consumatore: ricevo 0, lo elaboro
#   generatore: produco 1
# consumatore: ricevo 1, lo elaboro
#   generatore: produco 2
# consumatore: ricevo 2, lo elaboro
#   generatore: produco 3
# consumatore: ricevo 3, lo elaboro

Le righe si alternano esattamente: il generatore produce un valore, il consumatore lo elabora (compreso il sleep di 10 ms), e solo allora il generatore produce il successivo. Il modello è «pull»: non esiste una coda, quindi non c'è nulla da dimensionare. Il limite è che un generatore serve un solo consumatore alla volta e non sovrappone produzione ed elaborazione; per farlo serve di nuovo una coda limitata in mezzo.

Go e Rust: lo stesso principio, senza esempi qui

In Go un canale con buffer fa la stessa cosa della coda limitata: la specifica dice che l'invio attende quando il buffer è pieno, e il buffer determina la latenza massima come nel primo esperimento. In Tokio, la documentazione di mpsc::Sender::send dice «invia un valore, attendendo che ci sia capacità»; un'alternativa che non attende, try_send, restituisce un errore se il canale è pieno, ed è il modo di implementare «scartare il nuovo» o «fallire». Non riporto qui listati in Go o Rust, perché sulla macchina non ci sono i rispettivi compilatori e non sarebbero stati eseguiti.

Che cosa sai fare adesso

Sai calcolare quanto crescerebbe una coda senza limite e perché la sua dimensione coincide con la latenza massima, scegliere tra le quattro risposte in base al valore dell'elemento e contare ciò che scarti. Sai usare una coda limitata, un SubmissionPublisher o un generatore asincrono secondo il caso, e hai visto in numeri che una coda limitata non toglie capacità, la rende visibile.

Il prossimo e ultimo articolo porta gli stessi problemi sulla scala distribuita: CAP theorem, consistenza e CRDT. Per chi vuole prima il caso opposto, un lock che blocca tutti, vedi race condition e deadlock. Il modo in cui un event loop gestisce una coda di eventi che si riempie è nell'articolo sull'event loop JavaScript.

I conteggi (51.960, 2.564, 4.895, 1.001) e i tempi (1.015 ms, 2,3 s) provengono da esecuzioni singole su una macchina; l'ordine di grandezza e la somma «ricevuti + scartati» sono stabili, i valori puntuali no. Le affermazioni su Go e Tokio sono tratte dalle fonti sopra, non da codice eseguito.