Message passing e canali

1. Che cos'è il message passing

Nel modello a scambio di messaggi l'unico modo in cui i processi interagiscono è inviarsi messaggi con due primitive, send e receive. Non esistono altre astrazioni condivise: niente oggetti condivisi, monitor, lock o semafori. È il modello naturale del distribuito, ma lo si usa sempre più anche su una sola macchina, per evitare le insidie del multithreading: lo adottano i Web Worker, i framework ad attori, Go e Dart.

Nella prima parte del corso i thread interagivano attraverso componenti passivi, i monitor: chiamare un metodo del monitor significa eseguirne il codice con il proprio flusso di controllo, e magari restarci bloccati su una wait. Con i messaggi si trasferiscono solo dati, mai il controllo: ogni processo esegue solo il proprio codice, e si blocca solo sulle proprie receive (o send sincrone).

Canali e primitive

Il mezzo è il canale, un'astrazione di prima classe con un nome e un tipo di messaggio:

chan ch(type1 id1, ..., typen idn)   -- dichiarazione: nome e campi del messaggio
send ch(expr1, ..., exprn)            -- invia un messaggio con quei valori
receive ch(var1, ..., varn)           -- riceve un messaggio e lo copia nelle variabili

I tipi delle espressioni e delle variabili devono corrispondere ai campi dichiarati. L'accesso al contenuto di un canale è atomico: send e receive sullo stesso canale non si interferiscono, qualunque sia l'interleaving. I canali sono dichiarati globali ai processi, e sono l'unica cosa che i processi condividono, gestita dal supporto a run time. Un canale con buffer è, a tutti gli effetti, il bounded buffer del produttore-consumatore: chi fa send produce, chi fa receive consuma.

channel buf(int)
Producer: loop forever               Consumer: loop forever
  p1: x ← produce                      q1: receive buf(y)
  p2: send buf(x)                      q2: consume(y)

Il messaggio viaggia per copia: i processi non condividono memoria, quindi il destinatario riceve un valore, non un riferimento a un oggetto del mittente. In Java (e negli attori scritti in Java) inviare un oggetto significa invece inviare un riferimento: se mittente e destinatario poi lo modificano, la memoria condivisa rientra dalla finestra. La disciplina è usare messaggi immutabili, come i record.

Le origini: message passing asincrono con Brinch Hansen (1970, sistema operativo dell'RC 4000), porte di comunicazione con Balzer (1971), comunicazione sincrona con il CSP di Hoare (1978); gli attori di Hewitt (1973) sono asincroni. Rimasto a lungo materia di sistemi operativi, il modello è tornato con il multi-core. La cronologia completa è nel capitolo 15.

2. Sincrono o asincrono?

Nel caso sincrono il trasferimento avviene solo quando il puntatore di controllo del mittente è sulla send ch e quello del ricevente è sulla receive ch: chi arriva prima aspetta l'altro. Nel produttore-consumatore qui sopra il dato passa solo quando P è in p2 e Q è in q1. È il modello più primitivo, quello delle algebre di processi, dove si scrive ch ! msg per la send e ch ? msg per la receive. Nel caso asincrono i canali hanno un buffer FIFO; in alcuni sistemi un canale asincrono si chiama porta.

Un esempio rende la differenza osservabile. P esegue send ch(1); print("done"), Q esegue print("ready"); receive ch(v); print(v). Con la comunicazione sincrona done compare sempre dopo che Q ha eseguito la receive: quando la send si sblocca, il messaggio è stato ricevuto. Con quella asincrona done può comparire anche prima di ready, perché la send ha solo depositato il messaggio nel buffer. Attenzione: la send sincrona aspetta la ricezione, non una risposta. Se P vuole una risposta deve fare a sua volta una receive; la send che aspetta anche la risposta è il rendez-vous della sezione 7.

Il sincrono, introdotto dal CSP di Hoare (1978), è stato realizzato negli anni '80 dai Transputer programmati in Occam: processori con memoria propria collegati da canali senza buffer. Tra i due estremi stanno i canali con buffer di capacità k, dove la send si blocca solo a buffer pieno. Il sincrono dà un punto di sincronizzazione gratuito ma porta facilmente al deadlock (sezione 6); l'asincrono disaccoppia, ed è la scelta di quasi tutti i sistemi moderni.

3. Schemi di comunicazione e canali di risposta

Un canale può essere condiviso da quanti processi si vuole, da entrambi i lati. Le slide elencano tre schemi; il quarto è il duale del many-to-one.

Negli attori la topologia è fissata: ogni attore ha il proprio canale, la mailbox, su cui solo lui fa receive e tutti gli altri possono fare send. È many-to-one ovunque.

Chi riceve la risposta?

Il primo esempio delle slide: P chiede a Q di calcolare r*2 e a R di calcolare r+1, poi riceve due risposte sullo stesso canale e le somma.

chan requestA(int), requestB(int), response(int)
P: send requestA(5); send requestB(6)
   receive response(r1); receive response(r2); write(r1 + r2)
Q: receive requestA(r); send response(r * 2)
R: receive requestB(r); send response(r + 1)

Un solo canale di risposta basta, perché solo P riceve. Ma le risposte arrivano in ordine non deterministico: per una somma non importa, per una divisione sì, e servirebbe un canale di risposta per ciascuno.

Nel client-server i client sono molti, e un messaggio non dice chi l'ha inviato: il mittente va messo nel messaggio. Le slide danno a ogni client un canale di risposta in un array, reply[i], e il client mette il proprio indice nella richiesta. Un unico canale di risposta condiviso non funzionerebbe: la receive di base prende il primo messaggio disponibile, e un client potrebbe prendersi la risposta destinata a un altro. Per evitarlo servirebbe una receive con un pattern («il primo messaggio con il mio id»), che è la receive selettiva di Erlang (capitolo 15).

L'array di canali obbliga il server a conoscere in anticipo tutti i client. La soluzione elegante è trattare i canali come valori: il client crea un canale e lo mette nella richiesta («rispondimi qui»), così i client possono variare a run time. È lo schema che si usa in Go (sezione 8).

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;
  }
}

4. Il server come monitor attivo

Un server che riceve richieste e le serve una alla volta somiglia molto a un monitor: incapsula uno stato, e le operazioni su quello stato sono in mutua esclusione perché un solo processo le esegue. Per questo si chiama monitor attivo: fa il lavoro del monitor, ma è un processo con il proprio flusso di controllo. È il modo tipico di realizzare gli allocatori di risorse.

chan request(int clientID, op_kind kind, arg_type args)
chan reply[N](res_type result)

process Server {                          process Client[i = 0 to N-1] {
  while (true) {                            send request(i, opXXX, myargs);
    receive request(clientID, kind, args);  receive reply[i](myres);
    if (kind == op1) { body of op1 }      }
    else if ... else if (kind == opN) { body of opN }
    send reply[clientID](results);
  }
}

Resta la sincronizzazione condizionale. In un monitor, se la risorsa non c'è, il chiamante si sospende con una wait su una variabile condizione. Un server non può farlo, perché non c'è un flusso di controllo del client da bloccare. Il client è già bloccato da solo, sulla propria receive. Al server basta parcheggiare la richiesta in una coda e rispondere più tardi, quando potrà soddisfarla. È un idioma generale: rispondere a una richiesta non significa rispondere subito.

chan request(int clientID, op_kind kind, res_id id)
chan reply[N](res_id id)

process ResAllocator {
  int avail = MAXUNITS; set units = <valori iniziali>;
  queue pending;                                 # richieste in attesa
  while (true) {
    receive request(clientID, kind, id);
    if (kind == ACQUIRE) {
      if (avail > 0) { avail = avail - 1; id = remove(units); send reply[clientID](id); }
      else insert(pending, clientID);            # al posto della wait
    } else {                                     # RELEASE: il client restituisce id
      if (empty(pending)) { insert(units, id); avail = avail + 1; }
      else { clientID = remove(pending); send reply[clientID](id); }   # al posto della signal
    }
  }
}
Client[i]: send request(i, ACQUIRE, 0); receive reply[i](id); ... usa id ...; send request(i, RELEASE, id)

Quando arriva un rilascio e qualcuno aspetta, l'unità passa direttamente al primo in coda, senza tornare disponibile. È la stessa tecnica della versione con monitor delle slide, dove release non incrementa avail ma fa signal e passa l'unità al processo risvegliato (capitolo 7). Le slide mettono in parallelo le due versioni:

Il modello si estende senza cambiamenti al distribuito: se i canali sono connessioni di rete, client e server possono stare su macchine diverse, cosa impossibile con un monitor.

5. Ricevere da più canali: la comunicazione con guardia

Un server che ascolta su due canali ha un problema: la receive su uno si blocca, e intanto un messaggio può arrivare sull'altro. È il problema che risolvono la select sulle socket o Java NIO. La soluzione concettuale è la comunicazione con guardia (guarded communication) di Dijkstra (1974), valida nel caso sincrono e in quello asincrono.

B ; C → S

• B è un'espressione booleana sullo stato locale del processo; se manca vale true.
• C è un'istruzione di comunicazione, di solito una receive (ma può essere una send).
• S è il blocco di istruzioni da eseguire.

B e C insieme formano la guardia. La guardia ha successo se B è vera ed eseguire C non causerebbe attesa (c'è un messaggio da ricevere). Fallisce se B è falsa. Si blocca se B è vera ma C non può ancora essere eseguita.

Da sola una guardia serve a poco; serve quando ce ne sono più d'una in alternativa. Nell'if la semantica è in tre tempi:

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

1. Si valutano le guardie. Se falliscono tutte, l'if termina senza effetto. Se almeno una ha successo, se ne sceglie una in modo non deterministico. Se nessuna ha successo e alcune sono bloccate, si aspetta che una abbia successo.
2. Si esegue la comunicazione C della guardia scelta.
3. Si esegue il blocco S corrispondente.

Quindi, se più guardie sono soddisfatte, non c'è un ordine: ne viene eseguita una sola, scelta a caso, e i messaggi sugli altri canali restano dove sono per la prossima volta. Il costrutto do … od ripete la selezione finché tutte le guardie falliscono, ed è la forma tipica di un server. Nell'esempio delle slide il server accetta richieste di somma e di prodotto su due canali diversi, finché non ne ha servite max:

if nReqs < max; receive computeSum(a, b, i) → nReqs += 1; send result[i](a + b)
[] nReqs < max; receive computeMul(a, b, i) → nReqs += 1; send result[i](a * b)
fi

La parte booleana dice quando un ramo è abilitato. Nel processo Copy, un buffer circolare di 10 caratteri tra due canali, la seconda guardia contiene una send: con un canale sincrono anche la send può bloccarsi, quindi può stare in una guardia.

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
}

Quando il buffer non è né vuoto né pieno entrambe le guardie possono avere successo, e la scelta non deterministica lascia procedere l'una o l'altra.

Con lo stesso schema il bounded buffer diventa un processo gestore, cioè un monitor attivo con due canali di richiesta. La guardia nItems < maxElems disabilita le put a buffer pieno, nItems > 0 le get a buffer vuoto. Il consumatore che chiede un elemento a buffer vuoto resta bloccato sulla propria receive finché il gestore non gli risponde: nessun processo blocca mai il flusso di controllo di un altro.

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

6. Peer-to-peer: lo scambio dei valori

Nell'interazione peer-to-peer le responsabilità non sono concentrate in un server ma distribuite tra processi pari (peer), che si coordinano con un protocollo di messaggi. L'esempio del corso è l'exchanging values problem: n processi hanno ciascuno un intero v, e ognuno deve scoprire il minimo e il massimo di tutti i valori. Le tre soluzioni delle slide usano tre topologie diverse, con proprietà diverse.

chan values(int), results[n](int smallest, int largest)
process P[0] {                                   # coordinatore
  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 = 1 to n-1] {
  int v = ..., smallest, largest;
  send values(v); receive results[i](smallest, largest);
}

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);                           # il mio valore a tutti
  for k in [1 .. n-1] {                          # gli n-1 valori degli altri
    receive values[i](new);
    if (new < smallest) smallest = new;
    if (new > largest) largest = new;
  }
}

chan values[n](int smallest, int largest)
process P[0] {
  int v = ...; int smallest = v, largest = v;
  send values[1](smallest, largest);             # 1° giro: raccolta
  receive values[0](smallest, largest);          # ora li conosce solo P0
  send values[1](smallest, largest);             # 2° giro: diffusione
}
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);
}

Il confronto è un compromesso tra numero di messaggi, distribuzione del lavoro e grado di concorrenza. La centralizzata e l'anello usano pochi messaggi; la simmetrica paga in messaggi l'assenza di qualunque punto centrale.

Laboratorio: le stesse soluzioni con canali sincroni

Lo pseudocodice delle slide assume canali asincroni. Qui scegli soluzione, N e capacità dei canali (0 = sincroni, come in Go senza buffer), poi esegui a passi o verifica tutti gli interleaving. Da provare: la simmetrica con canali sincroni, dove tutti fanno send e nessuno riceve; la capacità minima che la sblocca; l'anello sincrono, dove l'ultima send verso P0 non trova nessuno che riceva, finché non spunti «P0 riceve anche l'ultimo messaggio».

La lezione generale: una strategia corretta con canali asincroni può andare in deadlock con canali sincroni. Nella simmetrica ogni canale deve poter contenere n − 1 messaggi, uno per ogni altro processo, altrimenti serve una strategia diversa: un buffer di 1 non basta.

chan getFork[N](), getForkReply[N](), releaseFork[N](), releaseForkReply[N]()
process Waiter[i = 0 to N-1] {
  loop { receive getFork[i](); send getForkReply[i]();
         receive releaseFork[i](); send releaseForkReply[i](); }
}
process Philosopher[i = 0 to N-1] {
  int first = i, second = (i+1) % N;
  if (second < first) { first = second; second = i; }   # prima l'indice minore
  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]();
  }
}

7. Rendez-vous: call e accept

Nelle slide il rendez-vous è una sincronizzazione estesa: il mittente aspetta non solo che il destinatario riceva la richiesta, ma anche che gli restituisca il risultato. Il nome evoca due persone che si danno appuntamento: la prima che arriva aspetta la seconda. I punti di sincronizzazione sono due invece di uno, sempre senza trasferimento di controllo.

• Il processo chiamante esegue una call su una entry del processo accettante. Deve conoscere l'identità dell'accettante e il nome della entry.
• Il processo accettante esegue accept sulle proprie entry, calcola e restituisce il risultato. Non ha bisogno di conoscere l'identità del chiamante.

Client                                      Server
loop forever                                loop forever
  p1: param ← ...                             q1: accept service(p, r)
  p2: call server.service(param, result)      q2:   <esegue il servizio con p>
  p3: use(result)                             qN:   r ← <risultato>

La call si blocca finché il server non esegue l'accept, e resta bloccata finché il corpo dell'accept non è finito e il risultato non è tornato.

Ada, il linguaggio nato negli anni '70–'80 per i sistemi della difesa statunitense, ha il rendez-vous come meccanismo base di comunicazione tra task. Le entry si dichiarano nell'interfaccia pubblica del task e si implementano nel corpo con l'istruzione accept; il chiamante scrive T.E(parametri).

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          -- il chiamante aspetta fino a end
            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;              -- entry call: rendez-vous
      delay 0.1;
      Put_Line("Eat the resulting hot dog");
      New_Line;
   end loop;
   Put_Line("I am not hungry any longer");
end HotDog;

A ogni chiamata il programma principale resta fermo per gli 0,8 secondi del corpo dell'accept, quindi ogni «Put hot dog in bun and add mustard» è seguito da «Eat the resulting hot dog». Solo dopo il quarto accept il Gourmet stampa «I am out of hot dogs» prima dell'ultimo «Eat», perché il chiamante è ancora nel suo delay 0.1.

Oggi il rendez-vous come primitiva è raro. L'idea sopravvive nelle chiamate di procedura remota, e in forma asincrona nell'ask degli attori (capitolo 15), che invia una richiesta e restituisce una future per la risposta.

8. Go: goroutine, canali e select

Go è un linguaggio a tipizzazione statica nato in Google nel 2007 (Griesemer, Pike e Thompson): sintassi derivata dal C, compilazione in codice nativo, garbage collection. Per il corso interessa perché ha nel linguaggio tre primitive di concorrenza di origine CSP: processi leggeri (goroutine), canali e l'istruzione select. Il motto è «Do not communicate by sharing memory; instead, share memory by communicating». Go offre comunque anche mutex e gli altri meccanismi classici nel package sync.

Per leggere gli esempi basta poco. Il tipo segue il nome (x int), := dichiara e inferisce il tipo, le funzioni si dichiarano con func nome(parametri) tipoRisultato. L'unico ciclo è for, anche nella forma for i, x := range s, e []T è una slice, cioè un array dinamico che cresce con append. Non ci sono classi: le struct ricevono metodi dall'esterno e le interfacce sono soddisfatte implicitamente da chi ha i metodi giusti.

La select è la comunicazione con guardia della sezione 5, senza la parte booleana: ogni case è una send o una receive. Si blocca finché uno dei casi non può procedere; se ne sono pronti più d'uno, ne sceglie uno a caso, con probabilità uniforme. Un ramo default la rende non bloccante.

Esempio: client-server con canale di risposta

Il server ascolta su due canali, uno per le somme e uno per i prodotti. Ogni client crea un canale privato e lo mette nella richiesta, come nella sezione 3, così il server non deve conoscere i client in anticipo.

type Request struct {
	X, Y  int
	Reply chan int // dove rispondere
}

func server(add, mul <-chan Request) {
	for {
		select { // attesa su due canali: il primo pronto vince
		case r := <-add:
			r.Reply <- r.X + r.Y
		case r := <-mul:
			r.Reply <- r.X * r.Y
		}
	}
}

func client(name string, ch chan<- Request, done chan<- string) {
	myChan := make(chan int) // canale privato per la risposta
	ch <- Request{2, 3, myChan}
	done <- fmt.Sprintf("%s: %d", name, <-myChan)
}

func main() {
	add, mul, done := make(chan Request), make(chan Request), make(chan string)
	go server(add, mul)
	go client("somma", add, done)
	go client("prodotto", mul, done)
	fmt.Println(<-done)
	fmt.Println(<-done) // l'ordine delle due righe non è deterministico
}

Tutti i canali sono senza buffer: ogni send si completa solo quando dall'altra parte c'è una receive. Il main aspetta i due messaggi su done prima di terminare, e non è un dettaglio: in Go, quando il main termina, il programma finisce e tutte le goroutine vengono interrotte.

func copyProc(in <-chan rune, out chan<- rune) {
	var buf [10]rune
	front, rear, count := 0, 0, 0
	for {
		var inC <-chan rune // nil: ramo disabilitato
		var outC chan<- rune
		if count < 10 { inC = in }  // guardia del primo ramo
		if count > 0 { outC = out } // guardia del secondo ramo
		select {
		case c := <-inC:
			buf[rear] = c
			rear, count = (rear+1)%10, count+1
		case outC <- buf[front]:
			front, count = (front+1)%10, count-1
		}
	}
}

Tre trappole viste in laboratorio

• Canali nil negli array. make([]chan int, n) crea una slice di n canali nil, non n canali. Una send o una receive su un canale nil si blocca per sempre senza eccezioni: ogni elemento va creato con make(chan int).
• La fine del main. Non aspetta le goroutine. Invece di un ciclo infinito in fondo al main si usa un canale done su cui ognuno segnala la fine, oppure un sync.WaitGroup.
• Deadlock da canali sincroni. Un algoritmo pensato per canali asincroni, come la soluzione simmetrica dello scambio dei valori, con canali senza buffer si blocca. Se tutte le goroutine sono bloccate il runtime termina con fatal error: all goroutines are asleep - deadlock!; se qualcuna è ancora viva, per esempio un server in ciclo infinito, il programma resta semplicemente fermo.

func spawnAgent(id int) <-chan string {
	ch := make(chan string)
	go func() { ch <- fmt.Sprint("ciao da ", id) }()
	return ch
}

func main() {
	const n = 1_000_000
	start := time.Now()
	agents := make([]<-chan string, 0, n)
	for i := 0; i < n; i++ {
		agents = append(agents, spawnAgent(i))
	}
	for _, ch := range agents {
		<-ch
	}
	fmt.Println(n, "goroutine in", time.Since(start))
}

func peer(i, v int, values []chan int, wg *sync.WaitGroup) {
	defer wg.Done()
	for j, ch := range values {
		if j != i {
			ch <- v // prima invio il mio valore a tutti...
		}
	}
	smallest, largest := v, v
	for k := 1; k < len(values); k++ { // ...poi ricevo gli altri n-1
		x := <-values[i]
		smallest, largest = min(smallest, x), max(largest, x)
	}
	fmt.Printf("P%d: min %d, max %d\n", i, smallest, largest)
}

func main() {
	vs := []int{7, 3, 9, 1, 5}
	n := len(vs)
	values := make([]chan int, n) // n canali nil: vanno creati uno per uno
	for i := range values {
		values[i] = make(chan int, n-1) // con make(chan int) è deadlock
	}
	var wg sync.WaitGroup
	for i, v := range vs {
		wg.Add(1)
		go peer(i, v, values, &wg)
	}
	wg.Wait()
}

L'assignment 3 comprende anche un piccolo esercizio in Go con i canali: le domande tipiche sono nella pagina di preparazione. Il prossimo capitolo passa al modello opposto, gli attori, dove tutto è asincrono e il canale diventa la mailbox di ciascuno.

Verifica le tue conoscenze