Programmazione asincrona

1. Eventi invece di thread: l'event loop

Un sistema reattivo deve rispondere prontamente a stimoli che arrivano in momenti imprevedibili: click, richieste di rete, letture di sensori, risposte a richieste precedenti. Storicamente le risposte sono state due, i thread e gli eventi. Il panorama asincrono di oggi ha quattro famiglie, che il capitolo attraversa nell'ordine:

È un campo ancora in evoluzione: Microsoft è passata dall'Asynchronous Programming Model (APM) all'Event-based Asynchronous Pattern (EAP, .NET 2.0) e poi al Task-based Asynchronous Pattern (TAP, .NET 4.5), deprecando ogni volta il precedente.

Programmi fatti di handler

Un programma event-driven è una collezione di routine, gli handler, eseguite quando si verifica l'evento a cui sono associate. Il flusso di controllo lo decidono gli eventi, non il programmatore, quindi l'esecuzione è non deterministica. Un sistema a eventi deve stabilire come si associano gli handler agli eventi, come si ricevono e si gestiscono gli eventi e come si eseguono gli handler. Il modello di esecuzione di riferimento è l'event loop: un solo flusso di controllo e una coda di eventi, prodotti dall'ambiente o dagli handler stessi.

loop {
  Event ev = waitForEvent(eventQueue)     // l'unico punto in cui il loop si blocca
  Handler handler = selectHandler(ev)
  execute(handler)                        // fino in fondo, senza interruzioni
}

Gli handler sono eseguiti atomicamente: un evento che arriva durante un handler resta in coda finché l'handler non è terminato. Alla fine di ogni handler lo stack è vuoto, senza chiamate pendenti: per questo si parla di «programmazione senza call stack». La coda può avere priorità, ma gli eventi si servono comunque uno alla volta.

Due handler possono essere eseguiti insieme? Non nello stesso event loop: li esegue un solo thread, uno dopo l'altro. Per questo, a questo livello, non ci sono race condition e non servono lock: lo stato lo tocca solo l'event loop. Con più consumatori sulla stessa coda gli handler potrebbero interferire, quindi il loop di una coda è uno solo. Un sistema può però avere più event loop, uno per componente attivo, ciascuno con coda e stato propri: i loro handler girano in parallelo, non condividono memoria e comunicano generando eventi (l'event bus di Vert.x, sezione 9). Da qui nascono gli attori.

Il pattern reactor

Nella letteratura sui pattern (Schmidt, POSA vol. 2, 2000) l'event loop è il reactor, che separa il demultiplexing e il dispatching degli eventi dalla logica applicativa. I partecipanti sono cinque:

• Handle: identifica una sorgente di eventi gestita dal sistema operativo, ad esempio una socket;
• Synchronous Event Demultiplexer: attende, bloccandosi, che uno degli handle diventi «pronto» (come select());
• Initiation Dispatcher: registra gli handler e, quando un handle è pronto, lo usa come chiave per trovare l'handler e chiamarne il metodo hook;
• Event Handler: l'interfaccia con il metodo hook (handle_event);
• Concrete Event Handler: l'implementazione con la logica applicativa.

È l'architettura che sta sotto JavaScript nei browser, Node.js, Vert.x e le GUI.

2. La regola del never-blocking

Dal modello discende la regola su cui si regge tutto: un handler non deve mai bloccarsi e deve sempre terminare. Una chiamata bloccante o un ciclo infinito imprigionano l'event loop, gli eventi in coda non vengono più serviti e il sistema smette di essere reattivo. Anche un handler solo lungo ritarda tutti gli altri eventi.

Una chiamata bloccante va sostituita da una richiesta asincrona: l'handler affida il lavoro e prosegue. Il lavoro lo fanno thread in background, che il programmatore di solito non vede. Quando hanno finito, non toccano lo stato dell'event loop: accodano un evento con il risultato o con l'errore. In un web server a eventi le richieste sono coordinate in sequenza da un solo thread, ma l'I/O procede concorrentemente: è uno schema master-worker in cui l'event loop fa il master. Un handler, quindi:

• è eseguito dall'event loop quando arriva l'evento a cui è associato, anche registrato dinamicamente;
• va fino in fondo senza interruzioni (atomicità) e deve terminare presto;
• non contiene chiamate bloccanti né attese attive: ogni operazione lenta diventa asincrona;
• può generare nuovi eventi, che il loop vedrà solo dopo che l'handler è finito;
• accede solo allo stato del proprio event loop.

Un esempio classico di errore in JavaScript è aspettare una risposta HTTP con un ciclo:

const req = new XMLHttpRequest();
req.open('GET', url);
req.send();
while (req.readyState !== XMLHttpRequest.DONE) {}   // SBAGLIATO: non esce mai

Lo stato di req cambia solo quando l'event loop elabora l'evento di risposta, e l'event loop è bloccato proprio in quel ciclo. Si registra invece un handler, req.onload = () => { … }, anche dopo send(): la risposta non verrà elaborata prima che l'handler corrente sia finito.

3. Callback e continuation-passing style

Se un handler non può aspettare un risultato, come lo elabora? Passando alla funzione asincrona, come ultimo argomento, una callback da eseguire quando il risultato sarà pronto: un handler registrato dinamicamente per l'evento «l'operazione è finita». Gli eventi spariscono dal codice, ma sotto ci sono sempre.

// sincrono (bloccante)               // asincrono
r = syncTask(p);                      asyncTask(p, (r) => {
s = r + 1;                              s = r + 1;          // eseguita quando r è pronto
                                      });

Il CPS in generale

Questo stile è il continuation-passing style (CPS), introdotto da Sussman e Steele nel 1975 per Scheme. Non nasce per la concorrenza: è un modo di scrivere qualsiasi programma passando il controllo in avanti, in forma esplicita, come continuazione.

int sum(int a, int b) { return a + b; }              // stile diretto
System.out.println(sum(1, 3));

void sum(int a, int b, Consumer<Integer> k) {         // CPS
  k.accept(a + b);                                    // passa il risultato in avanti
}
sum(1, 3, res -> System.out.println(res));           // la continuazione: "poi stampalo"

Le callback sono di solito closure: una funzione insieme al suo ambiente, cioè ai binding delle variabili libere nel momento in cui è stata creata. Così la callback, eseguita più tardi, vede ancora le variabili del contesto che l'ha creata. È lo stesso meccanismo della domanda su var e let (sezione 6).

Chi esegue la callback?

Potrebbe eseguirla il thread che ha svolto il lavoro, appena ha il risultato: sembra più reattivo, ma la callback girerebbe in concorrenza con l'event loop, e tornerebbero le race. Oppure la esegue l'event loop, il flusso di controllo che ha fatto la richiesta, e il worker genera solo l'evento. È il modello delle web app moderne, lato client (JavaScript, Dart) e lato server (Node.js, Vert.x).

x = 0                          // L1
asynctask(x, (res) => {
  x = x + res                  // L2
});
x = x + 1                      // L3

Laboratorio: un event loop passo per passo

Fai avanzare l'event loop un passo alla volta e guarda stack, code e lavoro in background. Nello scenario «Callback» decidi tu quando finisce il worker: prova prima di L3, oppure mai. Gli altri scenari riprendono l'esempio con promise e async/await, e le domande della sezione 6.

Vantaggi e flussi di controllo

In sintesi, un solo thread coordina molte operazioni eseguite da thread esterni, con poca memoria rispetto a un thread (e uno stack) per richiesta. Non ci sono race a basso livello, perché lo stato lo tocca un solo thread, né deadlock a basso livello, perché gli handler non si bloccano. Sequenza e parallelo si scrivono così:

// sequenza: task2 parte quando task1 è finito
asyncTask1(a, (res1) => {
  asyncTask2(res1, (res2) => { … });
});

// parallelo: partono entrambi nello stesso ciclo
asyncTask1(a, (res1) => { … });
asyncTask2(b, (res2) => { … });

Nel caso parallelo l'ordine delle due callback non è determinato: dipende da quale task finisce prima, cioè dall'ordine in cui arrivano gli eventi.

4. Il prezzo delle callback: callback hell

Il CPS ha due problemi noti, riassunti come callback hell:

• Spaghetti asincrono: il calcolo si frammenta in tanti handler, e la struttura e la modularità del programma ne soffrono.
• Pyramid of doom: comporre in sequenza operazioni asincrone significa annidare callback. Il codice cresce verso destra invece che verso il basso ed è difficile da leggere, riusare ed estendere. Nel web server Node.js delle slide, per esempio, path.exists ha una callback che chiama fs.readFile, che ha una sua callback, e così via.

step1((result1) => {
  step2(result1, (result2) => {
    step3(result2, (result3) => {
      // e così via…
    });
  });
});

«I love async, but I can't code like this», scriveva uno sviluppatore sul gruppo di Node.js. Le risposte, promise e poi async/await, cambiano il modo di scrivere, non il motore.

5. Promise

Una promise (Friedman e Wise, 1976) rappresenta il risultato, non ancora disponibile, di una singola operazione asincrona. La funzione asincrona non riceve più la callback, che le toglieva la firma naturale: restituisce subito una promise, e il chiamante la configura con then.

Una promise è in uno di tre stati: pending alla creazione, fulfilled (o resolved) se l'operazione termina bene, rejected se termina con un errore. Quando non è più pending si dice settled: una promise si risolve o si rigetta una sola volta, e da quel momento stato e valore non cambiano più.

function delayWithRandom(t, p) {            // funzione asincrona: restituisce una promise
  return new Promise((resolve, reject) => {
    setTimeout(() => {
      const r = Math.random();
      if (r > p) resolve(r); else reject(r);
    }, t);
  });
}

delayWithRandom(1000, 0.5).then(
  (r) => console.log('success:', r),       // onComplete
  (r) => console.log('fail:', r));         // onError (facoltativa)

Chi usa una libreria asincrona riceve promise già fatte e le configura. Chi la scrive crea la promise con il costruttore, che passa le due continuazioni resolve e reject; riscrivere così una vecchia API a callback si dice promisificare. then non esegue nulla, registra una reazione: anche se l'operazione è già finita, la callback verrà eseguita dall'event loop dopo il codice corrente. Per questo non importa in che punto dell'handler si configura una promise.

Il chaining appiattisce la piramide

then restituisce una nuova promise, che si risolve con il valore restituito dalla callback. Se la callback restituisce a sua volta una promise, la nuova promise ne segue l'esito. Una sequenza di operazioni asincrone diventa una catena lineare, che si legge dall'alto in basso:

findUserById('john')
  .then((user) => findPic(user.picId))      // restituire la promise è ciò che appiattisce
  .then((pic) => ui.show(pic));

Due then sulla stessa promise danno invece il flusso parallelo: entrambe le callback ricevono lo stesso valore. Per comporre più promise indipendenti ci sono Promise.all, che si risolve quando tutte sono risolte (con l'array dei valori) e si rigetta appena una è rigettata, e Promise.race, che si conclude con la prima promise che si risolve o si rigetta.

Gli errori seguono regole simili alle eccezioni. Un handler che restituisce un valore passa un successo alla promise successiva (anche se è un handler di errore), uno che lancia un'eccezione le passa un rigetto. Un rigetto senza handler scorre lungo la catena finché ne trova uno (.catch), poi l'elaborazione riprende normalmente.

x = 0;                                   // L1
const p = asyncTask(x);                  // restituisce subito una promise pending
p.then((res) => { x = x + res; });       // L2: registrata, non eseguita
x = x + 1;                               // L3   → l'ordine resta L1, L3, L2

// CPS                                   // promise
x = 0;                                   x = 0;
asyncTask1(x, (res1) => {                asyncTask1(x)
  asyncTask2(x + res1, (res2) => {         .then((res1) => asyncTask2(x + res1))
    console.log(res2);                     .then((res2) => console.log(res2));
  });                                    x = x + 1;
});
x = x + 1;

Promise e future

Entrambe rappresentano un risultato che arriverà, ma vivono in mondi diversi.

• La Future di Java, restituita da submit di un executor, vive nel mondo a thread: si interroga (isDone) o ci si blocca con get(). Executor e future sono nel capitolo 10.
• La promise vive nel mondo a eventi: non ha una get bloccante, si configura con callback eseguite dall'event loop.
• Alcune librerie dividono la promise in due facce. Vert.x chiama Promise il lato di chi produce il risultato (complete, fail) e Future il lato in sola lettura del cliente (onSuccess, compose), che non ha get: non è la Future di Java.

CompletableFuture e supplyAsync. CompletableFuture.supplyAsync(supplier) esegue il Supplier su un thread del ForkJoinPool.commonPool(), o sull'executor passato come secondo argomento, e restituisce subito una CompletableFuture. Vi si attaccano callback (thenApply, thenAccept, thenCompose), ma resta una future Java con get() e join() bloccanti: per Ricci è un ibrido che mescola i due paradigmi. Va bene per lanciare calcoli o I/O su un pool di thread. Ma le callback non passano da un event loop: le esegue il thread che completa lo stadio precedente, o il chiamante se è già completo. Se toccano stato condiviso tornano le race, quindi in un framework a eventi non va usata come una promise.

CompletableFuture.supplyAsync(() -> loadReport(dir))     // su un thread del common pool
    .thenApply(Report::summary)                          // su un thread del pool, non su un event loop
    .thenAccept(System.out::println);

I limiti delle promise

• Eagerness e parametri di then. Una promise lavora appena è costruita, e a then si passa una funzione, senza parametri. .then(delay(1000)) chiama delay subito, mentre si configura la catena, e i due ritardi partono insieme; serve .then(() => delay(1000)).
• Cicli. for (let i = 0; i < 3; i++) delay(1000).then(() => console.log(i)) crea le tre promise nello stesso ciclo: dopo un secondo arrivano insieme 0, 1, 2, non una stampa al secondo. Serve la ricorsione.
• Cancellazione. Una promise non si può annullare: non c'è controllo sul task, e le risorse possono andare sprecate.

6. Task e microtask

Nel mondo JavaScript e web la coda degli eventi (dispatch queue) contiene due tipi di elementi. I task (I/O, timer, input dell'utente) sono serviti uno alla volta, ciascuno fino in fondo, eventualmente con priorità. I microtask sono task leggeri con una garanzia in più: vengono eseguiti tutti prima che inizi il task successivo. Le reazioni delle promise sono microtask, e Promise.resolve().then(() => { … }) ne accoda uno esplicitamente, che gira subito dopo il codice corrente. Il rovescio: mentre l'event loop svuota i microtask, l'app non disegna, non risponde ai click e non reagisce all'I/O (esempio da Dart). Un microtask che ne accoda sempre un altro affama l'intera applicazione.

Due domande su setTimeout

setTimeout(callback, t) registra un handler per l'evento «il timer è scaduto». Anche con t = 0 la callback è un task: non verrà mai eseguita prima che l'handler corrente sia terminato.

for (var i = 0; i < 3; i++) {             for (let i = 0; i < 3; i++) {
  setTimeout(() => console.log(i), 0);       setTimeout(() => console.log(i), 0);
}                                         }
// stampa 3, 3, 3                         // stampa 0, 1, 2

L'event loop spiega solo quando girano le callback: in entrambi i casi dopo la fine del for, in tre task successivi. La differenza tra gli output la fanno le closure. Con var la variabile ha visibilità di funzione: esiste un solo binding i, condiviso dalle tre closure, che quando vengono eseguite vale 3. Con let il for crea un binding nuovo a ogni iterazione, copiando il valore prima dell'incremento: ogni closure cattura il suo, che nessuno modifica più. Gli scenari «var» e «let» del laboratorio mostrano quale closure cattura quale binding. L'esempio delle slide, con let i = 1 … 3, stampa i tre «setting timer…» e poi 1, 2, 3.

7. async/await

Con async e await il codice asincrono si scrive in stile sincrono, senza callback né catene di then, mentre il nucleo resta asincrono. Ci sono in JavaScript da ES2017, in C# con il TAP di .NET (prima ancora, dal 2012), in Python, Dart e altri linguaggi. Sono parole chiave, non una libreria: cambiano il modello di esecuzione del linguaggio.

• await p sospende la funzione, non il thread: la funzione salva il proprio stato, il controllo torna al chiamante, e l'event loop la riprende quando p si risolve. await trasforma la promise nel suo valore, o in un'eccezione se è rigettata, come una chiamata sincrona.
• async delimita il codice che un await può sospendere: await si usa solo dentro una funzione async. Una funzione async restituisce sempre una promise, anche senza await, e chiamarla esegue subito il corpo, nello stesso ciclo, fino al primo await.

async function pause(t) {
  console.log('before');
  await new Promise((resolve) => setTimeout(resolve, t));
  console.log('after');
}
console.log('before call');
pause(1000);                // senza await: before call, before, after call, after
console.log('after call');
// dentro una funzione async, con "await pause(1000)":  before call, before, after, after call

Sotto, await è un return che si può riprendere, realizzato con le coroutine (sezione 8). L'eagerness non è più un problema, perché le istruzioni si eseguono nell'ordine in cui sono scritte, e un await dentro un for aspetta davvero a ogni giro.

async function f() {                     async function run() {
  const res = await asyncTask(x);          const res1 = await asyncTask1(x);
  x = x + res;               // L2         const res2 = await asyncTask2(x + res1);
}                                          console.log(res2);
x = 0;                       // L1       }
f();                                     x = 0;
x = x + 1;                   // L3       run();
                                         x = x + 1;

I problemi che restano

• await non si usa al livello più esterno: serve almeno una funzione main asincrona. E l'async si propaga: chi fa await deve essere a sua volta async.
• Una funzione sospesa non riprende finché il thread è occupato con altri handler: la semantica dell'event loop resta quella.
• I blocchi {…} non sono più atomici. Un blocco può estendersi su più iterazioni dell'event loop, e tra un await e l'altro altri handler possono modificare lo stesso stato: non race di basso livello, ma race condition logiche.
• Per comporre serve ancora l'API delle promise, ad esempio await Promise.all([f(), g()]): le promise non spariscono, si integrano.

Resta infine uno scontro di progetto tra sincrono e asincrono: mescolarli richiede disciplina, e mancano concetti di più alto livello per progettare. Ne soffre la modularità: una g non async che chiama una async function f stampa «started g», «started f», «done g» e solo dopo «done f». Per aspettare f anche g deve diventare async e fare await f(), quindi il chiamante deve sapere se ciò che chiama è asincrono.

8. Coroutine e fibre

Una coroutine (Conway, 1963) generalizza la subroutine: l'esecuzione si può sospendere e riprendere. Invece di tornare sempre al chiamante, una coroutine può cedere il controllo a un'altra (yield), che più tardi glielo restituirà nel punto in cui si era fermata. È multitasking cooperativo, non preemptive: nessuno toglie il controllo a una coroutine, è lei a cederlo. Non ha a che fare con gli eventi: serve per costruire task cooperativi, eccezioni, event loop, iteratori, liste infinite e pipe.

var q := new queue
coroutine produce                       coroutine consume
  loop                                    loop
    while q is not full                     while q is not empty
      create some new items                   remove some items from q
      add the items to q                      use the items
    yield to consume                        yield to produce

• Un'istanza di coroutine ha uno stato che persiste tra un'invocazione e l'altra, e della stessa coroutine possono esistere più istanze.
• Una subroutine è il caso particolare di una coroutine che non fa mai yield.
• È facile da realizzare in linguaggi con continuazioni native, ma non come semplice libreria sopra un linguaggio che ha solo subroutine e metodi.

Nate in assembly, presenti in Simula, Smalltalk e Modula-2, le coroutine sono tornate con la programmazione asincrona in Ruby, Lua, Julia, Go e Kotlin. Sono il mattone dei thread leggeri: un thread ospita molte coroutine ma ne esegue una alla volta, e crearle o scambiarle non richiede context switch del sistema operativo. Possono quindi girare tutte su un solo thread. È un meccanismo di basso livello: usarlo direttamente per passarsi il controllo tra handler assomiglia a un goto, e rompe la modularità allo stesso modo.

In Kotlin: le funzioni sospendibili

• Computazioni sospendibili: si sospendono senza bloccare il thread, che nel frattempo esegue altro. Sono le suspending function, marcate suspend, e le suspending lambda, che possono chiamare funzioni sospendibili.
• Dispatcher: decidono su quale thread una coroutine parte o riprende.
• Builder: creano una coroutine da una suspending lambda. async quando serve un risultato, launch quando non serve, runBlocking come ponte tra codice bloccante e sospendibile.

fun main() = runBlocking {
    println("before async call")
    val result = async {
        println("inside the async call")
        delay(1000)
        println("exiting the async call")
        100
    }
    println("after the async call, before greet")
    greetDelayed(200)
    println("after greet trigger")
    println("${result.await()}")
}

suspend fun greetDelayed(delayMillis: Long) {
    delay(delayMillis)          // sospende greetDelayed, non il thread
    println("Hello, World!")
}
// before async call · after the async call, before greet · inside the async call
// Hello, World! · after greet trigger · exiting the async call · 100

Tutto gira sul thread di runBlocking: il blocco di async parte quando main si sospende in delay(200), e le due coroutine si alternano a ogni sospensione.

Fibre e virtual thread

Una fibra è un thread leggero che condivide lo spazio di indirizzi ma fa multitasking cooperativo: è la fibra a cedere il controllo, quindi tutte possono girare su un solo thread. La coroutine è un costrutto del linguaggio, una forma di controllo del flusso; la fibra è un costrutto di sistema, un thread che non gira in parallelo agli altri, e può implementare le coroutine o farne da base. Esistono in vari sistemi (ad esempio Windows); in Java sono i virtual thread, nel capitolo 10. Virtual thread o programmazione asincrona? Per Brian Goetz (2022) i virtual thread modularizzano meglio i flussi di controllo, ma, come i thread fisici, non trattano eventi e computazioni reattive come concetti di prima classe. Il dibattito è aperto.

9. L'event loop in Java: Vert.x

Vert.x (Eclipse) è un framework per applicazioni reattive sulla JVM: il nucleo realizza il reactor con più event loop («multi-reactor») e tutte le API sono asincrone, basate su promise. È il framework di riferimento del corso per l'assignment 2 (che si può svolgere anche in JavaScript o Python, purché con un event loop vero). Ogni componente attivo è un verticle con il proprio event loop: il codice della classe gira su quel loop, e i suoi campi li tocca solo lui. Le funzioni asincrone restituiscono una Future di Vert.x, il lato cliente di una promise, che si configura con onComplete, onSuccess e onFailure.

class MyReactiveAgent extends VerticleBase {
  private int cycle = 0;                          // stato del verticle: solo il suo event loop lo tocca

  @Override
  public Future<?> start() {
    FileSystem fs = vertx.fileSystem();
    fs.readFile("hello.md").onSuccess(b -> log(++cycle + ": hello.md"));
    fs.readFile("pom.xml").onSuccess(b -> log(++cycle + ": pom.xml"));
    return Future.succeededFuture();              // start è finito: le letture continuano in background
  }
}

Vertx.vertx().deployVerticle(new MyReactiveAgent());   // anch'esso asincrono: restituisce una Future

fs.readFile("hello.md")
  .compose(md -> fs.readFile("pom.xml"))          // compose = then che restituisce una promise
  .compose(pom -> fs.readDir("src"))
  .onSuccess(files -> log(files.size() + " file in src"))
  .onFailure(Throwable::printStackTrace);

Future.all(fs.readFile("hello.md"), fs.readFile("pom.xml"))   // join: entrambe completate
  .onSuccess(cf -> log("lette " + cf.size()));
// Future.any: la prima che si completa con successo

Future<Double> getDelayedRandom(long delay) {
  Promise<Double> p = Promise.promise();          // lato produttore
  vertx.setTimer(delay, id -> p.complete(Math.random()));
  return p.future();                              // lato cliente: niente complete, niente get
}

getDelayedRandom(1000).onSuccess(v -> log("valore: " + v));

vertx.executeBlocking(() -> {                     // gira su un worker thread, non sull'event loop
        Thread.sleep(5000);                       // qui si può bloccare
        return 100;                               // ma NON toccare i campi del verticle
      }, false)                                   // false: esecuzione non ordinata
     .onSuccess(v -> log("risultato " + v));     // di nuovo sull'event loop

vertx.eventBus().consumer("news", msg -> log("ricevuto " + msg.body()));   // in un verticle
vertx.eventBus().publish("news", "ciao");                                  // in un altro

Verifica le tue conoscenze