Attori avanzati: timer e cluster

1. Timer: un attore che agisce da solo

Un attore puramente reattivo, se nessuno gli scrive, resta fermo; il capitolo 15 lo rende proattivo con messaggi a se stesso. Quando il lavoro dipende dal tempo (un battito al secondo, un timeout, un nuovo tentativo) in Pekko si usa un timer: un messaggio che l'attore si fa recapitare nel futuro. Resta tutto scambio di messaggi, e il macro-step vale come sempre. Il TimerScheduler si ottiene avvolgendo il comportamento in Behaviors.withTimers:

Nessuno dei due timer periodici aspetta che l'attore abbia elaborato il messaggio precedente: entrambi misurano il tempo tra gli invii, e se l'attore è lento i messaggi si accodano in mailbox.

La chiave è un oggetto qualsiasi che identifica il timer: con chiavi diverse lo stesso messaggio può avere timer indipendenti. Avviare un timer con una chiave già attiva cancella il vecchio, ed è garantito che il suo messaggio non arrivi più, nemmeno se era già in mailbox: così si fa un timeout che riparte a ogni messaggio. Nelle varianti senza chiave la chiave è il messaggio stesso. I timer si cancellano quando l'attore si ferma o viene riavviato, e il TimerScheduler va usato solo dentro i suoi handler.

Un esempio del corso è un attore con due stati. In idle aspetta Start, avvia un timer di due secondi e passa in active; allo scadere riceve Ticked e torna in idle.

public class Blinker {
  public sealed interface Command permits Start, Ticked {}
  public record Start() implements Command {}
  public record Ticked() implements Command {}

  public static Behavior<Command> create() {
    return Behaviors.withTimers(Blinker::idle);
  }

  static Behavior<Command> idle(TimerScheduler<Command> timers) {
    return Behaviors.receive(Command.class)
        .onMessage(Start.class, s -> {
          timers.startSingleTimer("blink", new Ticked(), Duration.ofSeconds(2));
          return active(timers);
        })
        .build();
  }

  static Behavior<Command> active(TimerScheduler<Command> timers) {
    return Behaviors.receive(Command.class)
        .onMessage(Ticked.class, t -> idle(timers))
        .build();              // nessun caso per Start: in active è «non gestito»
  }
}

Se arrivano tre Start di fila, il primo avvia il timer e gli altri due arrivano in active, dove non c'è un caso per loro. Non aspettano in mailbox: vengono scartati come non gestiti, e il log li conta tra le dead letters. Con un comportamento parziale bisogna decidere che cosa fare degli altri messaggi: gestirli (per esempio riavviando il timer con la stessa chiave) o metterli nello stash.

2. Timeout di una conversazione: timer e stash

Lo stash (capitolo 15) mette da parte i messaggi che l'attore non può gestire nello stato attuale. Combinato con un timer risolve il caso tipico di una richiesta a un altro attore: mentre aspetta la risposta l'attore non perde gli altri messaggi, e non aspetta per sempre se la risposta non arriva.

static Behavior<Command> waiting(TimerScheduler<Command> timers, StashBuffer<Command> stash) {
  return Behaviors.receive(Command.class)
      .onMessage(Reply.class, r -> {
        timers.cancel("timeout");
        return stash.unstashAll(ready(r.data()));     // risposta: riprendo il resto
      })
      .onMessage(TimedOut.class, t -> stash.unstashAll(failed()))
      .onMessage(Command.class, m -> {                // tutto il resto aspetta
        stash.stash(m);
        return Behaviors.same();
      })
      .build();
}
// entrando in waiting: service.tell(new Job(payload, ctx.getSelf()));
//                      timers.startSingleTimer("timeout", new TimedOut(), Duration.ofSeconds(5));

ready e failed decidono se proseguire, riprovare o segnalare l'errore. Lo stash ha la capacità fissata in Behaviors.withStash(capacità, …), e se si riempie stash lancia un'eccezione. Timer e stash servono nel sistema d'allarme dell'assignment 3, e la parte temporale resta la stessa nella versione distribuita dell'assignment 4.

3. Da un nodo a molti: Artery, indirizzi e serializzazione

Per la location transparency il codice che invia un messaggio non cambia se il destinatario è su un'altra macchina: si usa sempre un ActorRef. Sotto, i messaggi tra nodi e il gossip del cluster passano per Artery, il sottosistema di remoting di Pekko, peer-to-peer, con tre trasporti: tcp, tls-tcp se il traffico va cifrato, aeron-udp per throughput alto e bassa latenza a un costo maggiore di CPU. Il remoting non si usa direttamente ma attraverso il cluster, e non va mai esposto fuori dalla rete del cluster; verso l'esterno si usano servizi come HTTP o gRPC.

Ogni attore ha un indirizzo univoco nel cluster, il path: pekko://ClusterSystem@node1:2551/user/coordinator dà protocollo, actor system, host e porta del nodo, posizione nell'albero. Host e porta sono quelli di canonical.hostname e canonical.port, l'indirizzo che il nodo comunica agli altri: 127.0.0.1 funziona solo se tutti i nodi sono sulla stessa macchina, e in Docker o Kubernetes si usano i nomi dei servizi.

Un ActorRef remoto si ottiene in tre modi:

1. dentro un messaggio, come il replyTo di una richiesta: è il modo più comune;
2. dal path, strada fragile perché lega il codice alla posizione dell'attore; l'API typed non la offre, solo quella classica (ActorSelection);
3. dal receptionist (sezione 6), che risolve anche il primo contatto, quando non si conosce ancora nessuno.

Tra attori della stessa JVM i messaggi passano per riferimento; verso un altro nodo vanno serializzati. Lo standard è Jackson, in JSON o in CBOR (binario e compatto), ma si può usare anche Protobuf. Ogni messaggio remoto va associato a un serializzatore: la convenzione è un'interfaccia marcatore vuota, legata al serializzatore nella configurazione, che i messaggi implementano. Gli ActorRef dentro i messaggi viaggiano come path; la serializzazione Java nativa è disabilitata.

pekko {
  actor {
    provider = cluster
    serialization-bindings { "it.unibo.pcd.CborSerializable" = jackson-cbor }
  }
  remote.artery {
    transport = tcp
    canonical.hostname = "node1"      # nome raggiungibile dagli altri nodi, non 127.0.0.1
    canonical.port = 2551
  }
  cluster {
    seed-nodes = ["pekko://ClusterSystem@seed:2551"]
    downing-provider-class = "org.apache.pekko.cluster.sbr.SplitBrainResolverProvider"
  }
}

Nel codice: public interface CborSerializable {} e record Ping(ActorRef<Pong> replyTo) implements CborSerializable {}, con le dipendenze pekko-cluster-typed e pekko-serialization-jackson.

4. Formare un cluster: seed node, membership e gossip

Un cluster Pekko è un insieme di nodi, cioè di actor system (di solito uno per processo), che si conoscono e comunicano. È peer-to-peer: nessun master né registro centrale, più difficile da gestire di uno schema master-worker ma più tollerante ai guasti. Il cluster tiene traccia dei suoi membri, del loro stato e dei loro ruoli.

Entrare e uscire

I seed node sono nodi con indirizzo noto, elencati nella configurazione e avviati per primi: i nuovi nodi li contattano per unirsi. In alternativa il join si chiede a mano all'attore manager del cluster: anche la gestione del cluster passa per messaggi.

Cluster cluster = Cluster.get(system);
cluster.manager().tell(Join.create(seedAddress));                        // unisciti
cluster.manager().tell(Leave.create(cluster.selfMember().address()));    // esci in modo ordinato
cluster.subscriptions().tell(Subscribe.create(listener, ClusterEvent.MemberEvent.class));

Con subscriptions un attore riceve gli eventi di membership, se il suo comportamento ne dipende. Un membro passa per joining → up → leaving → exiting → removed; un nodo guasto diventa unreachable e poi down. Si esce con un Leave esplicito, con un'uscita ordinata allo spegnimento del nodo, o con un crash che il cluster scopre da solo. I ruoli dei nodi (pekko.cluster.roles) servono a decidere quali attori avviare dove.

Gossip e convergenza

La membership si propaga con il gossip: periodicamente ogni nodo comunica la propria visione del cluster a un altro, come un pettegolezzo, finché l'informazione raggiunge tutti. Le visioni sono versionate con i vector clock, che permettono di confrontarle e di riconoscere quelle vecchie; il meccanismo è ispirato a Dynamo di Amazon. Il cluster è in convergenza quando tutti i membri hanno visto la stessa versione, e solo allora il leader esegue le transizioni di membership (da joining a up, da exiting a removed). Il leader non viene eletto: è il primo membro raggiungibile in un ordinamento deterministico degli indirizzi, quindi ogni nodo sa chi è, e se se ne va il ruolo passa al successivo. È un ruolo debole, non un single point of failure.

Rilevare i guasti

I nodi si scambiano heartbeat. Il phi accrual failure detector non risponde «vivo o morto» ma calcola un livello di sospetto φ, che cresce quanto più il battito atteso è in ritardo rispetto alla distribuzione degli intervalli passati. Oltre una soglia il nodo è unreachable, e torna raggiungibile se i battiti riprendono. Un nodo unreachable blocca la convergenza finché qualcuno non lo mette down.

5. Partizioni di rete: lo split brain

Se la rete si spezza, ogni parte vede l'altra come irraggiungibile: è lo split brain, e dall'interno un nodo morto e uno isolato sono indistinguibili. Se le due parti lavorassero ciascuna per conto suo, al ripristino ci sarebbero due stati divergenti, per esempio due copie dello stesso attore, e spesso non si saprebbe quale tenere. La scelta conservativa è spegnere una delle due parti.

In Pekko lo fa lo Split Brain Resolver, da abilitare nella configurazione (downing-provider-class, sezione 3); senza, nessun nodo viene rimosso da solo. Quando la situazione è stabile da un certo tempo, ogni parte applica la stessa strategia e decide da sola se sopravvivere. Ogni nodo sa quanti erano i membri prima della partizione, quindi capisce se sta nella parte più piccola, e in quel caso esce dal cluster.

Un numero dispari di nodi evita i pareggi: una partizione ha sempre una parte più numerosa dell'altra. Gli attori della parte spenta possono essere ricreati in quella sopravvissuta (con lo sharding, sezione 7), ma dallo stato iniziale: Pekko non replica lo stato degli attori. Se lo stato conta servono la persistenza (event sourcing) o strutture replicate come i CRDT di Distributed Data, fuori dal programma.

6. Trovare gli attori: receptionist e router

Il receptionist è un attore di sistema, presente su ogni nodo, che fa da registro distribuito. Un attore si registra sotto una ServiceKey, tipata con il protocollo che accetta ma senza informazioni sulla posizione; gli altri ricevono un Listing, l'insieme degli ActorRef registrati, anche remoti. Le richieste sono tre:

• register: associa il proprio ActorRef alla chiave; chi termina esce dal registro da solo;
• find: una risposta sola, con gli attori registrati in quel momento, adatta a un insieme stabile;
• subscribe: un nuovo Listing a ogni cambiamento, adatto se l'insieme cambia o se gli attori potrebbero non essere ancora partiti.

public static final ServiceKey<Worker.Command> KEY = ServiceKey.create(Worker.Command.class, "worker");

// nel setup del worker
ctx.getSystem().receptionist().tell(Receptionist.register(KEY, ctx.getSelf()));

// nel setup del coordinatore: un adapter traduce Listing in un messaggio del suo protocollo
ActorRef<Receptionist.Listing> adapter = ctx.messageAdapter(Receptionist.Listing.class, WorkersChanged::new);
ctx.getSystem().receptionist().tell(Receptionist.subscribe(Worker.KEY, adapter));

// nel coordinatore
.onMessage(WorkersChanged.class, m -> running(m.listing().getServiceInstances(Worker.KEY)))

Nell'esempio del corso un coordinatore, a ogni tick di un timer, distribuisce numeri a worker che ne calcolano il fattoriale, senza sapere quanti siano né dove; se un nodo worker si spegne il lavoro continua sugli altri. Il receptionist serve per il primo contatto, non per instradare ogni messaggio: non è pensato per scalare, e ciò che passa per lui deve essere serializzabile.

I router sono attori che inoltrano ogni messaggio ricevuto a uno di un gruppo di attori:

• il pool router (Routers.pool(4, Worker.create())) crea un numero fisso di figli nel proprio actor system e distribuisce in round robin: parallelizza su un nodo, non sul cluster;
• il group router (Routers.group(Worker.KEY)) inoltra agli attori registrati al receptionist con quella chiave, su qualunque nodo. Per default sceglie a caso, con withRoundRobinRouting() in round robin.

7. Distribuire gli attori: cluster sharding

Quando gli attori sono tanti e ciascuno ha un'identità (uno per utente, per sessione, per conto), il cluster sharding li distribuisce sui nodi senza che il programmatore scelga dove. Ogni entity è identificata da un tipo (EntityTypeKey, tipata con il protocollo) e da un entityId. Lo sharding raggruppa le entity in shard in base all'entityId, assegna gli shard ai nodi, crea ogni entity al primo messaggio e instrada i messaggi.

ClusterSharding sharding = ClusterSharding.get(system);
EntityTypeKey<Counter.Command> TYPE_KEY = EntityTypeKey.create(Counter.Command.class, "Counter");

sharding.init(Entity.of(TYPE_KEY, entityCtx -> Counter.create(entityCtx.getEntityId())));

EntityRef<Counter.Command> counter = sharding.entityRefFor(TYPE_KEY, "counter-2");
counter.tell(new Counter.Increment());       // non sappiamo, né ci interessa, su quale nodo sia

Un EntityRef si usa come un ActorRef (tell e ask), ma non è un ActorRef: non si può osservare con watch. Nell'esempio del corso, con entityId casuali tra pochi valori, i contatori finiscono su nodi diversi, compreso il seed. Se si spegne un nodo, i suoi shard vengono riassegnati ai nodi rimasti e le entity ricreate lì, ma con lo stato iniziale: senza persistenza lo stato va perso. È esattamente il caso che l'assignment 4 chiede di gestire: il sistema d'allarme su un cluster di almeno tre nodi non può dare per scontato che un attore ricreato sappia ancora se l'impianto era armato.

Verifica le tue conoscenze