Concorrenza in Java: thread, synchronized, task ed executor

1. Multithreading in Java: Thread e Runnable

Il professor Ricci introduce la programmazione multi-thread in Java con una precisazione importante: Java adotta un approccio ibrido. Da un lato, le classi per i thread sono librerie di base (J2SE, java.util.concurrent), indipendenti dal sistema operativo — a differenza dei PThread in C che sono specifici per sistemi POSIX. Dall'altro, il linguaggio include elementi nativi come la parola chiave synchronized e il fatto che ogni oggetto Java ha nativamente un lock associato.

Creazione di thread in Java

Due modi principali per creare thread in Java:

class MyThread extends Thread {
    private int id;

    MyThread(int id) {
        this.id = id;
    }

    public void run() {
        System.out.println("Thread " + id + " avviato");
        try { Thread.sleep(5000); } catch (InterruptedException e) {}
        System.out.println("Thread " + id + " done");
    }
}

// Uso:
MyThread t1 = new MyThread(1);
MyThread t2 = new MyThread(2);
t1.start();
t2.start();

class MyRunnable implements Runnable {
    private String name;

    MyRunnable(String name) {
        this.name = name;
    }

    public void run() {
        System.out.println(name + " avviato");
    }
}

// Uso:
Thread t1 = new Thread(new MyRunnable("Thread-1"));
Thread t2 = new Thread(new MyRunnable("Thread-2"));
t1.start();
t2.start();

Il metodo run() non va chiamato direttamente

Comportamento attivo e terminazione

I componenti attivi spesso non devono terminare. A differenza di una computazione tradizionale (input, processa, output, termina), i thread tipicamente hanno un comportamento ciclico, persistente. Il metodo run() non restituisce nulla (void) e non viene chiamato da noi: lo chiama la JVM sul nuovo thread.

Un esempio: ordinamento parallelo con quicksort

Il professor Ricci anticipa un esempio che sara sviluppato nella prossima lezione. L'idea e ordinare un vettore enorme usando quicksort, ma sfruttando tutti i core del processore. Nel main si genera un array di 400 milioni di elementi, si chiama Arrays.sort() (sequenziale) e si misura il tempo. La sfida e parallelizzare l'ordinamento.

2. Sezione critica in Java: synchronized

Il professore usa l'esempio concreto dal repo labactivity03 per mostrare come si realizza una sezione critica in Java con il costrutto synchronized. Due thread, WorkerA e WorkerB, eseguono azioni cicliche. WorkerA esegue A1, A2, A3; WorkerB esegue B1, B2, B3. L'obiettivo e fare in modo che A2 e A3 (oppure A2 da sola) siano eseguite in mutua esclusione rispetto a B2 e B3.

L'idea base e: un blocco synchronized(lock) crea una regione critica: prima di entrare, il thread cerca di acquisire il lock intrinseco dell'oggetto specificato. Se un altro thread lo possiede gia, il thread si blocca (viene messo nell'entry set dell'oggetto). Quando il lock viene rilasciato (usciendo dal blocco), un thread in attesa viene risvegliato e puo acquisirlo.

Schema dell'esempio WorkerA/WorkerB

Nel codice mostrato dal professore, i worker estendono una classe base Worker che fornisce metodi di utilita come waitSome() (attesa casuale). La struttura e:

// Idea: sezione critica con blocco synchronized su un oggetto condiviso
Object lock = new Object();

// Thread A
synchronized (lock) {
    // A2: azione in sezione critica
    // A3: azione in sezione critica
}

// Thread B
synchronized (lock) {
    // B2: azione in sezione critica
    // B3: azione in sezione critica
}

Il professore sottolinea che questo e il modo "piu raw, piu semplice, piu basso livello" per realizzare una sezione critica. Nella pratica si usano astrazioni di piu alto livello, ma comprendere synchronized e fondamentale.

3. Lost update e race condition

Il cuore del laboratorio e l'esempio del contatore condiviso. Il professore presenta una classe Counter con un campo int count e un metodo inc(). Due thread incrementano lo stesso contatore. Senza sincronizzazione, l'operazione count++ non e atomica: e una sequenza read-modify-write (leggi, incrementa, scrivi).

Nel codice del professore, questo si manifesta come lost update: due thread possono leggere lo stesso valore, incrementarlo ciascuno nella propria variabile locale, e riscriverlo, perdendo uno degli incrementi. Con 100.000 incrementi per thread (totale atteso 200.000), il risultato effettivo sara minore, e diverso a ogni esecuzione.

Dimostrazione dell'interleaving

Il professore simula mentalmente lo scenario: se entrambi i thread leggono count=5 prima che uno dei due lo abbia scritto, entrambi scriveranno 6 invece di 7. Ecco un simulatore interattivo per visualizzare il fenomeno:

4. Thread safety e metodi synchronized

Il professore introduce la definizione di thread safety tratta dal testo "Java Concurrency in Practice": una classe e thread-safe se puo essere usata correttamente in qualsiasi contesto, sequenziale o concorrente, senza che chi la usa debba preoccuparsi di sincronizzazione aggiuntiva. Le classi di java.util (es. ArrayList) non sono thread-safe perche la sincronizzazione ha un costo e sarebbe sprecata in contesti sequenziali.

synchronized a livello di metodo

Il professore mostra un'evoluzione: invece di un blocco synchronized esplicito, si puo dichiarare l'intero metodo come synchronized. In Java, scrivere:

public synchronized void inc() {
    count++;
}

equivale a:

public void inc() {
    synchronized (this) {
        count++;
    }
}

Il lock viene acquisito sull'oggetto stesso (this) prima di eseguire il corpo del metodo e rilasciato automaticamente all'uscita. Questo pattern si avvicina all'idioma del monitor: una classe i cui metodi pubblici sono eseguiti in mutua esclusione per costruzione.

public class Counter {
    private int count = 0;

    public void inc() { count++; }
    public int get() { return count; }
}

// Uso concorrente: PERICOLOSO!
// count++ non atomico -> lost update

public class SafeCounter {
    private int count = 0;

    public synchronized void inc() { count++; }
    public synchronized int get() { return count; }
}

// Uso concorrente: SICURO
// synchronized rende inc() e get() atomici

Il professore nota che get() deve essere synchronized anch'esso: altrimenti un thread potrebbe leggere un valore incoerente mentre un altro sta eseguendo inc() (anche se count++ e su una variabile int, la visibilita di memoria non e garantita senza sincronizzazione).

5. Lock espliciti: ReentrantLock

Oltre a synchronized, Java offre lock espliciti nel pacchetto java.util.concurrent.locks. Il professore presenta ReentrantLock e le sue varianti, sottolineando un aspetto critico: il lock va rilasciato esplicitamente, e per questo il pattern canonico prevede try/finally.

import java.util.concurrent.locks.ReentrantLock;

ReentrantLock lock = new ReentrantLock();

public void m() {
    lock.lock();        // acquisisce il lock (si blocca se necessario)
    try {
        // sezione critica
    } finally {
        lock.unlock();  // rilasciato SEMPRE, anche in caso di eccezione
    }
}

Il professore elenca le varianti offerte da ReentrantLock:

Il professore nota che Java offre anche una sintassi piu moderna con try(var lock = ...) (ARM per lock, introdotta in versioni recenti), ma il pattern classico rimane il piu diffuso.

6. Check-and-act e il BoundedCounter

Il professore introduce un esempio piu complesso che illustra il pattern check-and-act e le sue insidie. Un BoundedCounter e un contatore con soglie minime e massime: se si tenta di incrementare oltre il massimo, lancia OverflowException; se si tenta di decrementare sotto il minimo, lancia UnderflowException.

Il problema emerge quando due worker operano sullo stesso contatore:

• WorkerA: se il valore e maggiore di 0, lo decrementa.
• WorkerB: se il valore e minore di 1 (cioe minore del massimo), lo incrementa.

Il pattern check-and-act consiste nel controllare una condizione e poi agire di conseguenza. In un contesto concorrente, questo pattern e intrinsecamente non atomico: tra la lettura della condizione e l'azione, un altro thread puo modificare lo stato, invalidando la condizione appena letta.

7. Laboratorio: barriera e latch

Il professore ha dedicato la parte finale della lezione a mostrare due componenti di coordinazione riutilizzabili implementati come monitor: la barriera e il latch (countdown).

Barriera (Barrier)

Una barriera e un punto di sincronizzazione in cui N thread si incontrano. Ogni thread chiama await() e rimane bloccato finche tutti gli N partecipanti non sono arrivati. L'ultimo thread che arriva sblocca tutti gli altri.

Implementazione con synchronized:

public class CyclicBarrierMonitor {
    private int nParticipants;
    private int nArrived;

    public CyclicBarrierMonitor(int n) {
        nParticipants = n;
        nArrived = 0;
    }

    public synchronized void await() throws InterruptedException {
        nArrived++;
        while (nArrived < nParticipants) {
            wait();        // aspetta che arrivino tutti
        }
        notifyAll();       // sblocca tutti
    }
}

Versione con ReentrantLock + Condition: nella seconda implementazione, si usa una condition variable allArrived con await() e signalAll(). Attenzione: notifyAll chiamato sulla condition variable non equivale a signalAll — un errore subdolo che il professore ha mostrato in aula.

Latch (CountDown)

Un latch permette a thread di attendere (await()) finche non viene raggiunto un conteggio zero tramite chiamate countDown(). A differenza della barriera, non e ciclico e il conteggio e monotono decrescente.

public class CountDownLatchMonitor {
    private int count;

    public CountDownLatchMonitor(int count) { this.count = count; }

    public synchronized void await() throws InterruptedException {
        while (count > 0) { wait(); }
    }

    public synchronized void countDown() {
        if (count > 0) {
            count--;
            if (count == 0) { notifyAll(); }
        }
    }
}

8. Dal problema ai task: l'analisi task-oriented

Il professor Ricci apre la lezione introducendo il concetto fondamentale del design task-oriented: quando progettiamo un sistema concorrente, dobbiamo iniziare dall'analisi del problema per identificare i task — unita di lavoro ben definite, con un inizio e una fine — che compongono il sistema. Questa analisi puo partire dai dati (data-driven) o dalle funzionalita (function-driven).

L'analisi procede rispondendo a tre domande fondamentali:

1. Quali sono i task? — Identificare le attivita concorrenti guardando il problema, non la soluzione.
2. Quali sono le dipendenze tra i task? — Task completamente indipendenti possono essere assegnati a componenti attivi diversi senza coordinazione. Task con dipendenze richiedono meccanismi di sincronizzazione.
3. Quali sono i confini dei task? — Definire chiaramente dove un task inizia e dove finisce, per facilitare il recovery da errori e la composizione.

Il professore sottolinea l'importanza di adottare un approccio limpido e lineare: esplicitare le cose, non nasconderle dietro astrazioni troppo sofisticate. La semplicita nel design paga sempre, soprattutto nei sistemi concorrenti dove la complessita intrinseca e gia elevata.

Division of Labor Pattern (Divide-et-Impera)

Il pattern Division of Labor, noto anche come strategia del divide-et-impera, consiste nel:

• Identificare i confini dei task — task come attivita indipendenti
• Scegliere una politica di esecuzione dei task — in base agli obiettivi di throughput, responsiveness e graceful degradation
• Separare la definizione del task dalla sua esecuzione

Il risultato di una buona analisi task-oriented e un'organizzazione del programma piu semplice, che facilita il recovery da errori (i task sono confini naturali per le transazioni) e promuove naturalmente la concorrenza.

9. Il Java Executor Framework

public interface Executor {
    void execute(Runnable task);
}

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);
            });
        }
    }
}

10. Callable, Future e task con risultato

public interface Callable<V> {
    V call() throws Exception;
}

public interface Future<V> {
    boolean cancel(boolean mayInterruptIfRunning);
    boolean isCancelled();
    boolean isDone();
    V get() throws InterruptedException, ExecutionException,
                 CancellationException();
    V get(long timeout, TimeUnit unit) throws ...;
}

// Esegue un task senza risultato
executor.execute(runnableTask);

// Sottomette un task con risultato e restituisce un Future
Future<Integer> future = executor.submit(callableTask);
Integer result = future.get();  // bloccante finche il task non completa

// ComputeAreaTask: calcola l'area di un singolo trapezio
class ComputeAreaTask implements Callable<Double> {
    private final double x0, delta;
    private final Function<Double,Double> f;
    // costruttore...
    public Double call() {
        return (f.apply(x0) + f.apply(x0 + delta)) * delta / 2.0;
    }
}

// Master: scompone l'intervallo e colleziona i risultati
List<Future<Double>> futures = new ArrayList<>();
for (double x = a; x < b; x += delta) {
    futures.add(executor.submit(new ComputeAreaTask(x, delta, f)));
}
double totale = 0.0;
for (Future<Double> f : futures) {
    totale += f.get();  // si blocca finche il task non e pronto
}

public class PrimeGenerator implements Runnable {
    private final List<BigInteger> primes = new ArrayList<>();
    private volatile boolean cancelled;

    public void run() {
        BigInteger p = BigInteger.ONE;
        while (!cancelled) {
            p = p.nextProbablePrime();
            synchronized (this) { primes.add(p); }
        }
    }

    public void cancel() { cancelled = true; }
    public synchronized List<BigInteger> get() {
        return new ArrayList<>(primes);
    }
}

// Uso: genera numeri primi per 1 secondo
List<BigInteger> aSecondOfPrimes() throws InterruptedException {
    PrimeGenerator gen = new PrimeGenerator();
    new Thread(gen).start();
    try { SECONDS.sleep(1); } finally { gen.cancel(); }
    return gen.get();
}

class PrimeProducer extends Thread {
    private BlockingQueue<BigInteger> queue;
    // costruttore...
    public void run() {
        BigInteger p = BigInteger.ONE;
        try {
            while (!Thread.currentThread().isInterrupted()) {
                queue.put(p.nextProbablePrime());
            }
        } catch (InterruptedException ex) {
            // allow thread to exit
        }
    }
    public void cancel() { interrupt(); }
}

flowchart BT
    Exec["<<interface>> Executor\nexecute(Runnable)"] --> ExecService["<<interface>> ExecutorService\nsubmit(Callable<T>): Future<T>\nshutdown(), awaitTermination()"]
    ExecService --> SchedExec["<<interface>> ScheduledExecutorService\nschedule(), scheduleAtFixedRate()"]
    ExecFactory["Executors (factory)\nnewFixedThreadPool(), newCachedThreadPool(),\nnewSingleThreadExecutor(), newScheduledThreadPool()"]
    ExecFactory -.->|crea| ExecService
    

// Callable + Future pattern
ExecutorService exec = Executors.newFixedThreadPool(4);
Future<Integer> future = exec.submit(() -> {
    // computazione lunga...
    return 42;
});

// ... altro lavoro concorrente ...

Integer risultato = future.get(); // bloccante se non ancora pronto

11. Cancellazione cooperativa e interruption

La cancellazione di un'attività concorrente segue un approccio cooperativo, non asincrono/preemptive. In Java, il meccanismo principale è l'interruption:

// Pattern di cancellazione cooperativa (dalle slide)
public class PrimeGenerator implements Runnable {
    private volatile boolean cancelled;

    public void run() {
        BigInteger p = BigInteger.ONE;
        while (!cancelled) {
            p = p.nextProbablePrime();
            // ... elaborazione ...
        }
    }

    public void cancel() { cancelled = true; }
}

Il flag volatile garantisce che le scritture di cancel() siano visibili immediatamente al thread in esecuzione. Se il thread è bloccato su una wait(), sleep() o join(), si usa il meccanismo di interrupt() che sblocca questi stati e lancia InterruptedException.

12. Cronometro MVC: il bug della reattivita

Il professor Ricci passa a un esempio pratico e doloroso: un cronometro implementato con pattern Model-View-Controller, ma con due bug classici della programmazione concorrente in GUI Java.

Il programma si chiama CronoMVCNotReactivePlusRace ed e l'esempio di come NON si fa. L'idea e semplice: un cronometro con pulsanti Start, Stop e Reset che conta i secondi in un text field.

Il bug principale e nel controller: l'azione di start viene eseguita direttamente sull'Event Dispatch Thread (EDT), il thread di Swing che gestisce tutti gli eventi della GUI. Cosa succede?

// VERSIONE BUGGY: il controller esegue il loop sull'EDT!
buttonStart.addActionListener(e -> {
    started = true;
    while (started) {
        model.updateState();
        display.setText(String.valueOf(model.getValue()));
        // ^^^^^ setText accoda un task all'EDT...
        // ma l'EDT e gia occupato qui nel loop!
        Thread.sleep(100);  // blocca tutto
    }
});

Il problema e duplice:

• setText() su un componente Swing non aggiorna immediatamente il display: accoda un task nella coda eventi dell'EDT.
• Ma l'EDT e gia in esecuzione! Sta eseguendo il while(started) del controller. Quindi non puo servire la coda eventi. Il display non si aggiorna mai e l'interfaccia e completamente bloccata.

13. Cronometro MVC: la soluzione

La versione corretta si chiama CronoMVCReactiveNonRaces. La regola fondamentale e: l'EDT si occupa solo della business logic che riguarda la view, non il model, non il controller.

// VERSIONE CORRETTA: il controller delega a un thread separato
class Controller {
    private final Model model;
    private final View view;
    private Thread countingThread;  // thread separato per il conteggio

    public void notifyStarted() {
        countingThread = new Thread(() -> {
            while (isRunning()) {
                model.updateState();       // aggiorno il modello
                SwingUtilities.invokeLater(() -> {
                    view.updateDisplay(    // aggiorno la view sull'EDT
                        model.getValue());
                });
                Thread.sleep(100);
            }
        });
        countingThread.start();
    }

    public void notifyStopped() {
        stopRunning();  // flag per fermare il loop
    }
}

SwingUtilities.invokeLater() accoda un task nella coda dell'EDT e torna immediatamente. Il thread di conteggio puo continuare a eseguire senza bloccare l'interfaccia. In questo modo l'EDT rimane libero di servire eventi (pulsanti, aggiornamenti) e il cronometro funziona correttamente.

Il docente sottolinea che invokeAndWait() e un'alternativa sincrona: il chiamante si blocca finche l'EDT non ha completato il task. Utile per sincronizzarsi, ma per l'aggiornamento del display e preferibile invokeLater().

14. Sketch2: input asincrono da tastiera

Il secondo esempio pratico e Sketch2, un programma GUI che illustra come gestire input asincrono da tastiera all'interno di un'architettura Model-View-Controller.

L'applicazione mostra un conteggio che viene aggiornato sia automaticamente (da un componente attivo che incrementa il valore periodicamente) sia manualmente (premendo il tasto I sulla tastiera). Il tasto R resetta il conteggio.

La sfida: la tastiera genera eventi in modo asincrono, indipendentemente dal ciclo di vita del programma. Come li integriamo nel MVC?

// Gestione dell'input da tastiera in Java con MVC
public class Sketch2 {
    public static void main(String[] args) {
        Model model = new Model();
        View view = new View(model);
        Controller controller = new Controller(model, view);

        // Il controller registra listener per gli eventi di tastiera
        view.addKeyListener(new KeyAdapter() {
            public void keyPressed(KeyEvent e) {
                if (e.getKeyCode() == KeyEvent.VK_I) {
                    controller.notifyIncrementRequest();
                } else if (e.getKeyCode() == KeyEvent.VK_R) {
                    controller.notifyResetRequest();
                }
            }
        });
    }
}

Il professore spiega che la gestione e concettualmente identica a quella del cronometro: il controller reagisce agli eventi, modifica il modello su un thread separato (non l'EDT), e usa SwingUtilities.invokeLater() per aggiornare la vista. In questo modo l'interfaccia rimane reattiva anche durante computazioni lunghe.

15. Simulazione fisica: palle che rimbalzano

L'esercizio delle palle che rimbalzano e un caso studio di simulazione fisica concorrente. Il professore presenta un modello in cui delle palline si muovono in uno spazio bidimensionale, con:

• Attrito — una decelerazione costante che dipende dalla velocita (friction factor)
• Rimbalzo ai bordi — quando una pallina raggiunge il bordo, inverte il vettore velocita
• Aggiornamento cinematico — posizione = posizione + velocita * delta tempo

Ecco la logica di aggiornamento del modello fisico:

class Particle {
    double x, y;       // posizione
    double vx, vy;     // velocita

    void update(double dt) {
        // Applica attrito (decelerazione costante)
        double friction = 0.98;  // coefficiente
        vx *= friction;
        vy *= friction;

        // Se la velocita e molto bassa, ferma del tutto
        if (Math.abs(vx) < 0.01) vx = 0;
        if (Math.abs(vy) < 0.01) vy = 0;

        // Aggiorna posizione (cinematica)
        x += vx * dt;
        y += vy * dt;

        // Boundary check: rimbalzo sui bordi
        if (x < 0 || x > maxX) { vx = -vx; x = clamp(x, 0, maxX); }
        if (y < 0 || y > maxY) { vy = -vy; y = clamp(y, 0, maxY); }
    }
}

16. Fork-Join, work-stealing e structured concurrency

// Esempio: somma degli elementi di un array (divide & conquer)
class SumTask extends RecursiveTask<Long> {
    static final int THRESHOLD = 1000;
    private final int[] array;
    private final int lo, hi;

    SumTask(int[] a, int lo, int hi) { this.array = a; this.lo = lo; this.hi = hi; }

    protected Long compute() {
        if (hi - lo < THRESHOLD) {
            long sum = 0;
            for (int i = lo; i < hi; i++) sum += array[i];
            return sum;
        }
        int mid = (lo + hi) / 2;
        SumTask left = new SumTask(array, lo, mid);
        SumTask right = new SumTask(array, mid, hi);
        left.fork();                // fork: avvia il task in parallelo
        long rightResult = right.compute();  // compute: esegui direttamente
        long leftResult = left.join();       // join: attendi il risultato
        return leftResult + rightResult;
    }
}

// Uso con ForkJoinPool
ForkJoinPool pool = new ForkJoinPool();
long total = pool.invoke(new SumTask(array, 0, array.length));

// Schema fork-join: merge sort parallelo (idea dalle slide)
class MergeSortTask extends RecursiveAction {
    private int[] array;
    private int lo, hi;

    protected void compute() {
        if (hi - lo < THRESHOLD) {
            Arrays.sort(array, lo, hi);
        } else {
            int mid = (lo + hi) / 2;
            var left = new MergeSortTask(array, lo, mid);
            var right = new MergeSortTask(array, mid, hi);
            left.fork();   // fork asincrono
            right.compute();
            left.join();   // attendi il risultato
            merge(array, lo, mid, hi);
        }
    }
}

// Esempio: fetch parallelo di due sorgenti
Response handle() throws ExecutionException, InterruptedException {
    try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
        Future<String> user = scope.fork(() -> findUser());
        Future<Integer> order = scope.fork(() -> fetchOrder());

        scope.join();            // attendi entrambi i fork
        scope.throwIfFailed();   // propaga errori

        // Entrambi completati con successo
        return new Response(user.resultNow(), order.resultNow());
    }  // scope si chiude: tutti i fork sono completati
}

Verifica le tue conoscenze