Message Passing Models & Attori

1. Fondamenti del Message Passing

Finora abbiamo studiato modelli di programmazione concorrente basati su memoria condivisa: semafori, monitor, variabili condition, lock. In tutti questi modelli i thread/processi interagiscono leggendo e scrivendo variabili in uno spazio di indirizzamento comune, protetto da meccanismi di mutua esclusione e sincronizzazione.

Il message passing capovolge completamente questo paradigma: l'unico modo per due processi di interagire e scambiarsi messaggi esplicitamente attraverso primitive di send e receive. Non esistono oggetti condivisi, monitor, lock o semafori. Questa filosofia, come sottolinea il Prof. Ricci, e alla base della programmazione distribuita ma sta guadagnando terreno anche nella programmazione concorrente tradizionale, perche elimina alla radice intere categorie di bug (data race, deadlock da lock dimenticati, etc.).

Il messaggio non e solo un dato: e l'unita atomica di interazione. Un processo invia un messaggio su un canale; un altro processo lo riceve dallo stesso canale. Non c'e accoppiamento forte come nella chiamata di procedura — i due processi possono essere su core diversi, macchine diverse, continenti diversi.

2. Primitive di comunicazione

Il modello introduce un nuovo tipo di dato astratto: il canale. Un canale e dichiarato globalmente ai processi e specifica la struttura dei messaggi che puo trasportare.

Dichiarazione di un canale

chan ch(type id1, ..., type idn)

Ad esempio: chan request(int value) definisce un canale chiamato request che trasporta messaggi contenenti un singolo intero.

Primitive

• send ch(expr1, expr2, ...) — invia un messaggio composto dalle espressioni sul canale ch. I tipi delle espressioni devono coincidere con quelli dichiarati.
• receive ch(var1, var2, ...) — riceve un messaggio dal canale ch depositando i valori nelle variabili specificate.

Atomicita

L'accesso al contenuto del canale e atomico: quando un processo esegue una send (o una receive), quell'operazione non puo essere interrotta da un altro processo sul medesimo canale. Questo garantisce che non ci siano race condition a livello del canale stesso.

3. Comunicazione sincrona vs asincrona

La distinzione fondamentale nel message passing e tra comunicazione sincrona e asincrona. Il Prof. Ricci dedica molta attenzione a questo punto, perche determina le proprieta di sincronizzazione, l'accoppiamento temporale, e la complessita dei sistemi.

Step simulator: confronto sincrono/asincrono

Il simulatore seguente mostra la differenza tra una send sincrona e una asincrona in un semplice schema produttore-consumatore. Clicca "Avanza" per eseguire un passo per volta.

Sintassi alternativa (notazione algebrica)

send ch(msg):   ch ! msg
receive ch(msg): ch ? msg

Questa notazione e usata frequentemente nella letteratura sugli algebre di processi (CSP, CCS) e in linguaggi come Go per i canali.

4. Schemi di comunicazione

Il Prof. Ricci distingue tre schemi fondamentali di comunicazione basati sul numero di mittenti e destinatari per un dato canale:

Esempio: produttore-consumatore semplice

chan buf(int);
process Producer {
  integer x
  loop forever:
    p1: x <- produce
    p2: send buf(x)
}
process Consumer {
  integer y
  loop forever:
    q1: receive buf(y)
    q2: consume(y)
}

Pipeline: CharToLine

chan input(char), output(char[MAXLINE]);
process CharToLine {
  char line[MAXLINE+1];
  int i = 0;
  while (true) {
    receive input(line[i]);
    while (line[i] != CR and i < MAXLINE) {
      i = i + 1;
      receive input(line[i]);
    }
    line[i] = EOL;
    send output(line);
    i = 0;
  }
}

5. Client-Server e Monitor Attivi

chan request(int, kind, arg_type);
chan[NCLIENTS] reply(arg_result);
process Client[i = 0 to N-1] {
  arg_type myargs;
  res_type myres;
  <init args>
  send request(i, opXXX, myargs);
  receive reply[i](myres);
}

Nota importante: ogni client ha un canale di reply dedicato (reply[i]). Il server usa l'ID del client (passato come primo campo della richiesta) per sapere su quale canale inviare la risposta.

6. Allocatori di risorse: passivi vs attivi

Il Prof. Ricci dedica una parte importante della lezione al confronto tra allocatori di risorse con monitor passivi (memoria condivisa) e processi attivi (message passing). La tabella seguente riassume la corrispondenza:

L'idea centrale: le condition variable nei monitor servono a bloccare un thread finche una condizione non e soddisfatta. Con i messaggi, lo stesso effetto si ottiene parcheggiando la richiesta in una coda interna e rimandando la risposta a quando la risorsa sara disponibile.

State Explorer: allocatore attivo

Esplora i possibili stati del processo allocatore attivo mentre gestisce richieste di ACQUIRE e RELEASE:

Codice dell'allocatore attivo

chan request(int clientID, int type);
chan[N] reply(res_id id);
process ResAllocator {
  int clientID;
  int avail = MAXUNITS;
  queue pending;
  set units = <valore iniziale>;
  op_kind kind; res_id id;
  while (true) {
    receive request(clientID, kind);
    if (kind == ACQUIRE) {
      if (avail > 0) {
        avail = avail - 1;
        id = remove(units);
        send reply[clientID](id);
      } else { insert(pending, clientID); }
    } else if (kind == RELEASE) {
      insert(units, id);
      if (empty(pending)) { avail = avail + 1; }
      else { remove(pending, clientID); send reply[clientID](id); }
    }
  }
}

7. Comunicazione guardata

Un problema fondamentale: come ricevere messaggi che possono arrivare su multipli canali contemporaneamente? La soluzione classica, introdotta da Dijkstra nel 1974, e la comunicazione guardata (guarded communication).

Statement di comunicazione guardata

// Forma generale
B ; C → S

• B e una guardia booleana (se omessa, vale true).
• C e uno statement di comunicazione (tipicamente una receive).
• S e il blocco di statement da eseguire.

La guardia ha successo se B e vera e C puo essere eseguito senza bloccarsi. Fallisce se B e falsa. Si blocca se B e vera ma C non puo ancora essere eseguito.

If con guarded communication

if B1; C1 → S1;
[] B2; C2 → S2;
[] B3; C3 → S3;
fi

Semantica:

1. Valuta tutte le espressioni booleane nelle guardie.
2. Se tutte falliscono, l'if termina senza effetto.
3. Se almeno una ha successo, ne sceglie una non-deterministicamente.
4. Se tutte si bloccano, aspetta finche una guardia ha successo.
5. Esegue C per la guardia scelta, poi S.

Loop con guarded communication

do B1; C1 → S1;
[] B2; C2 → S2;
[] B3; C3 → S3;
od

Esempio — processo Copy che trasferisce caratteri usando un buffer circolare:

process Copy(chan in(char), chan out(char)) {
  char buffer[10];
  int front = 0, rear = 0, count = 0;
  do count < 10; receive in(buffer[rear])
     → count++; rear = (rear + 1) % 10;
  [] count > 0; send out(buffer[front])
     → count--; front = (front + 1) % 10;
  od
}

8. Bounded Buffer Manager

process BoundedBufferManager {
  int nItems = 0;
  int maxElems = ...;
  Queue<ItemType> queue = ...;
  ItemType item;
  boolean ack = true;
  chan replyChan;
  do nItems < maxElems; receive put(item, replyChan)
     → queue.add(item); nItems++; send replyChan(ack);
  [] nItems > 0; receive get(replyChan)
     → ItemType el = queue.remove(); nItems--; send replyChan(el);
  od
}
process Producer(chan myChan) {
  boolean ack;
  loop { ItemType el = produce(); send put(el, myChan); receive myChan(ack); }
}
process Consumer(chan myChan) {
  loop { ItemType el; send get(myChan); receive myChan(el); consume(el); }
}

9. Exchanging Values Problem

Il Prof. Ricci introduce un problema classico per illustrare come topologie diverse portino a soluzioni con proprieta diverse: abbiamo N processi, ognuno con un valore locale v. Vogliamo che tutti conoscano il valore minimo e massimo tra tutti i valori. Tre soluzioni, tre topologie.

flowchart TB
    subgraph Centralizzato
        C0((P0)) ---|coordinator| C1((P1)) & C2((P2)) & C3((P3))
    end
    subgraph Simmetrico
        S0((P0)) --- S1((P1))
        S0 --- S2((P2))
        S0 --- S3((P3))
        S1 --- S2
        S1 --- S3
        S2 --- S3
    end
    subgraph Anello
        direction LR
        R0((P0)) --> R1((P1)) --> R2((P2)) --> R3((P3)) --> R0
    end
    

Soluzione centralizzata

chan values(int), results[n](int smallest, int largest);
process P[0] {
  int v = ....;
  int new, smallest = v, largest = v;
  for i in [1...n-1] {
    receive values(new);
    if (new < smallest) smallest = new;
    if (new > largest)  largest = new;
  }
  for i in [1...n-1] { send results[i](smallest, largest); }
}
process P[i] {
  int v = ..., smallest, largest;
  send values(v);
  receive results[i](smallest, largest);
}

• Vantaggio: pochi messaggi (2(n-1)).
• Svantaggio: bottleneck sul coordinatore; single point of failure.

Soluzione simmetrica

chan values[n](int);
process P[i = 0 to n-1] {
  int v = ....;
  int new, smallest = v, largest = v;
  for j in [0...n-1], j != i { send values[j](v); }
  for k in [1...n-1] {
    receive values[i](new);
    if (new < smallest) smallest = new;
    if (new > largest)  largest = new;
  }
}

• Vantaggio: massima distribuzione e parallelismo.
• Svantaggio: numero di messaggi N*(N-1) — quadratico.

Soluzione ad anello

process P[0] {
  int v = ....;
  int new, smallest = v, largest = v;
  send values[1](smallest, largest);
  receive values[0](smallest, largest);
  send values[1](smallest, largest);
}
process P[i = 1 to n-1] {
  int v = ..., smallest, largest;
  receive values[i](smallest, largest);
  if (v < smallest) smallest = v;
  if (v > largest)  largest = v;
  send values[(i+1)%n](smallest, largest);
  receive values[i](smallest, largest);
  send values[(i+1)%n](smallest, largest);
}

• Vantaggio: pochi messaggi (2n).
• Svantaggio: parallelismo limitato (serializzazione nell'anello).

10. Filosofi a cena distribuiti

Il problema dei filosofi a cena puo essere ripensato in termini di scambio di messaggi. Il Prof. Ricci osserva: "I canali diventano le forchette."

Soluzione centralizzata (con cameriere)

flowchart LR
    W[Waiter] --- Ph0[Ph 0] & Ph1[Ph 1] & Ph2[Ph 2] & Ph3[Ph 3] & Ph4[Ph 4]
    

chan getForks(int, int, chan);
chan releaseForks(int, int, chan);
process Waiter[i:0..N-1] {
  List<Request> pending = ...;
  boolean availForks[0..N-1] = {false, ...};
  do receive getForks(fork1, fork2, ReplyChanID)
     → if availForks[fork1] && availForks[fork2] {
         availForks[fork1] = false; availForks[fork2] = false;
         send ReplyChanID(fork1, fork2);
       } else { pending.add(new Request(Reply, fork1, fork2)); }
  [] receive releaseForks(fork1, fork2)
     → availForks[fork1] = true; availForks[fork2] = true;
       for each Request r in pending {
         if (availForks[r.fork1] && availForks[r.fork2]) {
           pending.remove(r);
           availForks[r.fork1] = false; availForks[r.fork2] = false;
           send req.ReplyChanID(req.fork1, req.fork2);
         }
       }
  od
}

Soluzione distribuita (un cameriere per forchetta)

flowchart LR
    W0[W 0] --- Ph0[Ph 0] & Ph1[Ph 1]
    W1[W 1] --- Ph1 & Ph2[Ph 2]
    W2[W 2] --- Ph2 & Ph3[Ph 3]
    W3[W 3] --- Ph3 & Ph4[Ph 4]
    W4[W 4] --- Ph4 & Ph0
    

process Waiter[i:0..N-1] {
  loop {
    receive getFork[i]();
    send getForkReply[i]();
    receive releaseFork[i]();
    send releaseForkReply[i]();
  }
}
process Philosopher[i:0..N-1] {
  int first = i; int second = (i+1) % N;
  if (second < first) { first = second; second = i; }
  loop {
    think();
    send getFork[first](); receive getForkReply[first]();
    send getFork[second](); receive getForkReply[second]();
    eat();
    send releaseFork[first](); receive releaseForkReply[first]();
    send releaseFork[second](); receive releaseForkReply[second]();
  }
}

11. Rendez-vous e Ada

Il rendez-vous e un'estensione della comunicazione sincrona in cui il mittente non solo aspetta che il ricevente accetti il messaggio, ma attende anche la risposta.

-- Call e Accept in Ada
task T;
entry E(formals);
...
T.E(actuals);  -- chiama l'entry

accept E(formals) do
  -- corpo del rendez-vous
end accept;

with Ada.Text_IO; use Ada.Text_IO;
procedure HotDog is
  task Gourmet is
    entry Make_A_Hot_Dog;
  end Gourmet;
  task body Gourmet is
  begin
    Put_Line("I am ready to make a hot dog for you");
    for Index in 1..4 loop
      accept Make_A_Hot_Dog do
        delay 0.8;
        Put("Put hot dog in bun "); Put_Line("and add mustard");
      end Make_A_Hot_Dog;
    end loop;
    Put_Line("I am out of hot dogs");
  end Gourmet;
begin
  for Index in 1..4 loop
    Gourmet.Make_A_Hot_Dog;
    delay 0.1;
    Put_Line("Eat the resulting hot dog"); New_Line;
  end loop;
  Put_Line("I am not hungry any longer");
end HotDog;

12. Introduzione agli Attori

Con la seconda parte della lezione, il Prof. Ricci introduce il modello Actor, sviluppato originariamente da Carl Hewitt al MIT negli anni Settanta nel contesto dell'intelligenza artificiale, e successivamente approfondito da Gul Agha come unificazione tra OOP e concorrenza.

Un attore e un'entita computazionale che incapsula stato, comportamento e un flusso di controllo logico. A differenza degli oggetti classici, gli attori sono autonomi e reattivi: lavorano solo quando ricevono un messaggio.

Le tre primitive fondamentali

1. send — invia un messaggio asincrono a un attore (identificato dal suo ID).
2. create — crea un nuovo attore con un comportamento specificato.
3. become — specifica un nuovo comportamento che l'attore usera per il messaggio successivo (history-sensitive).

flowchart LR
    subgraph A1 [Attore A]
        S1[(Stato)] & C1[Comportamento] & ID1[ID]
    end
    subgraph A2 [Attore B]
        S2[(Stato)] & C2[Comportamento] & ID2[ID]
    end
    A1 -->|messaggio asincrono| A2
    A2 -->|messaggio asincrono| A1
    

13. Semantiche chiave degli attori

State Explorer: ciclo di vita di un attore

14. Event Loop e macro-step

L'implementazione pratica del modello attore si basa su un event loop implicito o esplicito.

Event loop implicito

loop {
  msg <- waitForMsg()
  handler <- selectHandler(msg)
  execute(handler)
}

Esempi: ActorFoundry, Akka, Vert.x, Web Workers.

Event loop esplicito

Il programmatore scrive il loop di ricezione e usa pattern matching. Esempio: Erlang.

15. Attori in pratica: ActorFoundry e Akka

ActorFoundry (Java)

public class PingActor extends Actor {
  ActorName otherPinger;
  @message
  public void start(ActorName other) {
    otherPinger = other;
    send(otherPinger, "ping", self(), Id.stamp()+"called from "+self());
  }
  @message
  public void ping(ActorName caller, String msg) {
    send(stdout, "println", Id.stamp()+"Received ping ("+msg+") from "+caller+"...");
    send(caller, "alive", Id.stamp()+self().toString()+" is alive");
  }
  @message
  public void alive(String reply) {
    send(stdout, "println", Id.stamp()+"Received "+reply+" from pinged actor");
  }
}

Akka (Java — Untyped Actor)

import akka.actor.UntypedActor;
import akka.event.Logging;
import akka.event.LoggingAdapter;
public class MyUntypedActor extends UntypedActor {
  LoggingAdapter log = Logging.getLogger(getContext().system(), this);
  public void onReceive(Object message) throws Exception {
    if (message instanceof String) {
      log.info("Received String message: {}", message);
      getSender().tell("received");
    } else unhandled(message);
  }
}

Akka (Scala — Typed Actor)

import akka.actor.Actor
import akka.actor.Props
import akka.event.Logging
class MyActor extends Actor {
  val log = Logging(context.system, this)
  def receive = {
    case "test" => log.info("received test")
    case _      => log.info("received unknown message")
  }
}

16. Spaghetti asincrono e soluzioni

Il "asynchronous spaghetti": la logica applicativa viene frammentata in un insieme non strutturato di handler/callback, ciascuno dei quali reagisce a un evento specifico. Quando un flusso di lavoro richiede una sequenza di passi, ogni passo diventa un handler separato.

Esempio: sin(x) * cos(y)

Supponiamo di voler calcolare sin(x) * cos(y) usando attori separati per sin e cos. Con l'approccio a handler: invia x a Sin, invia y a Cos, attendi entrambe le risposte, poi moltiplica. Ogni risposta richiede un handler diverso, e bisogna tracciare lo stato per sapere se l'altra risposta e gia arrivata.

Soluzioni proposte

• Futures/Promises: oggetti proxy per un risultato non ancora disponibile.
• Promise pipelines: compongono future in sequenza.
• Stashing (Akka): mettere da parte i messaggi che non possono essere processati ora.
• Local Synchronization Constraints (ActorFoundry): condizioni booleane che abilitano/disabilitano la ricezione di messaggi.

Stashing in Akka

import akka.actor.Stash
class ActorWithProtocol extends Actor with Stash {
  def receive = {
    case "open" => unstashAll(); context.become({
      case "write" => // do writing...
      case "close" => unstashAll(); context.unbecome()
      case msg => stash()
    }, discardOld = false)
    case msg => stash()
  }
}

LSC in ActorFoundry

@Disable(messageName = "put")
public Boolean disablePut(Integer x) {
  if (bufferReady) { return (tail == bufferSize); }
  else return true;
}

17. Attori in Erlang

Erlang e il linguaggio che ha portato il modello attore nella produzione industriale. Sviluppato da Ericsson dal 1987 per telecomunicazioni.

% Creazione processo
Pid = spawn(math, fact, [999]).
% Invio messaggio
Pid ! Message.
% Ricezione con pattern matching
receive
  Pattern1 [when Guard1] -> Expression1;
  Pattern2 [when Guard2] -> Expression2;
  ...
end

Counter in Erlang

-module(counter).
-export([start/0]).
start() -> loop(0).
loop(Sum) ->
  receive
    {inc} -> loop(Sum + 1);
    {getValue, Pid} -> Pid ! {count_value, Sum}, loop(Sum)
  end.

Account bancario

account(Balance) ->
  receive
    {deposit, Amount, Whom} -> Whom ! {deposit_receipt, Amount},
                               account(Balance + Amount);
    {balance, Whom} -> Whom ! {balance, Balance},
                       account(Balance);
    {withdrawal, Amount, Whom} when Amount > Balance ->
                               Whom ! overdraft, account(Balance);
    {withdrawal, Amount, Whom} -> Whom ! {withdrawal_receipt, Amount},
                               account(Balance - Amount)
  end.

18. Pattern e idiomi negli attori

Pro-activity Problem

Gli attori sono entita puramente reattive. Come modellare un comportamento pro-attivo? Soluzioni: splitting in sotto-attori, self-sending di messaggi.

Futures

Un future e un attore proxy per un risultato non ancora disponibile. Tre stati: iniziale (nessun valore, nessun cliente), attesa (clienti accodati), risolto (clienti notificati, valore immutabile). Il future puo essere passato ad altri attori prima di essere risolto.

Actor Idioms (Schumacher, AGERE! 2012)

• Serializer: attore che serializza l'accesso a una risorsa.
• Fork-Join: attore che distribuisce lavoro e raccoglie risultati.
• Scheduler: attore che gestisce il passaggio del testimone.

Messaging Patterns (V. Vernon)

• Messaging Channels: Point-to-Point, Publish-Subscribe, Dead Letter, Message Bus.
• Message Construction: Command, Document, Event, Request-Reply, Correlation Identifier.
• Message Routing: Content-based Filter, Message Filter, Process Manager, Message Broker.
• Message Transformation: Envelope Wrapper, Content Enricher, Content Filter.

Timeline storica

1. 1970 — Brinch-Hansen introduce l'async message passing per l'RC4000.
2. 1971 — Bob Balzer introduce il concetto di communication port.
3. 1973 — Carl Hewitt introduce il modello Actor all'MIT.
4. 1974 — Dijkstra introduce le guarded commands.
5. 1978 — Hoare formalizza CSP (Communicating Sequential Processes).
6. 1986 — Gul Agha pubblica "Actors: A Model of Concurrent Computation".
7. 1987 — Ericsson sviluppa Erlang per telecomunicazioni.
8. 2007 — Go language (Google): goroutine, canali, select.
9. Oggi — Akka, Dart isolates, Web Workers, attori mainstream.

Verifica le tue conoscenze