Programmazione reattiva

1. Dall'assegnamento alla relazione

La programmazione a eventi ha problemi che le promise e async/await non eliminano. In un'applicazione asincrona l'ordine di arrivo degli eventi esterni non si può prevedere né controllare, e il controllo salta da un handler all'altro. È l'inversione del controllo: il flusso lo guidano gli eventi, non un ordine scelto dal programmatore. Quando un dato cambia, il programmatore deve aggiornare a mano tutto ciò che ne dipende, ed è facile farlo nel momento sbagliato o nell'ordine sbagliato. Tanti frammenti di codice isolati manipolano gli stessi dati in un ordine imprevedibile (lo spaghetti asincrono). E poiché le callback di solito non restituiscono nulla, possono agire solo con effetti collaterali sullo stato.

La programmazione reattiva (RP) è un paradigma orientato ai flussi di dati e alla propagazione del cambiamento. Il programma resta una reazione a eventi esterni, ma è il linguaggio a gestire lo scorrere del tempo e le dipendenze tra dati e calcoli. Il programmatore scrive che cosa fare, e il sistema decide quando: si astrae dalla gestione del tempo come il garbage collector astrae dalla gestione della memoria. Nell'esempio delle slide

var1 = 1
var2 = 2
var3 = var1 + var2

var3 dipende da var1 e var2: ogni volta che una delle due cambia, var3 viene ricalcolata automaticamente.

Le radici sono nella ricerca. La maggior parte del lavoro sulla RP discende da Fran (Elliott e Hudak, 1997), un linguaggio funzionale per la grafica e le animazioni interattive. Da lì viene la Functional Reactive Programming (FRP), basata sul paradigma synchronous dataflow (Lee e Messerschmitt, 1987) con vincoli real-time più rilassati. Il riferimento del corso è il survey di Bainomugisha e colleghi (ACM Computing Surveys, 2013). Nata negli anni '90, oggi la RP si usa per le app web reattive e per i big data.

2. Event stream e behaviour

La RP rappresenta come astrazioni i valori che variano nel tempo. Sono valori di prima classe: si passano ad altre funzioni e si compongono. Le astrazioni sono di due tipi:

• Event stream: valori discreti (o sparsi) nel tempo. Sono astrazioni asincrone di un flusso progressivo di dati intermittenti e sequenziali, generati da un evento ricorrente. Esempi: i click del mouse, un oggetto che passa su un nastro trasportatore.
• Behaviour (o signal): valori continui nel tempo, un flusso ininterrotto di dati da una sorgente costante, che ha un valore in ogni istante. Esempi: un timer, la temperatura letta da un sensore, anche se sotto viene campionata.

Un esempio in Flapjax, una libreria reattiva per JavaScript: un elemento della pagina che mostra i secondi trascorsi.

var timer   = timerB(100);                          // behaviour: il tempo, aggiornato ogni 100 ms
var seconds = liftB(function (time) {               // behaviour derivato da timer
                return Math.floor(time / 1000);
              }, timer);
insertDomB(seconds, 'timer-div');                   // l'elemento della pagina segue seconds

Non ci sono callback né listener. seconds si ottiene da timer come map ottiene una lista da un'altra: per ogni valore di timer c'è il valore corrispondente di seconds. La pagina si aggiorna da sola. Tutto è dichiarativo.

3. Come si propaga il cambiamento: push, pull, glitch

Il modello di valutazione di un linguaggio reattivo stabilisce come i cambiamenti si propagano lungo il grafo delle dipendenze. Per il programmatore la propagazione è automatica. Chi progetta il linguaggio deve però decidere chi la avvia.

• Pull: chi ha bisogno di un valore lo «tira» dalla sorgente. La propagazione è guidata dalla domanda (demand-driven) e si realizza naturalmente in linguaggi lazy come Haskell.
• Push: quando la sorgente ha un dato nuovo, lo «spinge» ai calcoli che ne dipendono. La propagazione è guidata dalla disponibilità di dati (data-driven) e si realizza in linguaggi eager, come Flapjax (JavaScript) e Scala.React.

In entrambi i casi i dati vanno dal produttore al consumatore: cambia chi prende l'iniziativa.

I glitch

Un glitch è un'inconsistenza temporanea durante la propagazione. Un calcolo viene eseguito prima che siano state valutate tutte le espressioni da cui dipende, e combina valori nuovi con valori vecchi (Cooper e Krishnamurthi, 2006). Riguarda solo il modello push. L'esempio delle slide:

var1 = 1
var2 = var1 * 1        // deve valere sempre quanto var1
var3 = var1 + var2     // deve valere sempre il doppio di var1

Rendere osservabile lo stato di t₂ vuol dire mostrare al mondo esterno uno stato che, secondo le relazioni dichiarate, non è mai esistito. Per evitarlo (glitch avoidance) le librerie costruiscono il grafo delle dipendenze e aggiornano un nodo solo dopo tutti quelli da cui dipende, in ordine topologico. Gli aggiornamenti diventano così atomici rispetto all'osservatore. Le implementazioni recenti evitano i glitch su un solo computer. Nei programmi reattivi distribuiti è ancora un problema aperto, per via di guasti, ritardi di rete e assenza di un orologio globale (Salvaneschi et al. 2014, Mogk et al. 2018): la RP distribuita, insieme alla tolleranza ai guasti, è una delle direzioni di ricerca attuali.

4. Lifting e composizione

Un valore che dipende da valori reattivi deve essere reattivo a sua volta: una variabile a cui si assegna un'espressione con behaviour o event stream diventa reattiva. Il lifting è la conversione di una variabile, o di una funzione, normale in una reattiva. liftB nell'esempio del timer applica la funzione ordinaria time => Math.floor(time / 1000) a un behaviour. È il lifting a registrare le dipendenze: per ricalcolare correttamente tutte le espressioni, la libreria costruisce dietro le quinte il grafo delle dipendenze. Quando timer cambia, il grafo sa che va rivalutata la funzione, poi aggiornato seconds e poi la pagina.

Il lifting può essere di tre tipi:

• implicito: lo fa la libreria (Bacon.js);
• esplicito: lo fa il programmatore (React.js);
• entrambi: Flapjax usato come libreria richiede il lifting esplicito, usato come compilatore trasforma il codice e lo rende implicito.

La composizione di astrazioni reattive è ciò che evita l'incubo delle callback. Tre callback separate per click, movimento e rilascio del mouse diventano un solo event stream che risponde a tutti e tre. Bacon.js compone i behaviour con combine, Flapjax gli event stream con mergeE:

var saveTimer   = timerE(10000);                          // un evento ogni 10 secondi
var saveClicked = extractEventE('save-button', 'click');  // un evento a ogni click su Save
var save        = mergeE(saveTimer, saveClicked);         // l'uno o l'altro
save.mapE(doSave);                                        // salva a ogni evento di save

«Salva ogni 10 secondi o quando l'utente preme Save» diventa quattro righe dichiarative, senza gestire né il tempo né il pulsante.

5. Reactive Extensions: l'Observer fatto bene

Nei linguaggi mainstream la RP arriva come libreria: le Reactive Extensions (Rx), un'API per la programmazione asincrona con flussi osservabili; Reactive Streams, uno standard per flussi asincroni con backpressure non bloccante (JVM e JavaScript); Project Reactor, una libreria per la JVM basata su Reactive Streams. Rx nasce su .NET dalla ricerca sulla FRP e oggi esiste per quasi ogni linguaggio (RxJS, RxJava, RxScala, RxPY…), con la stessa semantica; Angular la usa nel frontend e nel backend. Si presenta come «l'Observer fatto bene», una combinazione delle idee migliori del pattern Observer, del pattern Iterator e della programmazione funzionale (E. Meijer, Your mouse is a database, 2012).

Nel pattern Observer un oggetto osservabile (il subject) tiene una lista di osservatori e li notifica con un push quando accade qualcosa, senza conoscere chi li usa. Rx estende l'idea: una sorgente emette una sequenza di valori e ne segnala la fine o l'errore, le sequenze si compongono con operatori come le collezioni, e una sottoscrizione si può annullare.

interface IObservable<T> { IDisposable Subscribe(IObserver<T> observer); }
interface IObserver<T>   { void OnNext(T value); void OnError(Exception error); void OnCompleted(); }
interface IDisposable    { void Dispose(); }      // annulla la sottoscrizione

Le proprietà fondamentali di Rx sono tre:

• asincrona e a eventi: semplificare questi modelli di programmazione è la missione di Rx, ed è la chiave per interfacce, web app e cloud app reattive;
• composizione: comporre facilmente calcoli asincroni;
• collezioni osservabili: i calcoli asincroni sono sorgenti di dati da interrogare come con LINQ. Il mouse «diventa un database di movimenti e click», da filtrare, proiettare, unire (join) e trattare nel tempo.

Per le slide, in .NET la Subscribe è asincrona: il chiamante non resta bloccato fino alla fine della sequenza, che può girare su un altro thread. Rx si integra con i flussi sincroni (IEnumerable<T>) e con i calcoli asincroni a valore singolo (Task<T>), e gli operatori LINQ trasformano (aggregazione, proiezione), compongono (Zip, SelectMany) e interrogano (Where, Any). Concat di due Range(1, 3) dà 1, 2, 3, 1, 2, 3, perché la seconda parte solo quando la prima è finita; Merge le rende attive insieme, e l'esempio delle slide dà 1, 1, 2, 2, 3, 3.

Quattro idee comuni

La documentazione di Project Reactor riassume ciò che hanno in comune queste librerie:

• Componibilità e leggibilità. Componibilità è saper orchestrare più task asincroni, usando i risultati dei precedenti come input dei successivi. Va di pari passo con la leggibilità: con molti livelli asincroni il codice diventa difficile da comporre e da leggere, come nel callback hell.
• Dati come flusso, con un ricco vocabolario di operatori. Come in una catena di montaggio, la materia prima esce da una sorgente (il Publisher), passa per le stazioni di lavoro (gli operatori) e arriva al consumatore (il Subscriber). Ogni operatore avvolge il Publisher precedente in uno nuovo che aggiunge un comportamento.
• Nulla accade finché non ci si sottoscrive. Costruire la catena è solo una configurazione. La subscribe lega un Publisher a un Subscriber e fa scorrere i dati: la richiesta del Subscriber risale la catena fino alla sorgente.
• Backpressure: il consumatore può segnalare al produttore che il ritmo è troppo alto (sezione 9).

6. RxJava: creare, trasformare, osservare

RxJava è Rx per la JVM: estende il pattern Observer alle sequenze di dati ed eventi e aggiunge operatori per comporle, nascondendo thread, sincronizzazione, strutture dati concorrenti e I/O non bloccante. La 1.x era Rx sopra Java, la 2.x (2017) si è allineata a Reactive Streams, e le slide indicano la 4.x (2026) come ultima major release; gli esempi usano l'API di RxJava 3. I flussi sono Observable, leggero e senza backpressure, e Flowable, più costoso, per grandi quantità di dati e con backpressure; per il resto sono equivalenti.

Flowable.range(1, 20)                          // creazione: i numeri da 1 a 20
    .map(v -> v * v)                           // trasformazione: i quadrati
    .filter(v -> v % 3 == 0)                   // filtro: solo i multipli di 3
    .subscribe(System.out::println);           // 9, 36, 81, 144, 225, 324

Observable<Integer> source = Observable.create(emitter -> {   // un flusso scritto a mano
    for (int i = 0; i < 3; i++) emitter.onNext(i);
    emitter.onComplete();                      // chiude il flusso
});
Disposable d = source.subscribe(
    v -> log("ricevuto " + v),                 // onNext
    err -> log("errore " + err),               // onError: gli operatori propagano, gestisce il subscriber
    () -> log("completato"));                  // onComplete
// d.dispose() annulla la sottoscrizione (in RxJava 1 era subscription.unsubscribe())

Ogni operatore crea un flusso nuovo: nessun flusso viene modificato e non ci sono effetti collaterali, il che rende questi sistemi efficienti anche in concorrenza. La catena non contiene elementi, sa come generarli. Senza scheduler la subscribe è sincrona: il thread chiamante (qui il main) genera gli elementi ed esegue le lambda. Le eccezioni di un operatore non si gestiscono lì: arrivano all'onError del subscriber.

Per creare flussi: just, fromArray, fromIterable e range fanno lifting di dati esistenti; create riceve una lambda che chiama onNext sull'emitter; fromCallable dà il risultato di un calcolo; interval emette un intero a ogni periodo. Un flusso generalizza la programmazione asincrona a più risultati nel tempo, dove una promise ne ha uno solo. flatMap trasforma ogni elemento in un flusso e li fonde in uno solo. zipWith accoppia gli elementi di due flussi a due a due, qualunque sia il momento in cui arrivano:

Flowable.just("the", "quick", "brown", "fox")
    .zipWith(Flowable.range(1, 5), (w, n) -> n + ". " + w)
    .subscribe(System.out::println);           // 1. the, 2. quick, 3. brown, 4. fox

Ogni operatore è documentato da un marble diagram: flusso d'ingresso su una linea del tempo, operatore, flusso d'uscita. doOnNext lascia passare gli elementi eseguendo un'azione, utile per il debug; operatori temporali come debounce, sample e buffer esprimono dichiarativamente comportamenti legati al tempo (sezione 8).

7. Su quali thread: gli scheduler

In RxJava non si creano thread: si dichiara su quale scheduler, un'astrazione simile a un executor, deve girare una parte della catena.

• subscribeOn(s) sposta su s l'esecuzione dell'intera catena, dalla sorgente in giù, e non importa in che punto della catena lo si scrive. Stacca il lavoro dal thread che chiama subscribe.
• observeOn(s) sposta su s solo gli stadi a valle del punto in cui compare, fino al subscriber. Si può usare più volte per cambiare thread in punti diversi della catena.

Gli scheduler predefiniti sono tre. Schedulers.computation() è per il lavoro CPU-bound, su un numero fisso di thread dedicati, ed è il default della maggior parte degli operatori asincroni. Schedulers.io() è per l'I/O e le operazioni bloccanti, su un insieme di thread che cresce e cala. Schedulers.single() esegue tutto su un solo thread, in ordine FIFO. Con Schedulers.from(executor) si avvolge un executor esistente, ad esempio per avere un pool fisso più grande.

Flowable.fromCallable(() -> { Thread.sleep(1000); return "Done"; })   // calcolo costoso o I/O bloccante
    .subscribeOn(Schedulers.io())                  // …su un thread di I/O
    .observeOn(Schedulers.single())                // il risultato si elabora su un altro thread (es. la GUI)
    .subscribe(System.out::println, Throwable::printStackTrace);

Flowable.range(1, 10)
    .flatMap(v -> Flowable.just(v)
        .subscribeOn(Schedulers.computation())     // ogni sottoflusso su un thread del pool
        .map(w -> w * w))
    .blockingSubscribe(v -> log("sub " + v));      // osservati dal thread chiamante, uno alla volta

Il secondo esempio è lo schema del parallelismo. subscribeOn su una catena sola la sposta su un altro thread, ma non ne divide gli stadi. Dentro flatMap, invece, ogni sottoflusso ha il suo subscribeOn, e i quadrati sono calcolati in parallelo su thread diversi del pool, arrivando fuori ordine. blockingSubscribe riporta l'osservazione su un solo thread, quello che la chiama. Così gli elementi calcolati in parallelo si elaborano in sequenza, senza lock. Conviene prima far funzionare la catena senza scheduler e solo dopo inserire subscribeOn e observeOn dove serve parallelizzare: la logica non cambia.

8. Hot e cold

La famiglia Rx distingue due categorie di flussi, secondo come reagiscono ai subscriber.

• Un flusso cold fa partire un flusso nuovo per ogni subscriber, sorgente compresa, e ogni subscriber riceve tutti gli elementi, qualunque sia il momento in cui si è sottoscritto. Emette solo se c'è almeno un subscriber: è lazy, e le lab notes lo associano al modello pull. Sono cold i flussi creati da strutture dati (just, range, fromIterable) e da create. Nell'esempio con due subscriber e un thread creato dentro create, i thread diventano due: ogni sottoscrizione rimette in moto il generatore.
• Un flusso hot non riparte per ogni subscriber: chi arriva tardi riceve solo i segnali emessi dopo la propria sottoscrizione. Emette al proprio ritmo, anche se nessuno ascolta, ed è l'eccezione a «nulla accade finché non ci si sottoscrive». È il modello push: produttore e consumatori sono del tutto disaccoppiati, e tocca agli osservatori tenere il passo. Alcuni flussi hot memorizzano e rigiocano la storia delle emissioni, tutta o in parte. Esempi: sensori, dati dalla rete, eventi della GUI.

ConnectableObservable<Long> hot = Observable.interval(10, TimeUnit.MILLISECONDS).publish();
hot.connect();                                     // parte subito, anche senza subscriber
Thread.sleep(100);
hot.subscribe(v -> log("A " + v));                 // vede solo gli elementi da qui in poi

In questa forma di base, il thread che genera gli elementi di un flusso hot è anche quello che esegue gli osservatori: un osservatore lento rallenta la sorgente. Per disaccoppiarli si usa observeOn.

PublishSubject: un ponte con il codice non reattivo

Nei flussi creati con create gli elementi li genera la lambda passata alla creazione. Un PublishSubject invece è un flusso hot che fa anche da canale: chiunque, da qualsiasi thread, può chiamare onNext per inserirvi elementi (non da più thread contemporaneamente: un subject non è thread-safe, e per questo esiste toSerialized()), e chiunque può sottoscriversi. È il modo più semplice di collegare codice legacy o una GUI al mondo reattivo. Va però usato con giudizio, perché permette di scrivere codice poco leggibile. Un esempio classico: rilevare i «multi-click», cioè almeno due click separati da meno di 250 ms.

PublishSubject<Integer> clicks = PublishSubject.create();
button.addActionListener(e -> clicks.onNext(1));          // Swing produce elementi nel flusso

clicks.buffer(clicks.debounce(250, TimeUnit.MILLISECONDS))  // raggruppa i click fino a 250 ms di silenzio
      .map(List::size)
      .filter(n -> n >= 2)
      .subscribe(n -> System.out.println("multi-click: " + n));

debounce emette un elemento solo dopo 250 ms senza altri click, e segna così la fine di ogni raffica. buffer usa quei segnali come confini per raggruppare i click. Il resto è map e filter. Senza operatori, la stessa logica richiederebbe timer, stato e callback.

9. Backpressure

Ogni stadio di una catena consuma ciò che produce il precedente, e gli stadi possono avere velocità molto diverse. In un produttore-consumatore classico il buffer pieno blocca il produttore; qui bloccare non si può, e gli elementi in eccesso si accumulano in buffer sempre più grandi o vanno scartati. La backpressure è un controllo di flusso in cui ogni stadio dichiara quanti elementi è pronto a elaborare: il segnale che una stazione lenta della catena di montaggio manda a monte per limitare il flusso.

Un subscriber può lavorare in modalità unbounded, lasciando che la sorgente spinga i dati alla massima velocità. Oppure può usare il meccanismo di request: chiede al massimo n elementi. Il modello push diventa un ibrido push-pull. Il consumatore tira n elementi se sono già disponibili; se non lo sono, la sorgente li spinge appena li produce. Nel caso cold il problema quasi non si pone, perché è il subscriber a tirare gli elementi. Nel caso hot è il generatore a spingerli. In RxJava la backpressure c'è solo per Flowable, ed è obbligatoria nello standard Reactive Streams.

Flowable.interval(5, TimeUnit.MILLISECONDS)                     // un elemento ogni 5 ms
    .observeOn(Schedulers.single())
    .subscribe(v -> { Thread.sleep(100); log("consumo " + v); },  // ne consuma uno ogni 100 ms
               err -> log("errore: " + err));                     // MissingBackpressureException

Dopo poco il buffer interno si riempie e RxJava segnala un errore: non può emettere un valore perché nessuno lo ha richiesto. Le strategie per evitarlo sono di due tipi. Si può ridurre il flusso con operatori: buffer(n) raggruppa gli elementi in pacchetti, window(n) li raggruppa in lotti che sono flussi, e sample(t), debounce(t) e throttleFirst(t) ne saltano una parte. Oppure si può dire che cosa fare quando il buffer si riempie:

source.onBackpressureBuffer(5000,                                // un buffer più grande…
        () -> log("buffer pieno"),                              // …con una reazione al riempimento
        BackpressureOverflowStrategy.DROP_OLDEST);              // ERROR, DROP_OLDEST o DROP_LATEST
source.onBackpressureDrop();                                     // scarta tutto ciò che è in eccesso
Flowable.create(emitter -> { /* onNext… */ }, BackpressureStrategy.BUFFER);   // o la strategia scelta alla creazione

Nessuna pallottola d'argento

Né la RP né Rx sono il modello generale per la programmazione asincrona. Sono efficaci per gestire flussi di dati e di eventi in stile funzionale, e il loro vantaggio si vede solo se la logica del problema si esprime come composizione di flussi. Usare Rx solo per sottoscrivere lambda, come un Observer qualsiasi, non è programmazione reattiva. Resta aperta la sfida di integrare gli approcci: sincrono e asincrono, push e pull, concorrenza con più event loop che comunicano (cioè gli attori), e modelli che funzionino anche nei sistemi distribuiti. Le domande su come l'assignment 2 usa la programmazione reattiva e le promise sono nella pagina di preparazione.

Verifica le tue conoscenze