08 - Strutture Dati Lock-Free: Operazioni Atomiche, Memory Ordering e Compare-and-Swap

Quando un lock è il problema, e che cosa lo sostituisce

Un lock serializza: mentre un thread lavora nella sezione critica gli altri aspettano, e se quel thread viene fermato dallo scheduler gli altri aspettano con lui. Una struttura lock-free toglie il lock e usa istruzioni atomiche della CPU, in particolare il compare-and-swap (CAS): «scrivi il valore nuovo solo se in memoria c'è ancora il valore che ho letto». La garanzia che si ottiene è precisa e più debole di quanto si creda: lock-free significa che, in ogni momento, almeno un thread completa la sua operazione anche se altri sono fermi o rallentati; non significa che ogni thread avanzi (quello è wait-free) né che la struttura sia più veloce di un lock.

Prima di costruire qualcosa conviene guardare che cosa offre già la libreria standard, perché nella pratica quasi mai scrivi le strutture a mano: AtomicInteger, AtomicReference, LongAdder, ConcurrentLinkedQueue. Gli esperimenti qui sotto servono a capire perché funzionano e dove si rompono. Tutti i listati Java sono stati compilati ed eseguiti con OpenJDK 21.0.13; i due in C++ con g++ (C++17) su x86-64. I conteggi sono esatti, i tempi cambiano da una macchina all'altra. Per la parte sui lock tradizionali vedi race condition e deadlock.

Il ciclo CAS: leggi, decidi, prova a scrivere

Il ciclo CAS ha sempre la stessa forma: leggi il valore, calcola quello nuovo, tenta il compareAndSet; se un altro thread ha cambiato il valore nel frattempo, il tentativo fallisce e ricominci. L'esempio concreto è un magazzino con 100.000 pezzi e otto thread che tentano 160.000 prelievi: deve riuscirne esattamente 100.000, e il residuo deve essere zero. Il listato lo prova con un contatore normale e con il ciclo CAS.

import java.util.concurrent.*;
import java.util.concurrent.atomic.*;

// 100.000 pezzi in magazzino, 8 thread che tentano di prelevarne uno alla volta.
public class Scorta {
    static int semplice = 100_000;                          // get-then-set: NON atomico
    static final AtomicInteger atomica = new AtomicInteger(100_000);

    // CAS loop: "decrementa solo se positivo"
    static boolean prelevaCas() {
        while (true) {
            int corrente = atomica.get();                   // 1. leggi
            if (corrente <= 0) return false;                // 2. condizione di business
            if (atomica.compareAndSet(corrente, corrente - 1)) {
                return true;                                // 3. nessuno ha toccato il valore: fatto
            }
            // CAS fallito: qualcuno ha cambiato il valore, rileggi e riprova
        }
    }

    static boolean prelevaSemplice() {
        int corrente = semplice;
        if (corrente <= 0) return false;
        semplice = corrente - 1;                            // altri thread possono aver scritto nel frattempo
        return true;
    }

    public static void main(String[] args) throws Exception {
        for (String modo : new String[] {"semplice", "cas"}) {
            AtomicInteger riusciti = new AtomicInteger();
            ExecutorService exec = Executors.newFixedThreadPool(8);
            for (int t = 0; t < 8; t++) {
                exec.submit(() -> {
                    for (int i = 0; i < 20_000; i++) {      // 160.000 tentativi su 100.000 pezzi
                        boolean ok = modo.equals("cas") ? prelevaCas() : prelevaSemplice();
                        if (ok) riusciti.incrementAndGet();
                    }
                });
            }
            exec.shutdown();
            exec.awaitTermination(30, TimeUnit.SECONDS);
            int residuo = modo.equals("cas") ? atomica.get() : semplice;
            System.out.printf("%-9s prelievi riusciti: %d, residuo: %d, riusciti + residuo = %d%n",
                modo, riusciti.get(), residuo, riusciti.get() + residuo);
        }
    }
}

// Output:
// semplice  prelievi riusciti: 160000, residuo: 27118, riusciti + residuo = 187118
// cas       prelievi riusciti: 100000, residuo: 0, riusciti + residuo = 100000

Con l'int semplice i prelievi riusciti sono 160.000 su 100.000 pezzi disponibili e il residuo resta 27.118: la somma «riusciti + residuo» doveva fare 100.000, fa 187.118. Il programma ha venduto più di quanto avesse, e ne ha anche lasciato. Con il ciclo CAS la somma torna: 100.000 prelievi riusciti, residuo zero. La condizione di business («solo se positivo») è sempre nel ciclo prima del compareAndSet, e il CAS verifica che la condizione valga ancora quando scrivi. Un AtomicInteger.decrementAndGet() non basterebbe, perché non può rifiutarsi di scendere sotto zero.

Pila di Treiber: un solo puntatore, un solo CAS

La pila lock-free classica tiene un solo puntatore, top. Per push si crea il nodo, si fa puntare al top corrente e si tenta il CAS su top; per pop si legge top, si tenta di spostarlo al nodo successivo. L'algoritmo è attribuito a R. K. Treiber (rapporto IBM RJ 5118, aprile 1986, citato nell'articolo di Michael e Scott). Il listato lo sottopone a quattro produttori e quattro consumatori con un milione di elementi e verifica che nulla vada perso né duplicato confrontando le somme.

import java.util.concurrent.*;
import java.util.concurrent.atomic.*;

public class Treiber {
    static class Nodo<T> {
        final T valore;
        Nodo<T> next;
        Nodo(T valore) { this.valore = valore; }
    }

    static class Pila<T> {
        private final AtomicReference<Nodo<T>> top = new AtomicReference<>();

        void push(T v) {
            Nodo<T> nuovo = new Nodo<>(v);
            while (true) {
                Nodo<T> vecchio = top.get();
                nuovo.next = vecchio;                         // il nuovo punta al vecchio top
                if (top.compareAndSet(vecchio, nuovo)) return;
            }
        }

        T pop() {
            while (true) {
                Nodo<T> vecchio = top.get();
                if (vecchio == null) return null;
                if (top.compareAndSet(vecchio, vecchio.next)) return vecchio.valore;
            }
        }
    }

    public static void main(String[] args) throws Exception {
        Pila<Integer> pila = new Pila<>();
        int produttori = 4, perProduttore = 250_000;
        AtomicLong sommaPush = new AtomicLong(), sommaPop = new AtomicLong();
        AtomicInteger contati = new AtomicInteger();
        ExecutorService exec = Executors.newFixedThreadPool(8);
        CountDownLatch fine = new CountDownLatch(produttori);

        for (int p = 0; p < produttori; p++) {
            final int base = p * perProduttore;
            exec.submit(() -> {
                for (int i = 1; i <= perProduttore; i++) {
                    pila.push(base + i);
                    sommaPush.addAndGet(base + i);
                }
                fine.countDown();
            });
        }
        for (int c = 0; c < 4; c++) {
            exec.submit(() -> {
                while (fine.getCount() > 0 || contati.get() < produttori * perProduttore) {
                    Integer v = pila.pop();
                    if (v != null) { sommaPop.addAndGet(v); contati.incrementAndGet(); }
                }
            });
        }
        exec.shutdown();
        exec.awaitTermination(60, TimeUnit.SECONDS);
        System.out.println("inseriti: " + produttori * perProduttore + ", estratti: " + contati.get());
        System.out.println("somma inseriti = " + sommaPush.get() + ", somma estratti = " + sommaPop.get());
        System.out.println("pila vuota alla fine: " + (pila.pop() == null));
    }
}

// Output:
// inseriti: 1000000, estratti: 1000000
// somma inseriti = 500000500000, somma estratti = 500000500000
// pila vuota alla fine: true

Un milione inseriti, un milione estratti, somma identica (500.000.500.000): la conservazione regge. Questo non dimostra la correttezza, perché un test concorrente può passare per caso, ma è un controllo sensato per scoprire perdite e duplicazioni. Si noti l'uso di un nodo nuovo per ogni push: in Java il garbage collector impedisce che un nodo venga riutilizzato finché qualcuno lo referenzia, ed è per questo che in questa implementazione il problema seguente non si presenta.

Il problema ABA

Il CAS controlla il valore di top, non la sua storia. Se tra la lettura e il CAS il valore passa da A a B e torna ad A, il CAS riesce anche se la struttura sotto è cambiata. Succede quando i nodi vengono riutilizzati (un pool, o un allocatore che restituisce lo stesso indirizzo), tipicamente in C e C++. L'articolo di Michael e Scott lo definisce così, e propone come soluzione comune un contatore di modifiche associato al puntatore; precisa che il contatore non garantisce l'assenza di ABA (può avvolgersi) ma lo rende estremamente improbabile.

import java.util.*;
import java.util.concurrent.atomic.*;

// ABA riprodotto in modo deterministico: i "thread" sono simulati eseguendo i passi a mano
// nell'ordine problematico. La pila RIUSA i nodi (pool), come nel C/C++ senza garbage collector.
public class Aba {
    static class Nodo {
        final String nome;
        Nodo next;
        Nodo(String nome) { this.nome = nome; }
    }

    static final AtomicReference<Nodo> top = new AtomicReference<>();
    static final AtomicStampedReference<Nodo> topStamp = new AtomicStampedReference<>(null, 0);

    public static void main(String[] args) {
        Nodo a = new Nodo("A"), b = new Nodo("B"), c = new Nodo("C");
        a.next = b; b.next = c;
        top.set(a);
        topStamp.set(a, 0);

        // Thread 1 inizia pop(): legge top e top.next, poi viene sospeso
        Nodo visto = top.get();                 // A
        Nodo prossimo = visto.next;             // B
        int[] bollo = new int[1];
        Nodo vistoStamp = topStamp.get(bollo);  // A, bollo 0
        int bolloVisto = bollo[0];

        // Thread 2: pop A, pop B, push A (stesso oggetto, riusato)
        top.set(b);                             // pop A
        top.set(c);                             // pop B  (B esce dalla pila: nessuno lo usa piu')
        a.next = c;                             // push A riusando il nodo
        top.set(a);                             // pila: A -> C
        topStamp.set(a, 3);                     // con il bollo, ognuna delle tre modifiche lo incrementa: ora vale 3

        // Thread 1 riprende: CAS(top, A, B)
        boolean riuscito = top.compareAndSet(visto, prossimo);
        System.out.println("senza bollo:  CAS riuscito = " + riuscito + ", top = " + top.get().nome
            + "  <- B era gia' stato estratto: lo stato e' corrotto");

        boolean riuscitoStamp = topStamp.compareAndSet(vistoStamp, prossimo, bolloVisto, bolloVisto + 1);
        System.out.println("con bollo:    CAS riuscito = " + riuscitoStamp + ", top = " + topStamp.getReference().nome
            + "  <- il bollo e' cambiato, il CAS fallisce e il thread rilegge");
    }
}

// Output:
// senza bollo:  CAS riuscito = true, top = B  <- B era gia' stato estratto: lo stato e' corrotto
// con bollo:    CAS riuscito = false, top = A  <- il bollo e' cambiato, il CAS fallisce e il thread rilegge

Qui i thread sono simulati: eseguo a mano, nell'ordine problematico, i passi che l'interleaving produce. Il thread 1 ha letto top = A e A.next = B ed è sospeso. Nel frattempo il thread 2 estrae A, estrae B e reinserisce A (lo stesso oggetto), quindi la pila è A→C. Il thread 1 riprende e fa CAS(top, A, B): riesce, e top diventa B, un nodo già estratto. Con AtomicStampedReference, che affianca al riferimento un intero («bollo»), il CAS confronta riferimento e bollo: il bollo è passato da 0 a 3, e il tentativo fallisce. Le alternative più sofisticate sono gli hazard pointer e la epoch-based reclamation (è il meccanismo del crate Rust crossbeam-epoch): non le ho provate qui.

Coda di Michael e Scott: due puntatori e l'aiuto reciproco

Una coda ha due estremi, quindi il CAS singolo non basta: l'inserimento deve prima agganciare il nuovo nodo dopo l'ultimo, poi spostare tail, e fra i due passi un altro thread può arrivare. La soluzione di Michael e Scott (1996) è che chiunque trovi tail in ritardo la fa avanzare lui stesso: è l'aiuto reciproco che rende l'algoritmo lock-free. La documentazione di ConcurrentLinkedQueue dice che usa un algoritmo non bloccante basato su quell'articolo. Il listato è un'implementazione didattica, con la stessa verifica di conservazione e un controllo in più: l'ordine FIFO per produttore.

import java.util.concurrent.*;
import java.util.concurrent.atomic.*;

public class CodaMs {
    static class Nodo<T> {
        volatile T valore;
        final AtomicReference<Nodo<T>> next = new AtomicReference<>();
        Nodo(T valore) { this.valore = valore; }
    }

    static class Coda<T> {
        private final AtomicReference<Nodo<T>> head, tail;

        Coda() {
            Nodo<T> sentinella = new Nodo<>(null);
            head = new AtomicReference<>(sentinella);
            tail = new AtomicReference<>(sentinella);
        }

        void enqueue(T v) {
            Nodo<T> nuovo = new Nodo<>(v);
            while (true) {
                Nodo<T> ultimo = tail.get();
                Nodo<T> dopo = ultimo.next.get();
                if (ultimo != tail.get()) continue;           // tail e' cambiata: rileggi
                if (dopo == null) {
                    if (ultimo.next.compareAndSet(null, nuovo)) {   // aggancia il nodo
                        tail.compareAndSet(ultimo, nuovo);          // sposta tail (se fallisce, ci pensa un altro)
                        return;
                    }
                } else {
                    tail.compareAndSet(ultimo, dopo);         // tail in ritardo: aiutala ad avanzare
                }
            }
        }

        T dequeue() {
            while (true) {
                Nodo<T> primo = head.get();
                Nodo<T> ultimo = tail.get();
                Nodo<T> dopo = primo.next.get();
                if (primo != head.get()) continue;
                if (primo == ultimo) {
                    if (dopo == null) return null;            // vuota
                    tail.compareAndSet(ultimo, dopo);         // tail in ritardo
                } else {
                    T v = dopo.valore;
                    if (head.compareAndSet(primo, dopo)) return v;
                }
            }
        }
    }

    public static void main(String[] args) throws Exception {
        Coda<Long> coda = new Coda<>();
        int produttori = 4, perProduttore = 200_000;
        AtomicLong sommaPop = new AtomicLong();
        AtomicInteger contati = new AtomicInteger();
        AtomicInteger fuoriOrdine = new AtomicInteger();
        ExecutorService exec = Executors.newFixedThreadPool(8);
        CountDownLatch fine = new CountDownLatch(produttori);

        for (int p = 0; p < produttori; p++) {
            final long id = p;
            exec.submit(() -> {
                for (long i = 1; i <= perProduttore; i++) coda.enqueue(id * 1_000_000L + i);
                fine.countDown();
            });
        }
        for (int c = 0; c < 4; c++) {
            exec.submit(() -> {
                long[] ultimoPerProduttore = new long[produttori];
                while (fine.getCount() > 0 || contati.get() < produttori * perProduttore) {
                    Long v = coda.dequeue();
                    if (v == null) continue;
                    sommaPop.addAndGet(v);
                    contati.incrementAndGet();
                    int p = (int) (v / 1_000_000L);
                    if (v < ultimoPerProduttore[p]) fuoriOrdine.incrementAndGet(); // FIFO per produttore, visto da questo consumatore
                    ultimoPerProduttore[p] = v;
                }
            });
        }
        exec.shutdown();
        exec.awaitTermination(60, TimeUnit.SECONDS);

        long atteso = 0;
        for (long p = 0; p < produttori; p++)
            for (long i = 1; i <= perProduttore; i++) atteso += p * 1_000_000L + i;
        System.out.println("estratti: " + contati.get() + " su " + produttori * perProduttore);
        System.out.println("somma attesa = " + atteso + ", somma estratta = " + sommaPop.get());
        System.out.println("estrazioni fuori ordine per produttore: " + fuoriOrdine.get());
    }
}

// Output:
// estratti: 800000 su 800000
// somma attesa = 1280000400000, somma estratta = 1280000400000
// estrazioni fuori ordine per produttore: 0

Gli 800.000 elementi inseriti da quattro produttori escono tutti, la somma coincide (1.280.000.400.000) e, per ogni consumatore, gli elementi di uno stesso produttore escono sempre in ordine crescente: zero inversioni. La sentinella (un nodo vuoto iniziale) evita il caso speciale della coda vuota nei due puntatori. In produzione usa ConcurrentLinkedQueue, già verificata e ottimizzata: l'esercizio serve a vedere dove ogni compareAndSet è indispensabile.

Memory ordering: perché non basta l'atomicità

Un'operazione atomica garantisce che non si veda un valore a metà; non garantisce quando gli altri thread vedono le altre scritture. CPU e compilatori riordinano gli accessi, e il modello di memoria del linguaggio dice quali riordinamenti sono ammessi. In Java il pilastro è la relazione happens-before: la specifica del linguaggio dice che una lettura può vedere una scrittura se nessun ordine happens-before lo impedisce. Il listato mostra la conseguenza più concreta, la visibilità di un flag.

public class Visibilita {
    static boolean semplice = false;            // nessuna garanzia di visibilita'
    static volatile boolean volatile_ = false;  // scrittura e lettura sincronizzate

    static String prova(boolean usaVolatile) throws Exception {
        semplice = false; volatile_ = false;
        Thread t = new Thread(() -> {
            if (usaVolatile) { while (!volatile_) { } }
            else { while (!semplice) { } }       // il JIT puo' portare la lettura fuori dal ciclo
        });
        t.setDaemon(true);
        t.start();
        Thread.sleep(300);                      // lascia il tempo al JIT di compilare il ciclo
        if (usaVolatile) volatile_ = true; else semplice = true;
        t.join(2000);
        return t.isAlive() ? "ANCORA IN CICLO dopo 2 s" : "terminato";
    }

    public static void main(String[] args) throws Exception {
        System.out.println("campo volatile:     " + prova(true));
        System.out.println("campo non volatile: " + prova(false));
    }
}

// Output:
// campo volatile:     terminato
// campo non volatile: ANCORA IN CICLO dopo 2 s

Con volatile il ciclo termina appena il flag cambia. Senza, il thread era ancora in ciclo due secondi dopo: il compilatore JIT ha tolto la lettura dal ciclo, perché in assenza di sincronizzazione nulla gli vieta di leggere il campo una volta sola. Non è un difetto di HotSpot: il programma non ha un happens-before fra la scrittura e la lettura, e il risultato è consentito dalla specifica. L'esito dipende da JIT e macchina: ho osservato il blocco, ma non è garantito che accada ovunque, ed è questa la natura del bug.

Linguaggi come C++ e Rust offrono controllo più fine, con cinque livelli: Relaxed (solo atomicità), Release (scrittura che pubblica tutto ciò che la precede), Acquire (lettura che vede quanto pubblicato dal release corrispondente), AcqRel (le due insieme, per le operazioni di lettura-modifica-scrittura) e SeqCst (anche un ordine totale unico). La documentazione di Rust precisa che i suoi ordinamenti sono quelli di C++20. Il listato C++ pubblica un dato non atomico con una coppia release/acquire.

#include <atomic>
#include <cstdio>
#include <thread>

int main() {
    int dato = 0;                                // dato NON atomico, pubblicato tramite il flag
    std::atomic<bool> pronto{false};
    int visto = -1;

    std::thread produttore([&] {
        dato = 42;                                                 // 1. scrivi il dato
        pronto.store(true, std::memory_order_release);             // 2. pubblica: "tutto cio' che e' prima e' visibile"
    });
    std::thread consumatore([&] {
        while (!pronto.load(std::memory_order_acquire)) { }        // 3. attendi la pubblicazione
        visto = dato;                                              // 4. il dato e' garantito visibile
    });
    produttore.join();
    consumatore.join();
    printf("il consumatore vede dato = %d\n", visto);
}

// Output:
// $ g++ -std=c++17 -O2 -pthread Pubblica.cpp && ./a.out
// il consumatore vede dato = 42
//
// $ ... -fsanitize=thread, con release/acquire: nessuna segnalazione
// $ ... -fsanitize=thread, con memory_order_relaxed al posto di release: "WARNING: ThreadSanitizer: data race"

L'uscita normale è 42, ma non dimostra nulla: su x86-64, dove le scritture non vengono riordinate fra loro dalla CPU, anche la versione sbagliata dà spesso 42 (ho tentato un test di store buffering con 2.000.000 iterazioni e non ho osservato nessun esito anomalo, quindi non lo riporto). Il controllo utile è ThreadSanitizer: con release/acquire non segnala nulla; sostituendo release con relaxed, segnala un data race sulla variabile dato. Lo strumento verifica l'ordinamento formale, non il comportamento di una CPU in particolare.

Quando il lock-free conviene (e quando no)

Il caso in cui l'atomico vince nettamente è il contatore molto contestato. Il listato confronta AtomicLong e LongAdder, che divide il totale in più celle per ridurre la contesa, con otto thread e cinque milioni di incrementi ciascuno.

import java.util.concurrent.*;
import java.util.concurrent.atomic.*;

public class Adder {
    static long misura(Runnable incremento, int thread, int perThread) throws Exception {
        long t0 = System.nanoTime();
        ExecutorService exec = Executors.newFixedThreadPool(thread);
        for (int t = 0; t < thread; t++) {
            exec.submit(() -> { for (int i = 0; i < perThread; i++) incremento.run(); });
        }
        exec.shutdown();
        exec.awaitTermination(60, TimeUnit.SECONDS);
        return (System.nanoTime() - t0) / 1_000_000;
    }

    public static void main(String[] args) throws Exception {
        int thread = 8, perThread = 5_000_000;
        for (int giro = 0; giro < 2; giro++) {                 // il primo giro scalda il JIT
            AtomicLong atomico = new AtomicLong();
            LongAdder adder = new LongAdder();
            long a = misura(atomico::incrementAndGet, thread, perThread);
            long b = misura(adder::increment, thread, perThread);
            if (giro == 1) {
                System.out.println("AtomicLong: " + a + " ms, totale " + atomico.get());
                System.out.println("LongAdder:  " + b + " ms, totale " + adder.sum());
            }
        }
    }
}

// Output:
// AtomicLong: 538 ms, totale 40000000
// LongAdder:  20 ms, totale 40000000

Il risultato è identico (40.000.000), ma il LongAdder impiega 20 ms contro 538 ms, circa 27 volte meno su questa macchina. La documentazione lo raccomanda quando molti thread aggiornano una somma comune, per esempio per statistiche; il prezzo è che sum() non è un'istantanea atomica se ci sono scritture concorrenti. Il rapporto dipende dal numero di core e dall'hardware: il numero è una misura singola, non un benchmark.

Il lock-free non è il default. Introduce complessità (ABA, ordinamenti, memoria da recuperare), e non è automaticamente più veloce di un lock quando la contesa è bassa. La regola operativa è: parti dalle classi della libreria standard (ConcurrentLinkedQueue, ConcurrentHashMap, LongAdder), misura il collo di bottiglia con un profiler e solo allora valuta una struttura tua, sapendo che ti serviranno test come quelli sopra e uno strumento come ThreadSanitizer.

Che cosa sai fare adesso

Sai scrivere un ciclo CAS con una condizione di business, sai perché una pila di Treiber in Java non soffre di ABA e come lo riprodurresti con nodi riutilizzati, sai leggere una coda di Michael e Scott, e sai distinguere atomicità da visibilità. Sai che un flag senza volatile può restare invisibile per sempre e che ThreadSanitizer ti mostra la differenza fra release e relaxed.

Il prossimo articolo si sposta su un problema applicativo: backpressure e flow control, cioè che cosa fare quando chi produce è più veloce di chi consuma. Per i modelli di concorrenza a confronto, vedi il primo articolo della serie. Per la stessa questione fra macchine invece che fra thread, vedi i sistemi distribuiti.

I conteggi (160.000, 27.118, 1.000.000, 800.000) e i tempi (20 ms contro 538 ms) vengono da esecuzioni singole su una macchina x86-64; i conteggi di conservazione sono esatti, il residuo del magazzino senza CAS e i tempi cambiano a ogni esecuzione.