Progettare programmi concorrenti

1. Dal problema all'architettura in due passi

I capitoli precedenti hanno dato i meccanismi. Resta la domanda dell'ingegnere: dato un problema, come si organizza il programma? Il metodo del corso ha due passi.

1. Analisi. Si cerca la concorrenza che il problema offre: quali task ci sono, come si possono scomporre il lavoro e i dati, quali dipendenze legano i task. Si guarda il problema, non la soluzione. È la parte più difficile, perché viene spontaneo analizzare pensando già a come implementare.
2. Progetto. Si sceglie un'architettura concorrente. I task vengono assegnati ad agenti, componenti attivi che ne sono responsabili e ne incapsulano la logica e il controllo. Per lavorare insieme gli agenti usano componenti passivi condivisi, di due tipi: risorse che raccolgono il risultato del loro lavoro, e mezzi di coordinazione che danno forma alla loro interazione e gestiscono le dipendenze tra i task.

I due passi non sono una cascata: analisi, progetto e implementazione si ripetono a spirale, come nei metodi agili. Ma non c'è buon progetto senza una buona analisi.

Un task è, in senso operativo, una sequenza di istruzioni che operano insieme come un gruppo e corrispondono a una parte logica del programma; in senso più astratto, un lavoro da svolgere come responsabilità di un agente. In entrambi i casi è un'unità di lavoro logica che inizia e finisce, non un thread: il thread, o l'agente, è ciò che lo esegue.

2. Componenti attivi e passivi

Un oggetto incapsula stato e comportamento, ma non il flusso di controllo: i suoi metodi sono eseguiti dal thread che li chiama, e leggendo la classe non si sa quale sarà. La disciplina del corso aggiunge l'incapsulamento del flusso di controllo, con una partizione netta dei componenti in due tipi.

Le regole sono tre:

• Gli agenti non hanno interfaccia pubblica. Nessun altro può chiamare i loro metodi, quindi il loro codice è eseguito solo dal loro flusso di controllo. Un metodo di un agente chiamato da un altro thread è una violazione: qualcun altro esegue il «mio» codice.
• Ogni interazione tra agenti passa per componenti passivi condivisi, progettati come monitor. Nella parte III gli agenti diventano attori, che interagiscono invece scambiandosi messaggi (capitolo 15).
• Niente entità ibride. Un'entità che ha un proprio thread e insieme offre metodi ad altri thread va scomposta in uno o più componenti attivi e passivi. Una sottoclasse di Thread con metodi pubblici synchronized chiamati da altri thread diventa un agente più un monitor.

Un agente è quindi un componente attivo, che porta avanti un compito di sua iniziativa, sia esso un thread, un event loop o un attore. Monitor, buffer e future sono passivi: agiscono solo quando un agente li chiama.

In Java un agente estende Thread: il costruttore riceve la configurazione e i componenti passivi da usare, il comportamento sta in run, e i metodi privati o protetti sono le sue «abilità», che un sottotipo eredita. run è pubblico solo perché lo impone Thread. Nell'ordinamento concorrente del corso i worker ordinano ciascuno una parte dell'array e un merger le fonde; qui le parti si raccolgono in un monitor:

// Componente passivo: raccoglie le parti ordinate e sa quando ci sono tutte
class SortedParts {
  private final List<int[]> parts = new ArrayList<>();
  private final int expected;
  SortedParts(int expected) { this.expected = expected; }

  public synchronized void add(int[] part) { parts.add(part); notifyAll(); }

  public synchronized List<int[]> awaitAll() throws InterruptedException {
    while (parts.size() < expected) wait();
    return List.copyOf(parts);
  }
}

// Componente attivo: nessun metodo pubblico oltre a run(), imposto da Thread
class SortingWorker extends Thread {
  private final int[] slice;
  private final SortedParts out;
  SortingWorker(int[] slice, SortedParts out) { this.slice = slice; this.out = out; }

  @Override public void run() { Arrays.sort(slice); out.add(slice); }
}

class MergingWorker extends Thread {
  private final SortedParts in;
  MergingWorker(SortedParts in) { this.in = in; }

  @Override public void run() {
    try { int[] all = merge(in.awaitAll()); /* usa il risultato */ }
    catch (InterruptedException e) { Thread.currentThread().interrupt(); }
  }
  // abilità privata: a ogni passo prende il più piccolo tra i primi elementi delle parti
  private int[] merge(List<int[]> parts) {
    int[] out = new int[parts.stream().mapToInt(p -> p.length).sum()];
    int[] next = new int[parts.size()];          // indice del prossimo elemento di ogni parte
    for (int k = 0; k < out.length; k++) {
      int best = -1;
      for (int i = 0; i < parts.size(); i++)
        if (next[i] < parts.get(i).length
            && (best < 0 || parts.get(i)[next[i]] < parts.get(best)[next[best]])) best = i;
      out[k] = parts.get(best)[next[best]++];
    }
    return out;
  }
}

Si vede a colpo d'occhio chi fa cosa, e merge, privato, lo esegue solo il merger. Dare un nome esplicito a ruoli e componenti, invece di nasconderli in una lambda, rende visibile il progetto nel codice.

3. L'analisi: decomporre e trovare le dipendenze

L'analisi ha due fasi: la decomposizione in task e dati, poi l'analisi delle dipendenze. Per la decomposizione ci sono due pattern, che differiscono per il punto di partenza.

L'analisi delle dipendenze esamina le relazioni tra i task per raggrupparli e ordinarli, in modo da massimizzare le prestazioni. Le dipendenze sono di due tipi:

• temporali: un task può iniziare solo dopo che un altro è finito. Il merge viene dopo tutti gli ordinamenti; il passo k + 1 di una simulazione viene dopo il passo k;
• sui dati o sulle risorse: un task usa dati prodotti da un altro, o più task condividono una struttura o una risorsa che va protetta.

Task del tutto indipendenti vanno ad agenti che non si coordinano. Dove ci sono dipendenze, gli agenti si coordinano con un componente passivo: una barriera o un latch per l'ordine, un monitor per i dati condivisi. Malone e Crowston (1993) classificano le dipendenze nei problemi di coordinazione, un tema interdisciplinare.

4. Tre modi di pensare il parallelismo

Per passare dall'analisi al progetto, Carriero e Gelernter (1989) propongono tre classi architetturali concettuali. L'esempio è costruire una casa.

In tutte e tre gli agenti lavorano in parallelo fin dove lo consentono i vincoli del problema. Risultato e specialisti sono approcci opposti: il primo divide il prodotto, il secondo i mestieri, e lì pesa la coordinazione tra i lavoratori, per esempio in pipeline.

I confini tra le tre classi non sono rigidi. Un lavoro reale mescola spesso più approcci: un progetto a specialisti può assegnare una squadra di lavoratori a una specialità, e quella squadra lavora ad agenda. Restano però tre modi di pensare nettamente distinti. Il risultato guarda alla forma del prodotto finito, gli specialisti alla composizione della squadra, l'agenda alla lista delle cose da fare. L'approccio si può applicare ricorsivamente, seguendo la decomposizione. Il metodo suggerito ha tre passi:

1. scegliere la classe concettuale più naturale per il problema;
2. scrivere il programma con il metodo più naturale per quella classe;
3. se il programma non è abbastanza efficiente, trasformarlo metodicamente in una versione più efficiente.

5. Le architetture concorrenti

Tra la classe concettuale e l'implementazione fanno da ponte quattro architetture di riferimento, tratte da Magee e Kramer e da Ortega-Arjona: master-workers (con i framework a task e la variante fork-join), filter-pipeline (pattern a flussi), shared-space o blackboard (coordinazione tramite uno spazio) e announcer-listeners (coordinazione a eventi). Nelle figure i cerchi sono agenti e i rettangoli componenti passivi.

Master-workers

Un agente master e un insieme di agenti worker, anche dinamico, interagiscono attraverso un mezzo di coordinazione che funziona da bag of tasks, di solito una blackboard o un bounded buffer. Il master scompone il task globale in sottotask, li assegna inserendone la descrizione nel bag e raccoglie i risultati. Ogni worker preleva un task dal bag, lo esegue e ne comunica il risultato. In una variante i risultati si aggregano in un monitor apposito. Esempio: la quadratura, cioè l'integrale definito calcolato come somma delle aree di tanti trapezi, uno per task.

Nella variante semplice il master conosce i worker e crea ciascuno con la sua parte, come nell'ordinamento della sezione 2. In quella generale il bag disaccoppia master e worker, il cui numero dipende dall'hardware e non dai task: il prodotto di due matrici 1000 × 1000 genera un milione di task, ma su 16 core un compito CPU-bound vuole 17 worker.

I framework a task sono master-workers

Un executor è un master-workers di cui il programmatore scrive solo il master: il bag (di solito un bounded buffer) e il pool di worker, cioè la politica di esecuzione, sono nascosti in un componente passivo. submit ritorna subito, e il task è eseguito in modo asincrono rispetto a chi lo sottomette. I risultati si raccolgono in un bag dei risultati, in un monitor aggregatore, oppure con le future: oggetti restituiti subito che rappresentano il risultato futuro, e permettono di controllare se è pronto, di bloccarsi fino al suo arrivo, di cancellare il task e di intercettarne gli errori. Task e future sono asincroni ma ammettono ancora attese bloccanti, a differenza della programmazione a eventi (capitolo 12).

I task di un executor devono essere indipendenti: bloccare un task significa bloccare il thread del pool che lo esegue. Cento task che attendono tutti su una barriera da 100, con un pool di 16 thread, vanno in deadlock: i primi 16 occupano tutti i thread in attesa, e gli altri 84 non partono mai. Le sezioni critiche brevi invece vanno bene, perché il lock prima o poi si rilascia.

La domanda d'orale su questo caso (pool da 4 thread, 100 task) è svolta, con un laboratorio, nel capitolo 10.

Le slide collegano ai task anche i virtual thread, thread logici gestiti dalla JVM e slegati da quelli del sistema operativo, e la structured concurrency, che tratta un gruppo di task correlati come un'unica unità di lavoro per semplificare errori e cancellazione (capitolo 10).

Fork-join

È il master-workers applicato ricorsivamente: i worker sono a loro volta master, che producono sottotask e ne raccolgono i risultati. Realizza in concorrenza il divide et impera: un worker con un compito troppo grande lo divide, finché i pezzi si possono eseguire direttamente. Si realizza con un bag of tasks in cui anche i worker inseriscono task, oppure direttamente. Esempi: la quadratura adattiva e il merge sort. In Java c'è ForkJoinPool, dove un task può attendere i propri sottotask con join (il thread in attesa esegue altri task invece di restare fermo), cosa che negli executor normali rischia il deadlock descritto sopra.

Filter-pipeline

Una catena lineare di agenti collegati da pipe: bounded buffer o canali. Il generatore apre la catena e produce i dati da elaborare. Ogni filtro intermedio consuma dalla pipe in ingresso e produce nella pipe in uscita. Il sink chiude la catena e raccoglie i risultati. Esempio: l'elaborazione di immagini, con uno stadio per ogni trasformazione. Il parallelismo viene dal fatto che gli stadi lavorano insieme su elementi diversi del flusso.

Shared-space (blackboard)

Gli agenti interagiscono indirettamente attraverso uno spazio condiviso che funziona da lavagna (blackboard). Lo spazio offre operazioni per aggiungere informazioni e per cercarle, leggerle o rimuoverle, spesso con sincronizzazione: una lettura può sospendersi finché l'informazione cercata non c'è. La coordinazione è quindi guidata dai dati, perché un agente reagisce a ciò che compare nello spazio e non a chi lo ha scritto, e c'è disaccoppiamento temporale: chi scrive e chi legge non devono esistere nello stesso momento. Il riferimento classico sono gli spazi di tuple del linguaggio di coordinazione Linda (Gelernter, 1985).

Announcer-listeners

Un agente announcer e un insieme dinamico di agenti listener interagiscono attraverso un servizio di eventi. L'announcer annuncia sul servizio che un evento è accaduto. I listener si registrano sul servizio per essere avvisati degli eventi che li interessano. Il servizio disaccoppia i due ruoli, raccoglie gli eventi e li smista. Esempio: parallelizzare un'applicazione a eventi eseguendo in parallelo i suoi gestori di eventi.

L'ultima colonna è indicativa: le slide citano blackboard e servizi di eventi tra i mezzi degli specialisti, e la pipeline come loro strategia di lavoro.

6. Come si capisce che il lavoro è finito

Il master deve sapere quando tutti i task sono stati eseguiti, per aggregare i risultati, passare alla fase successiva o terminare. La tecnica dipende da come nascono i task.

• Numero di task noto in anticipo. Si contano i completamenti. Un monitor aggregatore attende finché non ha N risultati, come SortedParts nella sezione 2. Oppure un latch (il «cancello») inizializzato a N: ogni worker fa countDown e il master aspetta su await. Se ogni worker ha un solo task, basta la join sui thread.
• Con un executor. Si conservano le future restituite da submit e si chiama get su ciascuna, o si usa invokeAll. Se l'executor non serve più, si chiama shutdown e poi awaitTermination, che ritorna quando tutti i task sono finiti.
• Lavoro a fasi, come i passi di una simulazione. Il lavoro di ogni passo è noto, e si usa una barriera o un latch per passo: il passo k + 1, o il rendering, parte solo quando tutti i worker hanno finito il passo k.
• Task generati dinamicamente, come nel fork-join o in una visita di directory. Un bag vuoto non vuol dire lavoro finito, perché un worker ancora attivo può inserire sottotask. Serve un contatore dei task pendenti, incrementato a ogni inserimento e decrementato quando un task è completato (dopo che ha inserito i suoi sottotask): il lavoro è finito quando arriva a zero.

interface Task { void execute(BagOfTasks<Task> bag); }

class BagOfTasks<T> {
  private final Deque<T> tasks = new ArrayDeque<>();
  private int pending;                          // inseriti e non ancora completati

  public synchronized void put(T task) { tasks.addLast(task); pending++; notifyAll(); }

  /** Il prossimo task, oppure empty quando tutto il lavoro è finito. */
  public synchronized Optional<T> take() throws InterruptedException {
    while (tasks.isEmpty() && pending > 0) wait();
    return Optional.ofNullable(tasks.pollFirst());
  }

  /** Il worker lo chiama dopo aver completato un task e inserito i suoi sottotask. */
  public synchronized void done() { if (--pending == 0) notifyAll(); }

  public synchronized void awaitCompletion() throws InterruptedException {
    while (pending > 0) wait();
  }
}

// ciclo di ogni worker
for (Optional<Task> t = bag.take(); t.isPresent(); t = bag.take()) {
  try { t.get().execute(bag); }                 // può chiamare bag.put(sottotask)
  finally { bag.done(); }
}

Il master inserisce i task iniziali prima di avviare i worker e poi aspetta con awaitCompletion. A zero, take restituisce empty a tutti i worker, che terminano. Se i worker leggono da un semplice bounded buffer, il modo classico per fermarli è la poison pill: un elemento speciale, uno per worker, che significa «fine» (capitolo 10).

7. Model-View-Controller in un programma concorrente

Model-View-Controller separa il dominio (model), la presentazione (view), che osserva il model con il pattern observer, e la gestione dell'input (controller). Con componenti attivi e passivi diventa:

• il controllo sta negli agenti, i componenti attivi che incapsulano la logica di controllo;
• il model è fatto di componenti passivi condivisi, usati dagli agenti. Sono monitor che incapsulano la mutua esclusione, e ne fanno parte anche i mezzi di coordinazione;
• la view è fatta di componenti passivi condivisi che rendono possibile e mediano l'interazione con l'utente. Gli agenti la osservano e la usano, e la view può leggere lo stato dal model.

Due questioni restano aperte:

1. Le GUI hanno un solo componente attivo. I toolkit grafici affidano tutta la GUI a un thread, in Swing l'event dispatch thread, e i componenti non sono thread-safe. Quindi un listener non esegue logica di controllo lunga, che bloccherebbe l'EDT, ma la passa a un agente; e un agente aggiorna la view passando l'aggiornamento all'EDT con SwingUtilities.invokeLater, per evitare race condition, non per la reattività. Dettagli ed esempio del cronometro nel capitolo 10.
2. L'observer originale è sincrono e accoppia i flussi di controllo: notificare vuol dire chiamare il metodo dell'osservatore nel thread di chi notifica. In un sistema concorrente lo si ripensa come produttore-consumatore. Un bounded buffer fa da coda di eventi, chi genera gli eventi (per esempio la view, a ogni tasto premuto) è il produttore, e un agente che li preleva e reagisce, un controller attivo, è il consumatore. La coda è un componente passivo a sé, non un metodo pubblico del controller: altrimenti il controller sarebbe un'entità ibrida.

Verifica le tue conoscenze