10 - Sistemi Distribuiti e Concorrenza: CAP Theorem, Consistency Models e CRDT

Quando la concorrenza attraversa la rete

Finora la serie ha trattato thread, task e lock dentro un solo processo, dove la memoria è condivisa e un messaggio non si perde. Appena i dati vivono su più nodi, quelle garanzie vanno guadagnate: un messaggio può arrivare in ritardo o mai, un nodo può riavviarsi, due nodi possono non vedersi pur essendo entrambi vivi. Un caso concreto: un carrello della spesa replicato su tre server, e un cliente che aggiunge un articolo dal telefono proprio mentre un server è separato dagli altri. A quale server risponde il sito, e che cosa vede il cliente alla richiesta successiva?

Le risposte hanno nomi: CAP, modelli di consistenza, CRDT. Gli esempi qui sono simulazioni in Python (3.14.4, solo libreria standard, tutte eseguite): semplificano la rete a un insieme di nodi che si vedono o no, e non sostituiscono la prova su un database vero. Per la coordinazione dentro un processo, il filo della serie è in race condition e deadlock.

CAP: che cosa dice davvero

Il teorema CAP riguarda tre proprietà di un sistema replicato:

• Consistenza (C): ogni lettura vede l'ultima scrittura, o riceve un errore.
• Disponibilità (A): ogni richiesta ricevuta da un nodo funzionante riceve una risposta.
• Tolleranza alla partizione (P): il sistema continua a funzionare anche quando la rete perde o ritarda messaggi fra i nodi.

L'errore più comune è leggere «scegline due su tre» come un menù. Una partizione di rete non si sceglie: capita. Quindi, quando capita, il sistema deve decidere fra C e A, ed è tutto qui. Daniel Abadi, nell'articolo che propone PACELC, lo dice con parole precise: CAP impone restrizioni solo quando c'è una partizione e non limita le capacità del sistema in funzionamento normale. Il mio ordine di lettura è: C e A sono in conflitto soltanto durante la partizione; fuori dalla partizione resta un altro compromesso, quello fra latenza e consistenza, che CAP non descrive.

Per vedere la scelta, la simulazione seguente ha tre repliche A, B e C, e una partizione che separa {A, B} da {C}. Il sistema CP richiede la maggioranza per scrivere e per leggere; il sistema AP scrive sempre localmente e riconcilia dopo con «l'ultima scrittura vince» (last write wins, LWW, basata su un timestamp).

from dataclasses import dataclass, field

class Indisponibile(Exception):
    pass

@dataclass
class Replica:
    nome: str
    valore: str = "v0"
    ts: int = 0                      # marca temporale logica della scrittura

@dataclass
class Cluster:
    repliche: dict
    partizione: list = field(default_factory=list)   # gruppi di nodi che si vedono fra loro

    def raggiungibili(self, da: str) -> list:
        if not self.partizione:
            return list(self.repliche)
        return next(g for g in self.partizione if da in g)

    # CP: la scrittura richiede la MAGGIORANZA, altrimenti rifiuta
    def scrivi_cp(self, da: str, valore: str, ts: int) -> str:
        gruppo = self.raggiungibili(da)
        if len(gruppo) <= len(self.repliche) // 2:
            raise Indisponibile(f"{da} vede solo {len(gruppo)} su {len(self.repliche)}: scrittura rifiutata")
        for n in gruppo:
            self.repliche[n].valore, self.repliche[n].ts = valore, ts
        return f"{da}: scritto '{valore}' su {len(gruppo)} nodi"

    # CP: anche la lettura richiede la maggioranza, altrimenti rifiuta
    def leggi_cp(self, da: str) -> str:
        gruppo = self.raggiungibili(da)
        if len(gruppo) <= len(self.repliche) // 2:
            raise Indisponibile(f"{da} vede solo {len(gruppo)} su {len(self.repliche)}: lettura rifiutata")
        return max((self.repliche[n] for n in gruppo), key=lambda r: r.ts).valore

    # AP: scrive SEMPRE in locale (e nei nodi raggiungibili), riconcilia dopo
    def scrivi_ap(self, da: str, valore: str, ts: int) -> str:
        for n in self.raggiungibili(da):
            self.repliche[n].valore, self.repliche[n].ts = valore, ts
        return f"{da}: scritto '{valore}' su {len(self.raggiungibili(da))} nodi"

    def ripara_lww(self) -> None:           # last-write-wins: vince il timestamp piu' alto
        vincitore = max(self.repliche.values(), key=lambda r: (r.ts, r.nome))
        for r in self.repliche.values():
            r.valore, r.ts = vincitore.valore, vincitore.ts

def nuovo() -> Cluster:
    c = Cluster({n: Replica(n) for n in "ABC"})
    c.partizione = [["A", "B"], ["C"]]       # la rete separa {A,B} da {C}
    return c

print("--- CP con partizione {A,B} | {C}")
cp = nuovo()
print(cp.scrivi_cp("A", "da-A", ts=1))
try:
    cp.scrivi_cp("C", "da-C", ts=2)
except Indisponibile as e:
    print("C:", e)
print("lettura su A:", cp.leggi_cp("A"))
try:
    cp.leggi_cp("C")
except Indisponibile as e:
    print("C:", e)

print("--- AP con la stessa partizione")
ap = nuovo()
print(ap.scrivi_ap("A", "da-A", ts=1))
print(ap.scrivi_ap("C", "da-C", ts=2))
print("prima della riparazione:", {n: r.valore for n, r in ap.repliche.items()})
ap.partizione = []
ap.ripara_lww()
print("dopo la riparazione LWW:", {n: r.valore for n, r in ap.repliche.items()}, "-> la scrittura 'da-A' e' andata persa")

# Output:
# --- CP con partizione {A,B} | {C}
# A: scritto 'da-A' su 2 nodi
# C: C vede solo 1 su 3: scrittura rifiutata
# lettura su A: da-A
# C: C vede solo 1 su 3: lettura rifiutata
# --- AP con la stessa partizione
# A: scritto 'da-A' su 2 nodi
# C: scritto 'da-C' su 1 nodi
# prima della riparazione: {'A': 'da-A', 'B': 'da-A', 'C': 'da-C'}
# dopo la riparazione LWW: {'A': 'da-C', 'B': 'da-C', 'C': 'da-C'} -> la scrittura 'da-A' e' andata persa

Nel sistema CP, A scrive su due nodi su tre (la maggioranza) e può anche leggere; C, che vede solo se stesso, rifiuta sia la scrittura sia la lettura: è indisponibile, ma non può mai restituire un dato vecchio. Nel sistema AP entrambe le parti accettano la scrittura; alla ricomposizione la regola LWW sceglie il timestamp più alto e da-A sparisce senza traccia. Il costo dell'AP non è «un dato stale» ma un aggiornamento perso se la riconciliazione non è fatta con cura: per questo servono i vector clock e i CRDT, due sezioni più sotto.

Quorum: quando «maggioranza» significa davvero qualcosa

Un modo per regolare la consistenza è dire quante repliche devono rispondere a una scrittura (W) e a una lettura (R), su N totali. La documentazione di Cassandra lo presenta come il meccanismo di Dynamo: se R + W > N, le scritture risultano visibili alle letture successive perché i due insiemi di repliche si sovrappongono. Il programma seguente lo verifica per enumerazione, provando ogni coppia possibile di insiemi di repliche.

from itertools import combinations

def intersezione_garantita(n: int, w: int, r: int) -> bool:
    """Ogni insieme di W repliche (scrittura) incontra ogni insieme di R repliche (lettura)?"""
    repliche = range(n)
    return all(set(a) & set(b)
               for a in combinations(repliche, w)
               for b in combinations(repliche, r))

for n, w, r in [(3, 2, 2), (3, 1, 1), (3, 3, 1), (3, 1, 3), (5, 3, 3), (5, 2, 3)]:
    print(f"N={n} W={w} R={r}  W+R>N: {w + r > n!s:<5}  ogni lettura incontra l'ultima scrittura: {intersezione_garantita(n, w, r)}")

# Output:
# N=3 W=2 R=2  W+R>N: True   ogni lettura incontra l'ultima scrittura: True
# N=3 W=1 R=1  W+R>N: False  ogni lettura incontra l'ultima scrittura: False
# N=3 W=3 R=1  W+R>N: True   ogni lettura incontra l'ultima scrittura: True
# N=3 W=1 R=3  W+R>N: True   ogni lettura incontra l'ultima scrittura: True
# N=5 W=3 R=3  W+R>N: True   ogni lettura incontra l'ultima scrittura: True
# N=5 W=2 R=3  W+R>N: False  ogni lettura incontra l'ultima scrittura: False

Con N=3, W=2, R=2 ogni scrittura e ogni lettura condividono almeno una replica; con W=1, R=1 no, e una lettura può mancare l'ultima scrittura. Anche W=3, R=1 e W=1, R=3 funzionano, con costi opposti: nel primo la scrittura aspetta tutte le repliche, nel secondo è la lettura. Attenzione a non sovrastimare il risultato: l'articolo di Abadi osserva che aumentare R + W dà più consistenza al prezzo di più latenza, ma che non si ottiene la consistenza piena nel senso di Gilbert e Lynch (la linearizzabilità) nemmeno con R + W > N. L'intersezione dei quorum è una condizione necessaria, non sufficiente, per garantire che le letture vedano sempre l'ultima scrittura.

Un menù di consistenze, dal più forte al più debole

Fra «consistente» e «inconsistente» c'è uno spettro. Linearizzabilità: ogni operazione sembra avvenire in un istante fra la richiesta e la risposta, e tutti vedono lo stesso ordine; è il modello più forte e richiede coordinamento. Consistenza sequenziale: tutti concordano su un unico ordine delle operazioni, ma non è tenuto a rispettare il tempo reale. Consistenza causale: le operazioni legate da causa ed effetto sono viste nello stesso ordine da tutti, quelle indipendenti in ordine qualunque. Consistenza eventuale: se smettono le scritture, le repliche convergono, senza garanzie su quando. Non riporto qui quali prodotti offrano quale modello: dipende dalla configurazione, e le classifiche in giro sono spesso approssimate.

La consistenza causale è il livello che interessa il prossimo strumento: per offrirla, il sistema deve saper dire se due eventi sono ordinati o concorrenti.

Vector clock: riconoscere la concorrenza

L'orologio di sistema non basta: due nodi hanno orologi diversi, e LWW lo ha appena mostrato. Lamport, nel suo articolo del luglio 1978 sugli orologi logici, mostra come ordinare gli eventi con contatori incrementati a ogni messaggio; quella costruzione dà un ordine, ma non permette di riconoscere due eventi concorrenti. I vector clock (un contatore per ogni nodo) sono un'evoluzione successiva di quell'idea, e sono quelli che servono. Attribuirli a Lamport 1978, come si legge spesso, è un errore: l'articolo del 1978 parla di orologi logici scalari.

from dataclasses import dataclass, field

@dataclass(frozen=True)
class VectorClock:
    clocks: dict = field(default_factory=dict)

    def incrementa(self, nodo: str) -> "VectorClock":
        nuovo = dict(self.clocks)
        nuovo[nodo] = nuovo.get(nodo, 0) + 1
        return VectorClock(nuovo)

    def unisci(self, altro: "VectorClock") -> "VectorClock":
        chiavi = set(self.clocks) | set(altro.clocks)
        return VectorClock({k: max(self.clocks.get(k, 0), altro.clocks.get(k, 0)) for k in chiavi})

    def precede(self, altro: "VectorClock") -> bool:
        chiavi = set(self.clocks) | set(altro.clocks)
        return (all(self.clocks.get(k, 0) <= altro.clocks.get(k, 0) for k in chiavi)
                and any(self.clocks.get(k, 0) < altro.clocks.get(k, 0) for k in chiavi))

    def concorrente(self, altro: "VectorClock") -> bool:
        return not self.precede(altro) and not altro.precede(self)

base = VectorClock()
a1 = base.incrementa("A")                 # A modifica il carrello
b1 = a1.incrementa("B")                   # B ha ricevuto a1, poi modifica: causalmente DOPO a1
c1 = a1.incrementa("C")                   # C, che ha visto solo a1, modifica senza sapere di B

print("a1", a1.clocks, "precede b1", b1.clocks, ":", a1.precede(b1))
print("b1 e c1 sono concorrenti:", b1.concorrente(c1))
print("a1 e c1 sono concorrenti:", a1.concorrente(c1), "(a1 precede c1:", a1.precede(c1), ")")
print("unione b1+c1:", b1.unisci(c1).clocks)

# Output:
# a1 {'A': 1} precede b1 {'A': 1, 'B': 1} : True
# b1 e c1 sono concorrenti: True
# a1 e c1 sono concorrenti: False (a1 precede c1: True )
# unione b1+c1: {'C': 1, 'A': 1, 'B': 1}

precede confronta due vettori: A precede B se ogni contatore di A è minore o uguale a quello di B e almeno uno è strettamente minore. b1 deriva da a1 (B l'aveva ricevuto), quindi a1 lo precede; b1 e c1, invece, sono due modifiche indipendenti di uno stesso stato, e nessuno dei due precede l'altro: sono concorrenti. In un sistema AP questo è esattamente il segnale che serve: invece di sovrascrivere come LWW, il sistema sa che deve conservare entrambe le versioni e chiedere di unirle (all'applicazione, o a un CRDT). L'unione dei due vettori ha un contatore massimo per nodo.

CRDT: strutture dati che si uniscono da sole

Un CRDT (conflict-free replicated data type) è una struttura dati pensata per essere replicata: ogni replica si aggiorna da sola, senza coordinarsi, e le repliche si scambiano lo stato; l'unione porta sempre allo stesso risultato. Preguiça, Baquero e Shapiro riassumono la condizione per i CRDT basati su stato: gli stati formano un join semilattice, ogni operazione produce uno stato maggiore o uguale, e l'unione è l'estremo superiore dei due stati. In parole povere: l'unione deve essere commutativa, associativa e idempotente, così l'ordine e le ripetizioni dei messaggi non contano.

Il listato implementa tre CRDT (G-Counter, PN-Counter, OR-Set) e verifica le tre proprietà su 2.000 triple di stati casuali per due di essi, poi riproduce il caso più istruttivo dell'OR-Set: un'aggiunta e una rimozione concorrenti dello stesso elemento.

import random
from dataclasses import dataclass, field

@dataclass(frozen=True)
class GCounter:
    """Contatore che cresce: un contatore parziale per nodo."""
    conteggi: tuple = ()                       # coppie (nodo, valore), ordinate

    def _d(self): return dict(self.conteggi)

    def incrementa(self, nodo: str, n: int = 1) -> "GCounter":
        d = self._d(); d[nodo] = d.get(nodo, 0) + n
        return GCounter(tuple(sorted(d.items())))

    def valore(self) -> int: return sum(v for _, v in self.conteggi)

    def unisci(self, altro: "GCounter") -> "GCounter":
        a, b = self._d(), altro._d()
        return GCounter(tuple(sorted((k, max(a.get(k, 0), b.get(k, 0))) for k in a.keys() | b.keys())))

@dataclass(frozen=True)
class PNCounter:
    """Contatore che cresce e cala: due G-Counter."""
    pos: GCounter = GCounter()
    neg: GCounter = GCounter()

    def incrementa(self, nodo): return PNCounter(self.pos.incrementa(nodo), self.neg)
    def decrementa(self, nodo): return PNCounter(self.pos, self.neg.incrementa(nodo))
    def valore(self): return self.pos.valore() - self.neg.valore()
    def unisci(self, altro): return PNCounter(self.pos.unisci(altro.pos), self.neg.unisci(altro.neg))

@dataclass(frozen=True)
class ORSet:
    """Observed-Remove Set: ogni add porta un tag unico; remove cancella solo i tag che ha VISTO."""
    aggiunte: frozenset = frozenset()          # coppie (elemento, tag)
    rimosse: frozenset = frozenset()           # coppie (elemento, tag) cancellate

    def add(self, elemento, tag): return ORSet(self.aggiunte | {(elemento, tag)}, self.rimosse)
    def remove(self, elemento):
        visti = {(e, t) for (e, t) in self.aggiunte if e == elemento}
        return ORSet(self.aggiunte, self.rimosse | visti)
    def contiene(self, elemento): return any(e == elemento for (e, t) in self.aggiunte - self.rimosse)
    def unisci(self, altro): return ORSet(self.aggiunte | altro.aggiunte, self.rimosse | altro.rimosse)

# --- Proprieta' del merge, verificate su 2000 terne di stati casuali ---------------------------
random.seed(1)
def g_casuale():
    g = GCounter()
    for _ in range(random.randint(0, 6)): g = g.incrementa(random.choice("ABC"), random.randint(1, 5))
    return g

def or_casuale():
    s = ORSet()
    for i in range(random.randint(0, 6)):
        s = s.add(random.choice("xyz"), f"t{random.randint(0, 9)}")
        if random.random() < 0.3: s = s.remove(random.choice("xyz"))
    return s

for nome, gen in [("G-Counter", g_casuale), ("OR-Set", or_casuale)]:
    comm = assoc = idem = 0
    for _ in range(2000):
        a, b, c = gen(), gen(), gen()
        comm += a.unisci(b) == b.unisci(a)
        assoc += a.unisci(b).unisci(c) == a.unisci(b.unisci(c))
        idem += a.unisci(a) == a
    print(f"{nome:<10} commutativa {comm}/2000, associativa {assoc}/2000, idempotente {idem}/2000")

# --- Convergenza di un PN-Counter: tre repliche, aggiornamenti isolati, unioni in ordini diversi --
a = PNCounter().incrementa("A").incrementa("A")        # A: +2
b = PNCounter().incrementa("B").decrementa("B")         # B: +1 -1
c = PNCounter().decrementa("C")                         # C: -1
x = a.unisci(b).unisci(c)
y = c.unisci(a).unisci(b).unisci(a)                     # ordine diverso, con un merge ripetuto
print("PN-Counter: valore", x.valore(), "e", y.valore(), "- stesso risultato:", x == y)

# --- OR-Set: add concorrente e remove dello stesso elemento ----------------------------------------
base = ORSet().add("latte", "A1")
replica_a = base.remove("latte")                         # A toglie il latte che ha visto (tag A1)
replica_b = base.add("latte", "B1")                      # B, senza saperlo, lo riaggiunge (tag B1)
unito = replica_a.unisci(replica_b)
print("OR-Set: A rimuove, B aggiunge in concorrenza ->", "latte presente" if unito.contiene("latte") else "latte assente",
      "(vince l'add)", "| lo stesso a ordine invertito:", replica_b.unisci(replica_a).contiene("latte"))

# Output:
# G-Counter  commutativa 2000/2000, associativa 2000/2000, idempotente 2000/2000
# OR-Set     commutativa 2000/2000, associativa 2000/2000, idempotente 2000/2000
# PN-Counter: valore 1 e 1 - stesso risultato: True
# OR-Set: A rimuove, B aggiunge in concorrenza -> latte presente (vince l'add) | lo stesso a ordine invertito: True

Il G-Counter tiene un contatore per nodo e l'unione prende il massimo di ciascuno, quindi nessun incremento si perde anche se le comunicazioni si duplicano; il PN-Counter ne combina due per permettere decrementi. Le 2.000 prove per G-Counter e OR-Set danno sempre 2.000 su 2.000 per commutatività, associatività e idempotenza: non è una dimostrazione, è un controllo che il codice mantiene le promesse. Il PN-Counter converge a 1 sia unendo in un ordine sia in un altro, con un'unione ripetuta. Nell'OR-Set la rimozione cancella solo i tag che ha visto: la replica B ha aggiunto «latte» con un tag nuovo, che A non conosceva, quindi dopo l'unione l'elemento c'è. Per questo si chiama semantica add-wins, e il risultato è lo stesso in qualunque ordine.

Il limite dei CRDT è altrettanto concreto: il tipo di dato decide la semantica. Un contatore va bene per le visite, un insieme con add-wins per un carrello (con la sorpresa che un articolo rimosso su un dispositivo può riapparire), ma non esiste un CRDT per «il saldo non può scendere sotto zero»: i vincoli globali richiedono coordinamento.

PACELC: il compromesso che CAP non vede

Abadi estende CAP con PACELC: se c'è una partizione (P) si sceglie fra disponibilità e consistenza (A/C); altrimenti (E, else) si sceglie fra latenza e consistenza (L/C). Il suo articolo classifica Dynamo, Cassandra e Riak nella configurazione predefinita come PA/EL, cioè consistenza ceduta in entrambi i casi; sistemi completamente ACID come VoltDB e Megastore come PC/EC; MongoDB come PA/EC. L'ultima classificazione smentisce chi scrive che «PA/EC non esiste»: è nella tabella dell'articolo originale. Le classificazioni valgono per le versioni e le configurazioni del 2012 e vanno ricontrollate su quelle in uso.

Transazioni distribuite: la saga

Quando una operazione tocca più servizi, il commit a due fasi è spesso troppo costoso. La saga lo sostituisce con una sequenza di transazioni locali, ciascuna con la propria compensazione: se un passo fallisce, si eseguono le compensazioni dei passi già completati. La descrizione di microservices.io lo dice così: o tutti i passi riescono, o i cambiamenti dei passi completati vengono annullati da transazioni compensative.

# Saga: una sequenza di transazioni locali, ciascuna con la sua compensazione.
# Se un passo fallisce, si eseguono le compensazioni dei passi gia' completati, in ordine inverso.
def crea_ordine(s):        s["ordine"] = "creato"
def annulla_ordine(s):     s["ordine"] = "annullato"
def riserva_scorta(s):     s["scorta"] = "riservata"
def rilascia_scorta(s):    s["scorta"] = "rilasciata"
def paga(s):               raise RuntimeError("pagamento rifiutato")
def rimborsa(s):           s["pagamento"] = "rimborsato"

PASSI = [(crea_ordine, annulla_ordine), (riserva_scorta, rilascia_scorta), (paga, rimborsa)]

def esegui_saga(stato: dict) -> bool:
    completati = []
    try:
        for azione, compensazione in PASSI:
            azione(stato)
            completati.append(compensazione)
        return True
    except Exception as errore:
        print("fallimento:", errore)
        for compensazione in reversed(completati):    # ordine inverso
            print("  compenso con", compensazione.__name__)
            compensazione(stato)
        return False

stato = {}
print("saga riuscita:", esegui_saga(stato))
print("stato finale:", stato)

# Output:
# fallimento: pagamento rifiutato
#   compenso con rilascia_scorta
#   compenso con annulla_ordine
# saga riuscita: False
# stato finale: {'ordine': 'annullato', 'scorta': 'rilasciata'}

Il pagamento fallisce: il programma compensa solo i passi completati (rilascia_scorta e annulla_ordine), in ordine inverso, e non il pagamento stesso, che non è mai riuscito. Lo stato finale è coerente ma non è isolato: per un istante qualcuno ha visto l'ordine creato e la scorta riservata. Una saga garantisce una atomicità tardiva, non l'isolamento di una transazione; le compensazioni devono essere idempotenti, perché possono essere rieseguite. Un'annotazione @Transactional attorno a una saga, come si vede in certi esempi, è un controsenso: un commit locale non copre servizi remoti.

Che cosa sai fare adesso

Sai che CAP obbliga a scegliere fra consistenza e disponibilità solo durante una partizione, e che il resto del tempo il compromesso è fra latenza e consistenza. Sai verificare quando un quorum garantisce l'intersezione, riconoscere con un vector clock due modifiche concorrenti, scegliere un CRDT quando non vuoi perdere aggiornamenti, e comporre una saga con le sue compensazioni. Soprattutto, sai quando una risposta «dipende» è la più onesta: dalla perdita che puoi sopportare, dalla latenza che puoi permetterti e dall'operazione che stai proteggendo.

È l'ultimo articolo della serie. Il punto di partenza è il confronto fra i modelli di concorrenza; per il tema dell'ordine nel singolo processo, vedi strutture dati lock-free, e per i flussi, backpressure.

Non ho raggiunto il testo originale di Gilbert e Lynch (il PDF non è risultato leggibile da riga di comando): la formulazione di CAP qui riportata si appoggia all'articolo di Abadi che la cita. Le uscite delle simulazioni sono deterministiche, tranne l'ordine dei nodi nell'unione dei vector clock, che dipende dal seed di hash di Python.