Concorrenza in Java: thread, sincronizzazione, task ed executor

1. Come si crea e si controlla un thread

I meccanismi del linguaggio sono pochi. Thread e Runnable creano e controllano le attività; synchronized e volatile regolano il codice eseguito da più thread (sezioni 2 e 3); wait, notify e notifyAll di Object li coordinano (sezione 4). Sopra ci sono la libreria java.util.concurrent e, da JDK 21, i virtual thread (sezione 12). Un thread normale, o platform thread, corrisponde uno a uno a un thread del sistema operativo.

Un thread si definisce estendendo Thread e ridefinendo run, oppure passando al costruttore di Thread un Runnable, di solito una lambda. run contiene il comportamento; il thread termina quando run finisce.

public class MyWorker extends Thread {
  public MyWorker(String name) { super(name); }
  @Override public void run() {
    System.out.println(getName() + ": al lavoro");   // comportamento attivo
  }
}

public static void main(String[] args) throws InterruptedException {
  Thread a = new MyWorker("worker-A");
  Thread b = new Thread(() -> System.out.println("worker-B: al lavoro"), "worker-B");
  a.start();        // ritorna subito: da qui ci sono tre flussi (main, A, B)
  b.start();
  a.join();         // main si sospende finché A non termina
  b.join();
  System.out.println("entrambi terminati");
}

start, non run. start fa nascere un nuovo thread che esegue run, e ritorna subito. Chiamare run direttamente è una normale chiamata di metodo: gira nel thread chiamante, in sequenza. Un Thread si avvia una sola volta (un secondo start lancia IllegalThreadStateException), e ogni programma ha almeno il main thread.

stop, suspend, resume e destroy sono deprecati: fermare un thread dall'esterno può coglierlo a metà di un aggiornamento o con un lock in mano. Si usano schemi cooperativi (sezione 9).

Thread o Runnable? Il funzionamento è identico. Runnable serve quando la classe deve già estendere un'altra classe, perché Java non ha ereditarietà multipla. Ricci preferisce una classe con un nome quando il thread è un concetto del progetto (SortingWorker, MergingWorker): la lambda è più corta, ma lo nasconde. Il corso chiede anche una disciplina. Un thread è un oggetto attivo: i suoi metodi li esegue solo lui, quindi ha pochi metodi pubblici, e interagisce con gli altri tramite oggetti passivi condivisi, non chiamando i loro metodi (capitolo 8).

2. Come si protegge lo stato condiviso: synchronized e lock

Una classe è thread-safe se resta corretta, cioè conforme alla sua specifica (invarianti e postcondizioni), quando più thread la usano, con qualunque interleaving e senza sincronizzazione aggiuntiva da parte del chiamante. Il problema è lo stato condiviso e mutabile. I rimedi sono tre: non condividerlo, renderlo immutabile, oppure sincronizzare ogni accesso. Un oggetto senza stato è sempre thread-safe.

Una race condition si ha quando la correttezza dipende dai tempi relativi dei thread. I due casi tipici sono azioni composte eseguite senza atomicità:

• lost update: un leggi-modifica-scrivi come count++. Due thread leggono 5, entrambi scrivono 6, un incremento si perde. Con 100 incrementi per thread esce quasi sempre 200; con 10.000 esce un valore diverso a ogni esecuzione. Un test che vede 200 non dimostra nulla (capitolo 11);
• check-and-act: una decisione presa su un'osservazione che, al momento dell'azione, può essere vecchia. «Se il file non esiste, crealo»; oppure il singleton pigro, di cui due thread creano due istanze.

synchronized: il lock intrinseco

Ogni oggetto ha un lock intrinseco. synchronized (obj) { … } lo prende all'ingresso e lo rilascia all'uscita, anche per eccezione; se è occupato, il thread si sospende nell'entry set. Al rilascio uno dei thread in attesa riparte, ma non è specificato quale: l'entry set è un insieme, non una coda, e la starvation è possibile. L'esclusione vale solo tra blocchi sullo stesso oggetto. Un metodo synchronized equivale a synchronized (this); per un metodo static il lock è l'oggetto Class. Rendere synchronized tutti i metodi pubblici è l'idioma del monitor (capitolo 7).

class Counter {                        // NON thread-safe
  private int count;
  public void inc() { count++; }       // leggi, somma, scrivi: tre passi
  public int get() { return count; }
}

class SafeCounter {                    // thread-safe: idioma del monitor
  private int count;
  public synchronized void inc() { count++; }    // = synchronized (this) { count++; }
  public synchronized int get() { return count; } // anche la lettura: visibilità (sezione 3)
}

L'idioma rende atomico ogni metodo, non una sequenza di chiamate. Nel laboratorio un BoundedCounter tra 0 e 1, con tutti i metodi synchronized, va in underflow: due worker leggono 1 ed entrambi decrementano. La coppia va resa atomica con un blocco sullo stesso lock, o con un metodo apposito.

if (counter.getVal() > 0) counter.dec();        // SBAGLIATO: check-and-act non atomico

synchronized (counter) {                         // stesso lock usato dai metodi della classe
  if (counter.getVal() > 0) counter.dec();      // getVal e dec riprendono il lock: rientranza
}                                                // (meglio: un metodo synchronized tryDec())

Rientranza. Un lock è rientrante se è acquisito per thread e non per invocazione: chi lo possiede già può riprenderlo senza bloccarsi, e lo libera dopo un rilascio per ogni acquisizione. I lock intrinseci e ReentrantLock lo sono; i mutex dei Pthread e i semafori no. Senza rientranza il blocco qui sopra, o un LoggingWidget il cui metodo synchronized chiama super.doSomething(), anch'esso synchronized, si bloccherebbe su se stesso.

Costo. Un lock non conteso costa poco. Uno conteso richiede di sospendere il thread nel sistema operativo: in laboratorio, un miliardo di incrementi con due thread in competizione diventa più lento di uno o due ordini di grandezza. Regola: tenere il lock il meno possibile, ma mai a scapito della correttezza.

Lock espliciti

Lock, implementata da ReentrantLock, offre lock, unlock, lockInterruptibly, tryLock(), tryLock(timeout, unit) e newCondition().

lock.lock();          // fuori dal try: se l'acquisizione fallisce non c'è nulla da rilasciare
try {
  count++;
} finally {
  lock.unlock();      // sempre, anche se il corpo lancia un'eccezione
}

Quanto alle prestazioni, in Java 5 ReentrantLock rendeva circa 4 volte di più; da Java 6 le due soluzioni si equivalgono.

3. Che cosa garantisce volatile

Oltre all'atomicità c'è la visibilità. Senza sincronizzazione nulla garantisce che un thread veda le scritture di un altro, né in che ordine: il compilatore JIT può tenere una variabile in un registro, la CPU nella cache, ed entrambi riordinano. Un thread che cicla su while (!stopped) può non vedere mai stopped = true.

Su un campo volatile ogni lettura vede l'ultima scrittura. In più la scrittura happens-before le letture successive: chi legge il nuovo valore vede anche tutto ciò che lo scrittore aveva scritto prima, quindi volatile vieta anche i riordinamenti attorno all'accesso. Le slide dicono che il valore «viene letto e scritto in memoria a ogni accesso»: un'approssimazione che coglie l'effetto.

class Clock extends Thread {
  private volatile boolean stopped;     // senza volatile il ciclo potrebbe non vedere mai true
  @Override public void run() {
    while (!stopped) { tick(); }
  }
  public void notifyStop() { stopped = true; }   // chiamato da un altro thread
}

Leggere o scrivere un int, un boolean o un riferimento è comunque atomico. Per long e double non volatile la specifica ammette una scrittura in due metà da 32 bit; volatile la rende atomica. Anche synchronized dà visibilità, perché il rilascio di un lock happens-before la successiva acquisizione dello stesso lock: per questo in SafeCounter anche get è sincronizzato. volatile basta quando un thread scrive e gli altri leggono, senza decisioni che combinano lettura e scrittura: flag di stop e di cancellazione, pubblicazione di un riferimento, le variabili di Peterson (capitolo 4).

4. Come si coordinano i thread: wait/notify e i sincronizzatori

wait, notify e notifyAll realizzano un monitor con una sola variabile condizione implicita, l'oggetto stesso. Si chiamano solo tenendo il lock dell'oggetto, altrimenti si ottiene IllegalMonitorStateException. wait() rilascia il lock e sospende il thread. notify() risveglia un thread in attesa, scelto arbitrariamente, e se non c'è nessuno la notifica si perde; notifyAll() li risveglia tutti. Chi si risveglia deve riprendere il lock, e intanto la condizione può essere tornata falsa.

Quando viene lanciata InterruptedException? Quando un thread è sospeso in un'attesa interrompibile e un altro thread chiama interrupt() su di esso, oppure se l'interruzione era già stata chiesta prima dell'attesa. Il metodo lancia l'eccezione e lo stato di interruzione viene azzerato. Sono interrompibili sleep, wait, join e i metodi bloccanti della libreria (BlockingQueue.put/take, lockInterruptibly, Future.get, await di latch e barriere, Semaphore.acquire). Non lo sono l'attesa di un blocco synchronized e Lock.lock(). Che cosa fare dell'eccezione è nella sezione 9.

I mattoni della libreria

• Collezioni sincronizzate (Collections.synchronizedList…): ogni metodo è sincronizzato, ma le azioni composte (iterazione, «inserisci se assente») richiedono un lock lato client sulla collezione, e tenerlo a lungo blocca tutti.
• Collezioni concorrenti (ConcurrentHashMap, CopyOnWriteArrayList, code): pensate per l'accesso concorrente, scalano molto meglio.
• BlockingQueue: put si blocca se la coda è piena, take se è vuota (offer e poll con timeout). Limitata, è il mattone del produttore-consumatore e frena chi produce più lavoro di quanto il sistema smaltisca. LinkedBlockingQueue, ArrayBlockingQueue, PriorityBlockingQueue.
• Deque e work stealing (ArrayDeque, LinkedBlockingDeque): ogni consumatore ha la sua deque, e quando è vuota ruba lavoro dalla coda di quella di un altro. Scala meglio di una coda unica, perché di solito ognuno tocca solo la propria.

Un sincronizzatore è un oggetto passivo che coordina i thread in base al proprio stato: lo stato decide chi passa e chi aspetta, alcuni metodi lo modificano, altri aspettano che diventi quello voluto. La libreria ne ha quattro tipi: i lock (sezione 2); i semafori (Semaphore, con permessi, acquire/release e fairness opzionale, capitolo 5); i latch (CountDownLatch, un cancello che si apre quando i countDown portano il conteggio a zero e resta aperto per sempre); le barriere (CyclicBarrier, in cui ognuno aspetta finché non sono arrivati tutti, e poi la barriera si riarma). Come dicono le slide, i latch aspettano eventi, le barriere aspettano altri thread.

CountDownLatch ready = new CountDownLatch(services.size());
for (Service s : services)
  new Thread(() -> { s.start(); ready.countDown(); }).start();
ready.await();                  // passa quando il conteggio è 0; il cancello resta aperto

CyclicBarrier phase = new CyclicBarrier(nWorkers, () -> System.out.println("fase finita"));
// in ogni worker, a ogni iterazione:
computeStep();
phase.await();                  // tutti aspettano tutti, poi la barriera si riarma

Come si costruiscono una barriera e un latch con un monitor, esercizio del laboratorio, è nel capitolo 7.

5. Perché ragionare per task: executor e thread pool

Un task è un'unità di lavoro astratta, discreta e indipendente, con un inizio e una fine. Non è un ciclo infinito: i componenti che vivono per sempre, come un server o un agente, restano thread. Il thread diventa il meccanismo che esegue i task in modo asincrono, e il livello logico (i task, che vengono dal problema) si separa da quello fisico (i thread, che dipendono dalla macchina). Le slide chiamano la strategia Division of Labor: si individuano i task come attività che non dipendono da stato, risultati o effetti di altri task, e si sceglie una politica di esecuzione che dia buon throughput, buona reattività e degrado graduale sotto carico. Come scomporre un problema in task è il tema del capitolo 8.

class TaskExecutionWebServer {
  private static final int NTHREADS = Runtime.getRuntime().availableProcessors() + 1;
  private static final Executor exec = Executors.newFixedThreadPool(NTHREADS);

  public static void main(String[] args) throws IOException {
    ServerSocket socket = new ServerSocket(80);
    while (true) {
      Socket connection = socket.accept();
      exec.execute(() -> handleRequest(connection));   // consegna il task e torna ad accettare
    }
  }
}

L'Executor framework (JDK 5) separa la sottomissione di un task dalla sua esecuzione. Dentro c'è un produttore-consumatore: chi sottomette produce task, i thread del pool li consumano con un ciclo «prendi il prossimo, eseguilo, torna ad aspettare». È l'architettura master-worker con una bag of tasks: il master scompone il lavoro e mette i task in una struttura condivisa (un monitor, come un bounded buffer), da cui i worker li prelevano. Il master non sa quanti worker ci sono, e i thread restano nascosti nella politica dell'executor. Nel prodotto di due matrici n×n ci sono n² task, uno per elemento, ma solo N_cpu+1 thread.

Ciclo di vita

Un ExecutorService nasce running. shutdown() lo porta in shutting down in modo graduale: niente task nuovi, ma quelli già sottomessi, anche in coda, vengono eseguiti. shutdownNow() è brusco: interrompe i task in esecuzione, non avvia quelli in coda e li restituisce. I task sottomessi dopo la chiusura vanno al rejected execution handler (per default, RejectedExecutionException). Finiti tutti i task, l'executor è terminated. Per aspettare questo momento si usa awaitTermination. Creare un executor crea i suoi thread, quindi si crea una volta e si riusa: mai dentro un ciclo, e mai nel game loop dell'assignment.

exec.shutdown();                                   // niente nuovi task, finisci i vecchi
if (!exec.awaitTermination(1, TimeUnit.MINUTES))   // sospende il chiamante (al massimo un minuto)
  exec.shutdownNow();                              // non hanno finito: interrompi

6. Come si ottiene il risultato di un task: Callable e Future

Un task con risultato è un Callable<V>: call() restituisce un V e può lanciare eccezioni. Si sottomette con submit, che restituisce subito una Future<V>: un oggetto che rappresenta un risultato non ancora disponibile. isDone() e isCancelled() ne dicono lo stato; get() blocca finché il risultato non c'è e lo restituisce, oppure lancia ExecutionException (che avvolge l'eccezione del task), CancellationException o InterruptedException; get(timeout, unit) aspetta al massimo un certo tempo; cancel(mayInterruptIfRunning) cancella il task. Una Future è thread-safe, ma attenzione al check-and-act: se isDone() dice sì la risposta resta vera, se dice no un istante dopo il risultato può già esserci.

Lo schema tipico, nell'esempio del laboratorio sull'integrale definito calcolato con i trapezi, ha due fasi. Nell'espansione il master sottomette tutti i task senza bloccarsi e raccoglie le Future; nell'aggregazione chiama get su ciascuna. Non serve chiudere l'executor: ogni get sincronizza il master con la fine di quel task.

record ComputeAreaTask(DoubleUnaryOperator f, double x0, double dx) implements Callable<Double> {
  public Double call() { return (f.applyAsDouble(x0) + f.applyAsDouble(x0 + dx)) * dx / 2; }
}

double integrate(ExecutorService exec, DoubleUnaryOperator f, double a, double b, int n)
    throws InterruptedException, ExecutionException {
  double dx = (b - a) / n;
  List<Future<Double>> parts = new ArrayList<>();
  for (int i = 0; i < n; i++)                        // espansione: submit non blocca
    parts.add(exec.submit(new ComputeAreaTask(f, a + i * dx, dx)));
  double sum = 0;
  for (Future<Double> p : parts) sum += p.get();     // aggregazione: get blocca su quel task
  return sum;
}

In alternativa i task possono depositare i risultati in un monitor. Se il monitor si limita a sommare, il master non sa quando i task hanno finito e deve chiudere l'executor e aspettarne la terminazione. Se invece conta i contributi e la sua getResult() sospende il master finché non sono arrivati tutti, l'executor resta riutilizzabile. Il task può attendere il lock del monitor, ma per poco, perché nessuno lo tiene aspettando altro. Ricci consiglia comunque le Future.

7. Perché un task non deve aspettare un altro task

Le slide sono esplicite: nell'Executor framework i task devono essere indipendenti, e un task non può aspettare un altro task. Un task che si sospende su un semaforo, una barriera o la get di un'altra Future non sospende se stesso: sospende il thread del pool che lo esegue. Se si sospendono tutti i thread del pool, i task che li sbloccherebbero restano in coda senza nessuno che li esegua: è un deadlock, detto thread starvation deadlock.

È l'esempio che Ricci usa all'orale. Cento task eseguono ciascuno due operazioni atomiche, a e poi b, che stampano qualcosa. Una nuova specifica chiede che tutte le b vengano dopo tutte le a, e si aggiunge una barriera per 100 tra le due. Il pool è fisso, di 4 thread.

ExecutorService exec = Executors.newFixedThreadPool(4);
CyclicBarrier allA = new CyclicBarrier(100);
for (int i = 0; i < 100; i++) {
  int id = i;
  exec.execute(() -> {
    System.out.println("a" + id);
    try { allA.await(); }          // sospende il THREAD del pool, non solo il task
    catch (InterruptedException | BrokenBarrierException e) { return; }
    System.out.println("b" + id);
  });
}

Un pool di 100 thread lo «risolve», ma Ricci la chiama barare: il numero di thread torna a dipendere dal problema (con 10.000 task, 10.000 thread), proprio l'accoppiamento che l'executor doveva togliere. La correzione giusta è togliere l'attesa dai task, qui dividendo il lavoro in due fasi e lasciando l'attesa al master:

List<Future<?>> phaseA = new ArrayList<>();
for (int i = 0; i < 100; i++) { int id = i; phaseA.add(exec.submit(() -> System.out.println("a" + id))); }
for (Future<?> f : phaseA) f.get();      // l'attesa è nel master, fuori dal pool
for (int i = 0; i < 100; i++) { int id = i; exec.execute(() -> System.out.println("b" + id)); }

Una breve sezione critica dentro un task invece va bene: il lock viene preso e rilasciato senza aspettare altri task. Se i task devono davvero aspettare i propri sottotask, si usa Fork/Join (sezione 10).

Laboratorio: il pool che si blocca

Una coda FIFO di task e T thread: a ogni passo un thread libero prende un task, esegue a, arriva alla barriera o esegue b. Prova i cinque scenari. Con una barriera ciclica da 4 su 4 thread le stampe procedono a gruppi (a0 a1 a2 a3 b0 b1 b2 b3 a4 …) e tutto termina; con una da 5 si torna al blocco. Cambia i numeri e trova la regola.

8. Quanti thread: prestazioni e dimensionamento del pool

Più thread migliorano l'uso delle risorse e la reattività, ma costano: coordinazione (lock, segnalazioni, sincronizzazione della memoria), cambi di contesto, creazione e distruzione, scheduling. Un programma è CPU-bound, I/O-bound o DB-bound secondo la risorsa che lo limita. Le misure sono tre: throughput (task completati per unità di tempo), reattività o latenza (tempo tra richiesta e completamento di un'azione), scalabilità (quanto cresce il throughput aggiungendo risorse, di solito CPU).

Gli obiettivi sono usare bene le risorse esistenti e poter sfruttare quelle aggiunte. In pratica: tenere le CPU occupate con lavoro utile, perché un core fermo mentre c'è lavoro è capacità sprecata. Da qui la domanda su quanti thread mettere in un pool. La risposta dipende dal tipo di task e dal sistema, e non va scritta nel codice: si parte da Runtime.getRuntime().availableProcessors().

Task CPU-bound: N_cpu + 1

Con task di solo calcolo, N_cpu processori raggiungono di solito l'utilizzo ottimo con N_cpu + 1 thread. Quello in più serve perché anche un thread di calcolo ogni tanto si ferma, ad esempio per un page fault: la pagina che gli serve non è in memoria e il sistema operativo lo sospende per caricarla, una forma implicita di I/O. In quel momento il thread in più tiene il core occupato. Con 16 core, 17 thread.

Task con I/O o attese

Se i task si bloccano, il pool deve essere più grande, perché in ogni istante una parte dei thread non può usare la CPU. Il numero si trova in modo empirico, provando pool di dimensioni diverse sotto un carico di benchmark e osservando l'utilizzo della CPU, oppure con la formula

N_threads = N_cpu · U_cpu · (1 + W/C)

dove U_cpu è l'utilizzo di CPU voluto (tra 0 e 1) e W/C il rapporto tra tempo di attesa e tempo di calcolo di un task, stimato con un profiler. Con 8 core, CPU al 100% e task che aspettano 30 ms ogni 10 ms di calcolo (W/C = 3), servono 8 · 1 · 4 = 32 thread. Con W/C = 0 la formula dà N_cpu; il «+1» è il margine empirico.

Serializzazioni e contesa

Le parti serializzate limitano lo speed-up (legge di Amdahl, capitolo 1) e spesso sono nascoste nelle librerie: una LinkedList sincronizzata serializza gli accessi, una ConcurrentLinkedQueue no. Un cambio di contesto costa 5.000–10.000 cicli, qualche microsecondo (dato delle slide, 2008). La sincronizzazione non contesa resta nella JVM ed è veloce; quella contesa può coinvolgere il sistema operativo, ed è lì che si ottimizza. La contesa si riduce tenendo i lock per meno tempo (ambito più stretto, granularità più fine), chiedendoli meno spesso, oppure sostituendo i lock esclusivi con meccanismi più permissivi come i lock lettori-scrittori. Una tecnica tipica è il lock splitting o striping: lock diversi per variabili di stato indipendenti.

9. Come si ferma un thread o un task

Un'attività è cancellabile se codice esterno può farla terminare prima del tempo. Le slide elencano cinque motivi: richiesta dell'utente, limite di tempo, eventi applicativi (è stata trovata la prima soluzione), errori, spegnimento dell'applicazione. Fermare un thread dall'esterno in un punto qualsiasi, cioè l'approccio asincrono dei metodi deprecati, rompe la consistenza dello stato. Java adotta l'approccio cooperativo: si segnala la richiesta e il task la gestisce secondo la sua politica di cancellazione, che dice come si chiede lo stop, quando il task lo controlla e che cosa fa.

La forma più semplice è un flag volatile, come nel PrimeGenerator delle slide, che controlla cancelled a ogni iterazione. Ma se il generatore mette i primi in un buffer limitato e il buffer è pieno, il thread resta bloccato in put e il flag non viene più letto. Serve l'interruzione:

• ogni thread ha uno stato di interruzione: interrupt() lo imposta, isInterrupted() lo legge, Thread.interrupted() lo legge e lo azzera;
• le attese interrompibili (sezione 4) si sbloccano, azzerano lo stato e lanciano InterruptedException;
• interrupt non ferma nulla: consegna la richiesta. La politica di interruzione del thread ha quattro elementi: uno stato di shutdown esplicito, un metodo cancel, il controllo dell'interruzione nel ciclo, la gestione di InterruptedException.

class PrimeProducer extends Thread {
  private final BlockingQueue<BigInteger> queue;
  PrimeProducer(BlockingQueue<BigInteger> queue) { this.queue = queue; }

  @Override public void run() {
    try {
      BigInteger p = BigInteger.ONE;
      while (!isInterrupted()) {
        p = p.nextProbablePrime();
        queue.put(p);                  // a coda piena si blocca, ma interrupt() lo sblocca
      }
    } catch (InterruptedException e) {
      // politica di questo thread: uscire
    }
  }
  public void cancel() { interrupt(); }
}

Un'alternativa è la poison pill: un oggetto riconoscibile, ad esempio un elemento «stop», messo nel buffer per dire al consumatore di terminare. Negli executor, shutdownNow() e Future.cancel(true) fermano i task in corso interrompendone i thread: la Future risulta subito cancellata, ma il corpo del task esce solo quando reagisce all'interruzione.

10. Task che generano task: Fork/Join e structured concurrency

Negli algoritmi map-reduce o divide et impera si dividono i dati in pezzi indipendenti e si combinano i risultati parziali. Se la struttura dei dati è nota in anticipo, come nella somma di un array, bastano executor e Callable. Se non lo è, come in un albero, un grafo o un file system, un task deve creare sottotask e aspettarli; con un executor normale occuperebbe un thread senza fare nulla, fino al deadlock della sezione 7.

Per questo Java SE 7 ha il framework Fork/Join. Un ForkJoinPool esegue ForkJoinTask: fork() lancia un sottotask sullo stesso pool, join() lo aspetta e ne restituisce il risultato, invokeAll lancia un gruppo e aspetta tutti. RecursiveAction non ha risultato, RecursiveTask<V> sì. Quando un task aspetta e ci sono task pendenti, il thread non resta fermo ma ruba lavoro dalla deque di un altro worker (work stealing): la join blocca il task, non il pool.

L'esempio del laboratorio, tratto dall'articolo di Julien Ponge su Oracle Technology Network, conta le occorrenze di una parola nei documenti di una cartella e delle sue sottocartelle. Il FolderSearchTask di una cartella crea con fork un FolderSearchTask per ogni sottocartella e un DocumentSearchTask per ogni documento, poi somma i risultati con join. Il task di un documento, quindi, non lo crea un master centrale ma il task della cartella che lo contiene: sia l'espansione sia l'aggregazione sono ricorsive e distribuite.

class FolderSearchTask extends RecursiveTask<Long> {
  private final Folder folder; private final String word;
  FolderSearchTask(Folder folder, String word) { this.folder = folder; this.word = word; }

  @Override protected Long compute() {
    List<RecursiveTask<Long>> forks = new ArrayList<>();
    for (Folder sub : folder.getSubFolders()) {
      FolderSearchTask t = new FolderSearchTask(sub, word);
      forks.add(t); t.fork();                         // sottotask ricorsivo
    }
    for (Document doc : folder.getDocuments()) {
      DocumentSearchTask t = new DocumentSearchTask(doc, word);
      forks.add(t); t.fork();                         // un task per documento
    }
    long count = 0;
    for (RecursiveTask<Long> t : forks) count += t.join();   // aggregazione
    return count;
  }
}
// avvio: long n = new ForkJoinPool().invoke(new FolderSearchTask(root, "java"));

Structured concurrency

Estensione recente dell'approccio a task (JEP 437, JDK 20, in incubazione), per conservare leggibilità e osservabilità del codice sequenziale. Il principio: se un task si divide in sottotask concorrenti, questi tornano tutti allo stesso punto, il blocco di codice del task. StructuredTaskScope tratta i sottotask come un'unità: ognuno gira nel proprio thread (virtuale), si avviano uno per uno, si aspettano e se serve si cancellano insieme, e il padre raccoglie risultati ed eccezioni. La loro vita è confinata nel blocco.

Response handle() throws ExecutionException, InterruptedException {   // versione delle slide (JDK 20)
  try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
    Future<String>  user  = scope.fork(() -> findUser());
    Future<Integer> order = scope.fork(() -> fetchOrder());
    scope.join();            // aspetta entrambi i sottotask
    scope.throwIfFailed();   // se uno è fallito propaga l'errore (e l'altro viene cancellato)
    return new Response(user.resultNow(), order.resultNow());
  }                          // all'uscita dal blocco nessun sottotask è ancora vivo
}

11. Chi esegue il codice di una GUI: l'Event Dispatch Thread

Quasi tutti i framework grafici (Swing, Qt, Cocoa, X Window…) sono single-threaded: un thread dedicato, in Java l'Event Dispatch Thread (EDT), preleva gli eventi da una coda e li consegna agli handler. È un event loop. Le GUI multithread sono un «sogno fallito»: le azioni dell'utente risalgono dal sistema operativo al listener, quelle dell'applicazione scendono dal componente al sistema che disegna, ed entrambe prendono lock in ordine opposto, con race condition e deadlock.

La thread-safety si ottiene per confinamento: i componenti Swing e i loro modelli (TableModel, TreeModel) si usano solo dall'EDT. Fanno eccezione pochi metodi thread-safe: SwingUtilities.isEventDispatchThread(); invokeLater(r), che accoda r per l'EDT e ritorna subito; invokeAndWait(r), che aspetta l'esecuzione e non si può chiamare dall'EDT; le richieste di repaint e revalidate; l'aggiunta e la rimozione di listener, che però vengono sempre invocati nell'EDT.

Un task breve si esegue nell'EDT. Uno lungo va a un altro thread, meglio se di un pool che creato al volo; se deve aggiornare l'interfaccia, riconsegna l'aggiornamento all'EDT:

button.addActionListener(e -> {              // actionPerformed: lo esegue l'EDT
  button.setEnabled(false);
  label.setText("busy");
  exec.execute(() -> {                        // il lavoro lungo va a un thread del pool
    try {
      doBigComputation();
    } finally {
      SwingUtilities.invokeLater(() -> {      // l'aggiornamento torna all'EDT
        button.setEnabled(true);
        label.setText("idle");
      });
    }
  });
});

Esempio: il cronometro che non risponde

Nel cronometro MVC sbagliato del laboratorio (CronoMVCNotReactivePlusRace) il clic su Start arriva ad actionPerformed, che chiama notifyStarted() sul controller. Il controller entra in un ciclo: incrementa il modello e aspetta un secondo, finché un flag volatile non segnala lo stop. Il contatore avanza, come mostra il log, ma il display resta fermo e Stop non funziona. Il ciclo gira nell'EDT, perché è l'EDT che ha chiamato il controller. La setText del display accoda un ridisegno che l'EDT non eseguirà mai, e il clic su Stop resta in coda.

Nella versione corretta (CronoMVCReactiveNoRaces) l'EDT riceve solo i comandi. A Start il controller crea un agente, cioè un thread con il ciclo, e gli passa un flag di stop condiviso; a Stop alza il flag. Il modello, letto dall'agente e dalla view, è un monitor. La view riceve le notifiche dal thread dell'agente e riconsegna l'aggiornamento all'EDT:

public void notifyStarted() {                      // chiamato dall'EDT: deve ritornare subito
  stopFlag.reset();
  new CounterAgent(counter, stopFlag).start();     // il ciclo vive nel suo thread
}
public void notifyStopped() { stopFlag.set(); }    // l'EDT è libero, quindi Stop viene servito

// nella view, osservatore del modello: la chiama il thread dell'agente
public void counterChanged(int value) {
  SwingUtilities.invokeLater(() -> display.setText(String.valueOf(value)));
}

È il progetto proposto dalle slide per un MVC concorrente: il controllo negli agenti (componenti attivi), il modello fatto di componenti passivi condivisi che incapsulano la mutua esclusione, la view fatta di componenti passivi più un componente attivo, l'EDT.

12. Virtual thread

Un platform thread costa: il suo stack è un blocco unico di circa un megabyte, allocato alla creazione e non ridimensionabile (più piccolo si rischia StackOverflowError). Per questo i thread non possono essere tanti, e il modello più naturale per un server, un thread per richiesta, regge 1.000 richieste concorrenti ma non un milione, anche se CPU e banda basterebbero.

I virtual thread (Project Loom; JEP 425 in anteprima, definitivi con la JEP 444 in JDK 21) sono un'altra implementazione di java.lang.Thread. Gli stack frame stanno nello heap: partono da poche centinaia di byte e crescono o si riducono da soli. Come le pagine virtuali condividono la poca memoria fisica, i virtual thread, economici e numerosi, condividono i pochi platform thread, e gli stack di quelli inattivi sono «paginati» nello heap. Somigliano alle goroutine di Go e ai processi di Erlang, con la semantica dei thread Java.

Il sistema operativo conosce solo i platform thread. Per eseguire un virtual thread la JVM lo monta su un platform thread, il carrier, copiandone i frame sullo stack del carrier. Quando il virtual thread si bloccherebbe (I/O, un lock, una take, una sleep), viene smontato: i frame tornano nello heap e il carrier esegue un altro virtual thread. Quando l'operazione può finire, lo scheduler, un ForkJoinPool con un carrier per core, lo rimonta in ordine FIFO, magari su un altro carrier. Tutto è invisibile al codice: Thread.currentThread() restituisce il virtual thread e i lock vedono un'identità coerente.

L'API nuova è piccola: Thread.ofVirtual() (con start o unstarted) e Executors.newVirtualThreadPerTaskExecutor(). Dopo la creazione sono normali Thread: ThreadLocal, interruzione e stack walking funzionano come prima.

Thread t = Thread.ofVirtual().name("vt-1").start(() -> System.out.println("ciao"));

try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {   // un virtual thread per task
  IntStream.range(0, 100_000).forEach(i ->
      executor.submit(() -> {
        Thread.sleep(Duration.ofSeconds(1));   // simula I/O: il virtual thread viene smontato
        return i;
      }));
}   // close() implicita: aspetta tutti i task

Quando servono. Non sono più veloci: un virtual thread non calcola di più, e non ce ne possono essere più in esecuzione che core. Ma ce ne possono essere moltissimi inattivi, e una richiesta a un server passa quasi tutto il tempo ad aspettare rete, file o database. La legge di Little lega throughput T, task concorrenti N e latenza d: T = N/d. Se la latenza non scende, il throughput sale solo aumentando N, e con i platform thread N è limitato. Nell'esempio qui sopra 100.000 task che dormono un secondo finiscono in circa 1,1–1,6 s con i virtual thread. Con un cached pool il programma rischia OutOfMemoryError. Con un pool fisso di 1.000 thread la legge prevede 100 s.

Limiti. Alcune operazioni bloccanti non smontano, ma catturano il carrier; la JVM compensa aggiungendo temporaneamente un platform thread. Un virtual thread che esegue un metodo nativo, una funzione esterna o codice dentro un blocco synchronized è pinned, inchiodato al carrier, e non si smonta; in questo caso non viene aggiunto nulla. Il programma resta corretto ma scala peggio, e per questo le slide consigliano di scrivere i monitor per i virtual thread con ReentrantLock e Condition. Lo scheduler, poi, non fa preemption: un virtual thread che calcola senza mai bloccarsi tiene il carrier per sé.

Per Ricci i virtual thread sono la risposta di Java alla programmazione asincrona: il codice resta sincrono, senza callback né await, perché ogni chiamata bloccante diventa da sola un punto di smontaggio. Possono rappresentare componenti attivi che non terminano, come il ciclo di un server, in qualunque numero; un task resta un'unità di lavoro delimitata. Il confronto con event loop e programmazione reattiva, i tre approcci dell'assignment 2, è nel capitolo 12.

Verifica le tue conoscenze