Il framework di riferimento del corso e Apache Pekko, fork open-source di Akka con API identica (package org.apache.pekko). Lezione e slide usano i due nomi in modo intercambiabile; in questo capitolo normalizziamo a Pekko, mantenendo invariati i frammenti di codice.
La lezione inizia con un argomento pratico: i timer in Pekko. Nella lezione precedente avevamo introdotto il modello actor e l'API base di Pekko Typed, ma non avevamo fatto in tempo a parlare di come un attore possa compiere azioni in modo proattivo, non solo reagendo a messaggi ricevuti. L'attore puramente reattivo resta in attesa di un messaggio; se nessuno gli scrive, rimane bloccato. I timer risolvono questo problema: permettono a un attore di "svegliarsi" periodicamente o dopo un certo intervallo per compiere un'azione.
Gli attori sono entita puramente reattive: processano messaggi quando arrivano. I timer sono il meccanismo per introdurre comportamenti proattivi, ovvero azioni avviate dall'attore stesso in base allo scorrere del tempo.
Il costruttore fondamentale e Behaviors.withTimers. Questo factory method prende una funzione che riceve due parametri: il context dell'attore (come abbiamo visto nelle lezioni precedenti) e un oggetto timers di tipo TimerScheduler. Quest'ultimo e l'interfaccia attraverso cui programmiamo l'invio di messaggi futuri.
Ecco il codice mostrato a lezione: un attore che, una volta avviato, si autoschedula per ricevere un messaggio TIC ogni secondo.
La trascrizione mostra il codice in modo frammentario. Il frammento qui sopra ricostruisce il pattern standard di Pekko Typed con Behaviors.withTimers e un messaggio TIC schedulato periodicamente. Il professore ha enfatizzato che l'import di scala.concurrent.duration._ e necessario per la sintassi 1.second.
L'idea centrale e che timerScheduler.startTimerAtFixedRate fa si che l'attore riceva automaticamente il messaggio TIC ogni intervallo specificato (nel caso del professore, 1 secondo). Il timer continua all'infinito fino a quando non viene cancellato esplicitamente o l'attore viene fermato.
Questo pattern e estremamente comune in molte applicazioni: heartbeats, polling periodico, timeout per richieste, refresh di cache, scheduling di job ricorrenti. Nel contesto dell'assignment del corso, il professore ha sottolineato che i timer saranno molto probabilmente necessari.
Il professore ha esplicitamente detto: "questa roba dei timer vi servira nell'assignment". Ricordate i tre metodi principali: startSingleTimer (un solo scatto), startTimerAtFixedRate (periodico, si riprogramma da solo), e la possibilita di cancellare un timer con timers.cancel(key). Vedi la scheda di preparazione dell'Assignment 4, che si appoggia direttamente a questi meccanismi.
Il TimerScheduler offre diverse varianti del metodo di scheduling, ciascuna pensata per un caso d'uso specifico:
| Metodo | Comportamento | Quando usarlo |
|---|---|---|
startSingleTimer(key, msg, delay) |
Invia msg una sola volta dopo delay |
Timeout, azioni dilazionate, retry |
startTimerAtFixedRate(key, msg, interval) |
Invia msg ripetutamente ogni interval |
Heartbeat, polling, tick periodici |
startTimerWithFixedDelay(key, msg, delay) |
Come fixed rate ma aspetta che il messaggio precedente sia processato | Task che non devono sovrapporsi |
Un aspetto importante emerso dalla lezione e il parametro key. La key e un Any: puo essere qualsiasi oggetto, e serve a disambiguare timer diversi. Se avete lo stesso identico messaggio da mandare ma volete due timer indipendenti, usate due chiavi diverse. In questo modo l'attore sa gestire piu timer concorrenti, ciascuno con la propria periodicita o scadenza. La key permette anche di cancellare selettivamente un timer: timers.cancel(chiave).
Se chiamate startSingleTimer con una chiave gia attiva, il timer precedente viene cancellato e sostituito dal nuovo. Questo comportamento e utile per implementare timeout estendibili (es. "ricevo un messaggio, riprogrammo il timeout").
flowchart LR
subgraph Attore
SM[TimerScheduler]
MB[Message Queue]
BH[Behavior]
end
SM -->|"startSingleTimer(k,msg,d)"| T1[Timer k: scatta dopo d]
SM -->|"startTimerAtFixedRate(k,msg,i)"| T2[Timer k: tic ogni i]
T1 -->|scaduto| MB
T2 -->|scaduto| MB
MB --> BH
Il professore ha dedicato un passaggio importante alla combinazione di stash e timer. Lo stash, introdotto nella lezione precedente, permette di accodare messaggi ricevuti mentre l'attore si trova in uno stato transitorio (ad esempio in attesa di una risposta), per poi riprocessarli in seguito con unstashAll.
La combinazione con i timer e particolarmente utile per gestire timeout nelle conversazioni. Supponiamo che un attore A invii una richiesta a un attore B e si metta in attesa della risposta. Durante l'attesa, A usa lo stash per accumulare gli altri messaggi. Con un timer singolo, A puo anche gestire il caso in cui B non risponda entro un certo tempo: allo scadere del timer, A puo decidere se riprovare, fallire, o tornare allo stato normale e processare i messaggi accumulati.
Usare stash + timer in combinazione risolve elegantemente due problemi contemporaneamente: (1) non perdere messaggi durante uno stato di attesa, (2) non rimanere bloccati in attesa per sempre grazie al timeout.
// Pattern: attore che attende una risposta con timeout e stash
Behaviors.withTimers { (ctx, timers) =>
Behaviors.withStash(100) { stash =>
Behaviors.receiveMessage {
case Richiesta(destinatario, payload) =>
destinatario ! Esegui(payload, ctx.self)
timers.startSingleTimer("timeout", TimeoutScattato, 5.second)
Behaviors.receiveMessage {
case RisultatoOk(data) =>
timers.cancel("timeout")
stash.unstashAll(comportamentoNormale(data))
case TimeoutScattato =>
stash.unstashAll(comportamentoFallito)
case altro =>
stash.stash(altro)
Behaviors.same
}
}
}
}Nel pattern qui sopra, l'attore cambia behavior in attesa della risposta. Tutti i messaggi che arrivano durante l'attesa vengono stashati. Se arriva RisultatoOk, il timer viene cancellato e i messaggi accumulati vengono riprocessati. Se scatta il timeout, si passa a un behavior di fallimento e anch'esso processa i messaggi accumulati.
Dopo la parte introduttiva sui timer, il professore ha dedicato il nucleo centrale della lezione agli aspetti distribuiti di Pekko. Passare da un singolo actor system locale a un cluster di piu nodi introduce complessita sostanziali: i nodi possono unirsi o lasciare il cluster, possono fallire senza preavviso, la rete puo partizionarsi. Pekko gestisce tutto questo con un meccanismo di membership basato su gossip protocol.
Quando un nuovo nodo vuole unirsi al cluster, deve contattare uno o piu seed nodes. I seed sono nodi noti (tipicamente configurati tramite DNS o file di configurazione) che fungono da punti di contatto iniziali. Il professore ha mostrato due modalita:
Join al seed node, che lo inserisce nel cluster.I seed node sono consapevoli l'uno dell'altro: quando un seed si avvia, fa join del cluster da solo (oppure si mette in ascolto). In un setup Docker Compose (come quello preparato dal professore), i seed vengono referenziati tramite nome del servizio DNS interno.
Il professore ha raccomandato di non usare mai 127.0.0.1 negli indirizzi dei seed, perche in un contesto distribuito (anche simulato con Docker) ogni nodo vede se stesso su IP diversi. Usate i nomi simbolici dei servizi: docker-compose mette a disposizione un DNS interno.
Una volta che il cluster e formato, i nodi comunicano tra loro usando un gossip protocol per scambiarsi informazioni sullo stato del cluster: quali nodi sono vivi, quali sono morti, qual e lo stato corrente della membership. Il gossip e un protocollo epidemico: ogni nodo periodicamente condivide con un sottoinsieme di altri nodi tutto cio che sa. Con il tempo, l'informazione si propaga a tutto il cluster. Questo meccanismo e alla base del failure detection: se un nodo smette di fare gossip, gli altri lo marcano come unreachable.
flowchart TD
subgraph Cluster Pekko
N1[Nodo 1] --- N2[Nodo 2]
N2 --- N3[Nodo 3]
N3 --- N4[Nodo 4]
N1 -.- N4
end
S1[Seed 1:2552] --> N1
S2[Seed 2:2553] --> N2
SN[Nuovo Nodo] -->|join| S1
SN -->|gossip| N2
SN -->|gossip| N3
Uno dei problemi piu insidiosi nei cluster distribuiti e lo split-brain: quando un nodo o un gruppo di nodi non riesce piu a comunicare con un'altra parte del cluster a causa di un guasto di rete, si creano due (o piu) partizioni che non si parlano. Dal punto di vista di ciascuna partizione, l'altra sembra "morta".
Se entrambe le partizioni continuano a operare indipendentemente, e poi la rete viene ripristinata, ci si trova con due stati divergenti dello stesso sistema. Questo e particolarmente grave se ci sono attori persistenti con stato: quale dei due stati va mantenuto? Non e banale, anzi a volte non si saprebbe neppure come scegliere.
La strategia di default in Pekko Cluster e down una delle due partizioni (graceful degradation) piuttosto che tenerle entrambe vive con stati divergenti.
Pekko implementa la strategia della partizione piu numerosa: il sottocluster con il maggior numero di nodi sopravvive, mentre l'altro viene spento. Ma cosa succede in caso di parita (es. 2-2 su 4 nodi)? Il professore ha spiegato che esiste un meccanismo di tie-breaking che rompe la simmetria, basato su un ordinamento deterministico (es. indirizzi dei nodi). In questo modo solo una delle due partizioni sopravvive.
Il detection della partizione funziona perche il gossip protocol mantiene in ogni nodo la consapevolezza di quanti nodi ci sono nel cluster totale. Quando un nodo si accorge di poter comunicare solo con un sottoinsieme, capisce di essere in una partizione e applica la strategia di downing.
Il professore ha sottolineato che la split-brain resilience e particolarmente critica se si usa la persistenza degli attori. Pekko non replica automaticamente lo stato degli attori negli altri nodi. Se avete attori persistenti e si verifica uno split-brain, quando il cluster si ricompone potreste avere due versioni divergenti dello stesso attore. Ci sono strategie avanzate nella documentazione, ma vanno studiate e configurate esplicitamente.
In un contesto distribuito, gli attori comunicano sempre attraverso ActorRef, esattamente come in un contesto locale. La differenza e che l'ActorRef puo referenziare un attore che vive su un nodo diverso: quando inviamo un messaggio a quel ref, il sistema si occupa di serializzare il messaggio e spedirlo sulla rete al nodo giusto.
Il professore ha elencato tre modi per ottenere un ActorRef di un attore su un nodo remoto:
ctx.self a un attore remoto, che cosi ottiene un ref per rispondere. Questo e il modo piu comune.pekko://system@host:port/user/actorName), si puo chiedere al sistema di risolverlo. Tuttavia questo approccio e fragile perche il path puo cambiare.Quando si invia un messaggio attraverso la rete, e necessario serializzarlo. Pekko utilizza di default Jackson JSON per la serializzazione. Questo significa che tutti i messaggi scambiati tra attori in contesto distribuito devono essere (direttamente o indirettamente) serializzabili in JSON. Il professore ha accennato che non c'e molto altro da sapere se non che i messaggi custom vanno annotati o configurati per essere serializzabili correttamente.
Anche i messaggi che passano attraverso il receptionist devono essere serializzabili. Se usate tipi custom non standard, assicuratevi che Jackson possa serializzarli. Il professore ha detto che non e un problema nella pratica, ma va tenuto a mente.
Il receptionist e un attore di sistema che funge da registry distribuito: permette agli attori di registrarsi sotto una service key e ad altri attori di scoprirli. E la soluzione consigliata per il bootstrapping della comunicazione in un cluster Pekko.
Il receptionist offre tre operazioni fondamentali:
Register: un attore si registra con una chiave. Il receptionist associa l'ActorRef dell'attore a quella chiave.Find: richiede una istantanea (snapshot) degli attori attualmente registrati per una data chiave. Utile quando il set degli attori e stabile.Subscribe: si sottoscrive per ricevere notifiche ogni volta che un attore si registra o si deregistra per quella chiave. Utile quando il set e dinamico.Il professore ha enfatizzato la differenza tra find e subscribe: la find da una risposta una tantum, mentre la subscribe continua a notificare l'attore ogni volta che ci sono cambiamenti. La scelta dipende dal caso d'uso. Nell'esempio mostrato a lezione, un attore manager usa subscribe per ricevere notifiche man mano che i worker si registrano (visto che i worker potrebbero non essere ancora partiti al momento della subscribe).
Ecco il pattern mostrato dal professore:
// Registrazione (lato worker)
ctx.system.receptionist ! Receptionist.Register(WorkerServiceKey, ctx.self)
// Sottoscrizione (lato manager)
ctx.system.receptionist ! Receptionist.Subscribe(WorkerServiceKey, ctx.self)
// L'attore manager gestisce Listing per ottenere i worker
Behaviors.receiveMessage {
case WorkerServiceKey.Listing(actors) =>
// actors contiene la lista aggiornata dei worker
comportamentoAttivo(actors)
// ...
}Il receptionist e un ottimo modo per ottenere il primo contatto con attori remoti, ma il professore avverte: non sovraccaricatelo. Non e pensato per scalare a centinaia di migliaia di attori che fanno find/subscribe in continuazione. Usatelo per il bootstrap, non per il routing fine-grained. Nella preparazione dell'Assignment 4 il receptionist e il punto di partenza tipico per far scoprire i peer tra loro.
Quando usarlo: il set degli attori e stabile (es. un pool di worker che viene creato all'avvio e non cambia).
Comportamento: il receptionist risponde una sola volta con la lista corrente degli ActorRef associati alla chiave.
Vantaggio: semplice, un solo scambio di messaggi.
Limite: se un worker si registra dopo la find, non viene visto.
Quando usarlo: il set degli attori cambia nel tempo (worker che partono/arrivano dinamicamente).
Comportamento: il receptionist notifica l'attore ogni volta che un worker si registra o si deregistra.
Vantaggio: sempre aggiornato, adatto a scenari dinamici.
Limite: l'attore deve gestire i messaggi di notifica. Leggero overhead aggiuntivo.
Il cluster sharding e il meccanismo che permette di distribuire un insieme di attori (detti entities) attraverso i nodi del cluster in modo trasparente. E ideale per attori con identita univoca (es. un attore per ogni utente, per ogni sessione, per ogni conto corrente) dove non ci interessa su quale nodo l'attore vive, ma vogliamo che ci sia una distribuzione bilanciata.
Il professore ha mostrato un esempio semplice: un attore Counter che mantiene un contatore interno e risponde a messaggi Increment e GetValue. Lo scopo e creare piu istanze di Counter distribuite automaticamente nel cluster.
Per usare lo sharding:
Sharding con una EntityTypeKey che identifica il protocollo (i messaggi che l'attore sa gestire).ClusterSharding con l'entity factory (come creare l'attore quando serve).EntityRef ottenuta con sharding.entityRefFor(typeKey, entityId).// Inizializzazione dello sharding
val sharding = ClusterSharding(system)
val counterTypeKey = EntityTypeKey[Counter.Command]("counter")
val shard = sharding.init(
Entity(counterTypeKey, createBehavior = ctx => Counter())
)
// Ottenere un EntityRef e inviare un messaggio
val counter = sharding.entityRefFor(counterTypeKey, entityId = Random.nextInt(3).toString)
counter ! Counter.IncrementIn questo esempio, l'entityId e un numero casuale tra 1 e 3: questo fa si che gli attori counter vengano distribuiti su diversi nodi del cluster (in base alla funzione di hash dell'entityId). Il professore ha mostrato in diretta i log che evidenziano come i vari counter vengono creati su nodi diversi.
Il cluster sharding e strettamente legato alla split-brain resilience: se un nodo muore, le entity che vivevano su quel nodo vengono ricreate su un altro nodo dal meccanismo di sharding. Tuttavia, per attori persistenti, il recupero dello stato richiede un'attenta configurazione.
TimerScheduler?Behaviors.withTimers. Prende una funzione (ActorContext[T], TimerScheduler[T]) => Behavior[T] e restituisce un behavior che ha accesso allo scheduler dei timer.
startTimerAtFixedRate e startTimerWithFixedDelay?startTimerAtFixedRate tenta di mantenere l'intervallo fisso tra gli inizi di ogni attivazione, indipendentemente da quanto dura l'elaborazione. startTimerWithFixedDelay aspetta che l'elaborazione del tick precedente sia terminata prima di contare il ritardo successivo, evitando sovrapposizioni.
Un attore lavora solo quando riceve un messaggio: senza messaggi, rimane bloccato (reattivo puro). La pro-attivita si ottiene tramite timer (self-scheduling di messaggi futuri) o tramite attori multipli che si coordinano. Il timer permette all'attore di "svegliarsi" autonomamente e compiere azioni pianificate.
Per evitare stati divergenti. Se entrambe le partizioni continuano a operare indipendentemente, quando la rete viene ripristinata ci sono due versioni dello stesso stato (es. attori persistenti) che non sono riconciliabili. E meglio far sopravvivere una sola partizione (la piu numerosa) e sacrificare l'altra.
Find restituisce una tantum la lista attuale degli attori registrati per una service key. Subscribe registra l'attore per ricevere notifiche ogni volta che la lista cambia (nuove registrazioni o deregistrazioni). Find e per set statici, subscribe per set dinamici.
L'ordinamento causale garantisce che se send(m1) → send(m2) (m1 precede causalmente m2), allora rec(m1) → rec(m2) (m1 viene ricevuto prima di m2 ovunque). Il FIFO garantisce solo l'ordine tra coppie mittente-destinatario; il causale gestisce anche messaggi che transitano attraverso processi intermedi.
Usando canali FIFO e il meccanismo dei marker. Un processo che riceve un marker su un canale FIFO si "tinge di rosso" e salva lo stato locale; poiche i canali sono FIFO, nessun messaggio inviato da un processo bianco puo arrivare dopo il marker a un processo rosso. Questo assicura che la ricezione di un messaggio non preceda causalmente il suo invio nello snapshot.
In un sistema asincrono, anche con un solo processo che puo crashare (unannounced process death), il problema del consenso e impossibile da risolvere deterministicamente. Non esiste un algoritmo che garantisca sempre termination, agreement e integrity in queste condizioni.
Ricart-Agrawala: 2(N-1) messaggi per accesso alla CS (N-1 richieste + N-1 OK). Chang-Roberts: nel caso peggiore 2N-1 messaggi di elezione + N messaggi di leader propagation.