04 - Async/Await Python Avanzato: asyncio Internals, TaskGroup e Cancellation Robusta

Il problema: un task fallisce e gli altri?

Hai un servizio che, per costruire una pagina, interroga tre sorgenti insieme. Una risponde con un errore. Che cosa succede alle altre due? Con asyncio.gather, la risposta di default è sorprendente: continuano a girare, senza che nessuno aspetti il loro risultato. Con asyncio.TaskGroup vengono cancellate e l'errore ti arriva impacchettato in un ExceptionGroup. Questo articolo spiega la differenza, i suoi costi, e come scrivere coroutine che si lasciano cancellare in modo pulito.

Tutti i listati sono stati eseguiti con Python 3.14.4 e usano solo la libreria standard: dove l'originale chiamava una rete vera, c'è una asyncio.sleep al suo posto, che sospende la coroutine allo stesso modo. TaskGroup, asyncio.timeout e except* sono arrivati in Python 3.11, quindi il codice richiede almeno quella versione. Per il confronto con il modello a event loop di JavaScript, vedi l'articolo sull'event loop.

Come lavora asyncio: un thread, molte coroutine

asyncio ha un solo event loop che gira su un thread. Una coroutine è una funzione async def che può sospendersi a ogni await: in quel punto cede il controllo al loop, che ne fa partire un'altra. Quando l'operazione attesa è pronta (un socket leggibile, un timer scaduto) il loop riprende la coroutine da dove si era fermata. Il lavoro "contemporaneo" è quindi un'illusione ben fatta: si sovrappongono le attese, non i calcoli.

import asyncio
import time

async def fetch(nome: str) -> str:
    await asyncio.sleep(1)          # sospende la coroutine, non il thread
    return f"dati di {nome}"

async def main():
    t0 = time.monotonic()
    risultati = await asyncio.gather(fetch("a"), fetch("b"), fetch("c"))
    print(risultati)
    print(f"tempo totale: {time.monotonic() - t0:.1f} s")   # ~1 s, non 3 s

asyncio.run(main())

# Output:
# ['dati di a', 'dati di b', 'dati di c']
# tempo totale: 1.0 s

Tre coroutine da un secondo ciascuna finiscono in un secondo e non in tre, perché mentre una dorme le altre due sono già partite. asyncio.gather avvolge ciascuna coroutine in un task, cioè la schedula sul loop, e restituisce i risultati nell'ordine in cui le hai passate. Il tempo misurato (1,0 s) vale per questa macchina e per attese di rete simulate: con un calcolo CPU al posto di sleep le tre coroutine si eseguirebbero una dopo l'altra, perché nessuna cederebbe mai il thread.

gather: l'errore che lascia il lavoro in giro

La documentazione di gather è esplicita: con return_exceptions=False, il valore predefinito, la prima eccezione viene propagata subito a chi attende, e gli altri awaitable non vengono cancellati e continuano a girare. Il listato lo rende visibile: tre task, il secondo fallisce a 0,2 s, il terzo avrebbe finito a 0,3 s.

import asyncio

async def lavoro(n: int) -> int:
    try:
        await asyncio.sleep(n * 0.1)
        if n == 2:
            raise ValueError(f"task {n} fallito")
        print(f"task {n}: completato")
        return n * 10
    except asyncio.CancelledError:
        print(f"task {n}: cancellato")
        raise

async def main():
    try:
        await asyncio.gather(lavoro(1), lavoro(2), lavoro(3))
    except ValueError as e:
        print("gather ha sollevato:", e)
    await asyncio.sleep(0.5)        # diamo tempo al task 3 di mostrarsi
    print("fine main")

asyncio.run(main())

# Output:
# task 1: completato
# gather ha sollevato: task 2 fallito
# task 3: completato
# fine main

L'eccezione arriva a main prima che il task 3 abbia finito, ma la riga «task 3: completato» compare comunque, dopo: il task ha lavorato per un risultato che nessuno userà. Il try/except dentro lavoro serve a stampare l'eventuale cancellazione; qui non scatta mai. In un servizio reale quel task potrebbe essere una scrittura su database o una chiamata a pagamento, e ogni richiesta fallita ne lascerebbe uno vivo.

Il rimedio parziale è return_exceptions=True: le eccezioni diventano elementi della lista dei risultati e gather aspetta tutti.

import asyncio

async def lavoro(n: int) -> int:
    await asyncio.sleep(n * 0.1)
    if n == 2:
        raise ValueError(f"task {n} fallito")
    return n * 10

async def main():
    risultati = await asyncio.gather(
        lavoro(1), lavoro(2), lavoro(3), return_exceptions=True
    )
    print(risultati)
    for r in risultati:
        print("errore:" if isinstance(r, BaseException) else "ok:", r)

asyncio.run(main())

# Output:
# [10, ValueError('task 2 fallito'), 30]
# ok: 10
# errore: task 2 fallito
# ok: 30

Nessun task resta orfano, ma il prezzo è che l'errore non interrompe più nulla: devi scorrere la lista e distinguere i risultati dalle eccezioni con isinstance, e se dimentichi di farlo l'errore scompare in silenzio. È una scelta legittima quando vuoi tutti i risultati possibili (un aggregatore di prezzi che mostra le fonti raggiunte), sbagliata quando un fallimento rende inutile il resto.

TaskGroup: il blocco che non lascia nessuno indietro

asyncio.TaskGroup, nuovo in Python 3.11, applica la structured concurrency: i task nascono dentro un blocco async with e non possono sopravvivergli. La documentazione dice che alla prima eccezione diversa da CancelledError i task rimanenti del gruppo vengono cancellati, e che il blocco esce solo quando tutti sono terminati. Gli errori escono dal blocco raggruppati in un ExceptionGroup, da cui il nuovo except*.

import asyncio

async def lavoro(n: int) -> int:
    try:
        await asyncio.sleep(n * 0.1)
        if n == 2:
            raise ValueError(f"task {n} fallito")
        print(f"task {n}: completato")
        return n * 10
    except asyncio.CancelledError:
        print(f"task {n}: cancellato")
        raise

async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            t1 = tg.create_task(lavoro(1))
            tg.create_task(lavoro(2))
            tg.create_task(lavoro(3))
    except* ValueError as eg:
        print("gruppo:", type(eg).__name__, [str(e) for e in eg.exceptions])
    print("t1.result() =", t1.result())
    await asyncio.sleep(0.5)
    print("fine main: nessun task orfano")

asyncio.run(main())

# Output:
# task 1: completato
# task 3: cancellato
# gruppo: ExceptionGroup ['task 2 fallito']
# t1.result() = 10
# fine main: nessun task orfano

Stesso scenario di prima, esito opposto. Il task 1 finisce regolarmente e ne conservi il risultato; il task 2 fallisce; il task 3 riceve una CancelledError (la riga «cancellato») invece di continuare. Dopo il blocco, t1.result() vale 10 e l'attesa aggiuntiva di mezzo secondo non mostra alcun task residuo. Nota che l'errore è un ExceptionGroup anche se ne è fallito uno solo: per la prima volta il codice che lo cattura deve scriversi except*.

ExceptionGroup e except*: più errori in uno

Se due task falliscono nello stesso giro del loop, un'eccezione sola non basta a raccontare cosa è successo. Il PEP 654 ha introdotto ExceptionGroup per raccogliere errori non correlati, e except* per trattarli per tipo: ogni clausola prende i sottogruppi del tipo indicato e lascia passare il resto.

import asyncio

async def solleva(tipo, msg):
    await asyncio.sleep(0.1)
    raise tipo(msg)

async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(solleva(ValueError, "A"))
            tg.create_task(solleva(ValueError, "B"))
            tg.create_task(solleva(TypeError, "C"))
    except* ValueError as eg:
        print("ValueError:", sorted(str(e) for e in eg.exceptions))
    except* TypeError as eg:
        print("TypeError:", sorted(str(e) for e in eg.exceptions))

asyncio.run(main())

# Un gruppo annidato: eg.exceptions contiene a sua volta un gruppo.
try:
    raise ExceptionGroup("esterno", [
        ValueError("v1"),
        ExceptionGroup("interno", [KeyError("k1"), KeyError("k2")]),
    ])
except* KeyError as eg:
    print("exceptions:", eg.exceptions)
    print("derive:", eg.split(KeyError)[0].exceptions[0].exceptions)
except* ValueError as eg:
    print("ValueError:", eg.exceptions)

# Output:
# ValueError: ['A', 'B']
# TypeError: ['C']
# exceptions: (ExceptionGroup('interno', [KeyError('k1'), KeyError('k2')]),)
# derive: (KeyError('k1'), KeyError('k2'))
# ValueError: (ValueError('v1'),)

Nella prima parte, le due ValueError finiscono nella clausola except* ValueError e la TypeError nell'altra; entrambe le clausole girano, a differenza del normale except che si ferma alla prima corrispondenza. La seconda parte corregge un'imprecisione facile da commettere: con gruppi annidati, eg.exceptions non è la lista piatta delle KeyError, ma contiene ancora il gruppo interno. Per arrivare alle singole eccezioni serve scendere un livello o usare split, che restituisce un gruppo con la stessa struttura ma solo le eccezioni del tipo cercato.

Cancellazione: cooperativa, e facile da rompere

Task.cancel() non uccide nulla: chiede al task di ricevere una asyncio.CancelledError al prossimo await. La documentazione raccomanda di usare try/finally per la pulizia e, se la CancelledError viene intercettata, di rilanciarla a pulizia finita. Il motivo è strutturale: TaskGroup e asyncio.timeout sono costruiti sulla cancellazione e, parole della documentazione, possono funzionare male se una coroutine la inghiotte. Il listato confronta le due versioni.

import asyncio

async def corretto():
    try:
        await asyncio.sleep(10)
    except asyncio.CancelledError:
        print("corretto: pulizia, poi rilancio")
        raise

async def inghiotte(fermati: asyncio.Event):
    while not fermati.is_set():
        try:
            await asyncio.sleep(10)
        except asyncio.CancelledError:
            print("inghiotte: cancellazione ignorata")   # manca il raise

async def main():
    t = asyncio.create_task(corretto())
    await asyncio.sleep(0.1)
    t.cancel()
    await asyncio.wait([t])
    print("corretto: cancelled() =", t.cancelled())

    fermati = asyncio.Event()
    t = asyncio.create_task(inghiotte(fermati))
    await asyncio.sleep(0.1)
    t.cancel()
    await asyncio.wait([t], timeout=1)
    print("inghiotte: dopo 1 s done() =", t.done(), "cancelled() =", t.cancelled())
    fermati.set()                  # solo per chiudere l'esempio
    t.cancel()
    await asyncio.wait([t])
    print("inghiotte: ora done() =", t.done(), "cancelled() =", t.cancelled())

asyncio.run(main())

# Output:
# corretto: pulizia, poi rilancio
# corretto: cancelled() = True
# inghiotte: cancellazione ignorata
# inghiotte: dopo 1 s done() = False cancelled() = False
# inghiotte: cancellazione ignorata
# inghiotte: ora done() = True cancelled() = False

Il task corretto pulisce, rilancia, e risulta cancellato. Il task inghiotte stampa «cancellazione ignorata» e prosegue: dopo un secondo done() è ancora falso. Qui un ciclo while rende il problema evidente; ho dovuto usare un Event per far terminare l'esempio, e in un programma vero il task avrebbe tenuto viva la richiesta per sempre. In un TaskGroup vorrebbe dire che il blocco non esce mai. Nota anche il corollario: CancelledError è una sottoclasse di BaseException, quindi un banale except Exception non la intercetta, e va catturata solo di proposito.

Timeout: asyncio.timeout e wait_for

Il modo normale di dare una scadenza è asyncio.timeout (Python 3.11): un context manager che cancella il codice al suo interno allo scadere e lo trasforma in un TimeoutError al di fuori del blocco. Il più vecchio wait_for avvolge una singola awaitable e fa lo stesso. La differenza pratica è che il context manager copre più istruzioni e si combina con altri blocchi.

import asyncio

async def lenta():
    await asyncio.sleep(5)

async def main():
    try:
        async with asyncio.timeout(0.2):
            await lenta()
    except TimeoutError:
        print("asyncio.timeout: TimeoutError")
    try:
        await asyncio.wait_for(lenta(), timeout=0.2)
    except asyncio.TimeoutError as e:
        print("wait_for: asyncio.TimeoutError e' TimeoutError:", asyncio.TimeoutError is TimeoutError)

asyncio.run(main())

# Output:
# asyncio.timeout: TimeoutError
# wait_for: asyncio.TimeoutError e' TimeoutError: True

Entrambe le scadenze producono TimeoutError. Il secondo controllo del listato mostra che da Python 3.11 asyncio.TimeoutError è un alias dell'eccezione predefinita (la documentazione lo definisce «deprecated alias»): ho verificato l'identità con is, quindi un except TimeoutError basta ovunque. Il codice scritto per 3.10 o precedenti, dove i due nomi erano classi diverse, continua a funzionare. La scadenza, ricorda, funziona solo se la coroutine sotto si lascia cancellare: vale quanto detto sopra.

Limitare la concorrenza: semaforo e coda

Un TaskGroup non pone limiti: se gli passi diecimila URL, parte una richiesta per ciascuno, con il rischio di esaurire socket o far scattare il rate limit del server. Il limite si ottiene con un asyncio.Semaphore(N): al massimo N coroutine possono stare nel blocco async with sem, le altre aspettano il proprio turno. Per renderlo verificabile ho contato il picco di richieste simultanee su cento URL con limite 10.

import asyncio

attivi = 0
picco = 0

async def scarica(url: str, sem: asyncio.Semaphore) -> dict:
    global attivi, picco
    async with sem:                       # al massimo N dentro questo blocco
        attivi += 1
        picco = max(picco, attivi)
        await asyncio.sleep(0.05)         # al posto della richiesta di rete
        attivi -= 1
        return {"url": url, "status": 200}

async def main():
    sem = asyncio.Semaphore(10)
    urls = [f"https://example.test/{i}" for i in range(100)]
    async with asyncio.TaskGroup() as tg:
        task = [tg.create_task(scarica(u, sem)) for u in urls]
    print(len(task), "risposte, picco di richieste contemporanee:", picco)

asyncio.run(main())

# Output:
# 100 risposte, picco di richieste contemporanee: 10

Cento task vengono creati subito, ma il picco di quelli "dentro" resta 10: il contatore è incrementato e decrementato attorno al punto in cui una richiesta reale sarebbe in volo. Il vantaggio rispetto a un pool fisso di worker è la semplicità; il limite è che tutti i cento task esistono in memoria dall'inizio, quindi per milioni di elementi conviene la strada alternativa, una coda limitata con un numero fisso di consumatori.

import asyncio

async def main():
    coda: asyncio.Queue = asyncio.Queue(maxsize=5)   # limitata: il produttore rallenta
    risultati = []
    massimo = 0

    async def produttore():
        for i in range(30):
            await coda.put(i)             # attende se ci sono gia' 5 elementi
        await coda.put(None)              # sentinella: non arriva altro

    async def consumatore():
        nonlocal massimo
        while (voce := await coda.get()) is not None:
            massimo = max(massimo, coda.qsize())
            await asyncio.sleep(0.01)     # elaborazione lenta
            risultati.append(voce * 2)

    async with asyncio.TaskGroup() as tg:
        tg.create_task(produttore())
        tg.create_task(consumatore())

    print(len(risultati), "elaborati, coda mai oltre", massimo)

asyncio.run(main())

# Output:
# 30 elaborati, coda mai oltre 4

La Queue(maxsize=5) è il freno: quando contiene cinque elementi, await coda.put(i) sospende il produttore finché il consumatore non ne prende uno. La coda non ha mai superato quattro elementi nell'osservazione del consumatore, e tutti i trenta valori sono arrivati in ordine. Il None finale è una sentinella: senza, il consumatore resterebbe in attesa per sempre e il TaskGroup non uscirebbe. Il tema del produttore più veloce del consumatore è trattato a parte in backpressure e flow control.

gather o TaskGroup: come scegliere

Nel codice nuovo la scelta predefinita è TaskGroup, ma gather ha ancora un posto. Il criterio è che cosa vuoi che succeda al fallimento di un task.

• TaskGroup: i task sono parti di un'unica operazione e uno che fallisce rende inutile il resto; vuoi nessun orfano e task individuabili per nome.
• gather con return_exceptions=True: i task sono indipendenti e vuoi i risultati parziali, accettando di controllare ogni elemento.
• gather di default: quasi mai; salvo che tu sappia che, al primo errore, l'intero programma termina comunque.

La terza voce è la più importante: il comportamento predefinito di gather è quello che produce i task orfani visti prima. gather non è «sbagliato», ma è nato prima della structured concurrency e non la garantisce. Se la tua versione minima di Python è precedente alla 3.11, dovrai restare su gather e gestire la cancellazione a mano.

Debug: dare un nome ai task

Quando un servizio ha centinaia di task, un traceback che dice «Task-37» non aiuta. Il parametro name di create_task lo risolve, e asyncio.all_tasks() elenca ciò che è vivo.

import asyncio

async def lavoro():
    await asyncio.sleep(1)

async def main():
    async with asyncio.TaskGroup() as tg:
        tg.create_task(lavoro(), name="scrivi-db")
        tg.create_task(lavoro(), name="aggiorna-cache")
        await asyncio.sleep(0.1)
        for t in sorted(asyncio.all_tasks(), key=lambda t: t.get_name()):
            print(t.get_name(), "done =", t.done())

asyncio.run(main())

# Output:
# Task-1 done = False
# aggiorna-cache done = False
# scrivi-db done = False

I tre task (due creati con nome e quello che esegue main, che Python chiama Task-1) sono tutti non terminati al momento del controllo. L'ordine è quello che ho imposto con sorted: all_tasks() restituisce un insieme, senza ordine garantito. Per cercare coroutine mai attese, la modalità debug (asyncio.run(main(), debug=True) o la variabile PYTHONASYNCIODEBUG=1) registra, secondo la documentazione, le callback che durano più di 100 ms e rende più dettagliato l'avviso sulle coroutine mai attese; non l'ho eseguita qui.

Un'ultima avvertenza dalla documentazione: l'event loop tiene solo riferimenti deboli ai task. Un create_task il cui risultato non è conservato può essere raccolto dal garbage collector a metà esecuzione. Dentro un TaskGroup il problema non esiste perché il gruppo conserva i riferimenti; fuori, tieni i task in un insieme.

Che cosa sai fare adesso

Sai perché gather lascia task orfani e come TaskGroup lo evita, sai leggere e filtrare un ExceptionGroup con except* senza cadere nella trappola dei gruppi annidati, e sai scrivere una coroutine che si lascia cancellare. Sai anche limitare la concorrenza con un semaforo o con una coda limitata, e dare un nome ai task per ritrovarli nei log.

Il prossimo articolo guarda lo stesso problema dalla parte della JVM: virtual thread e structured concurrency in Java, dove invece di coroutine usi codice bloccante su thread molto leggeri. Per il quadro in cui asyncio si colloca accanto agli altri modelli, vedi il confronto fra i modelli di concorrenza. Lo stesso problema di task che falliscono e di cancellazione, in Rust, è nell'articolo su async di Rust e Tokio.

I tempi (1,0 s) e i conteggi (picco 10, coda massima 4) provengono da esecuzioni singole con Python 3.14.4 su una sola macchina; i valori esatti cambiano da un'esecuzione all'altra.