Messaggi

Un indirizzo per i messaggi

I processi non condividono memoria: l’unico modo di collaborare è mandarsi messaggi. Un messaggio è un valore Gleam qualsiasi, un numero, una stringa, un record, che un processo spedisce e un altro riceve.

Per spedire serve un indirizzo, e in Gleam l’indirizzo si chiama Subject. Lo crea il processo che vuole ricevere, con process.new_subject(), e lo passa a chi deve scrivergli. Chiunque abbia il Subject può mandare messaggi con process.send; il processo che l’ha creato li legge con process.receive.

import gleam/erlang/process
import gleam/io
import gleam/string

pub fn main() -> Nil {
  let inbox = process.new_subject()
  process.spawn(fn() { process.send(inbox, "Hello, main!") })
  let message = process.receive(inbox, within: 1000)
  io.println(string.inspect(message))
  let another = process.receive(inbox, within: 100)
  io.println(string.inspect(another))
}

Ok("Hello, main!")
Error(Nil)

Passo per passo:

1. main crea un Subject, inbox, e lo dà al nuovo processo (la funzione anonima lo cattura, come ogni chiusura).
2. Il processo spedisce la stringa "Hello, main!" e termina.
3. process.receive(inbox, within: 1000) aspetta un messaggio su inbox, al massimo per 1000 millisecondi. Il messaggio arriva, e il risultato è Ok("Hello, main!").
4. Il secondo receive aspetta 100 millisecondi, ma nessuno manda più niente: il risultato è Error(Nil).

receive restituisce un Result proprio perché un messaggio potrebbe non arrivare mai: il processo che doveva scriverlo è lento, o si è fermato. Il tempo massimo di attesa, chiamato timeout, ti costringe a decidere cosa fare in quel caso.

Il tipo del messaggio

Il tipo di inbox è Subject(String): un indirizzo a cui si possono mandare solo stringhe. Il compilatore l’ha dedotto dal primo send. Se da qualche parte provi a mandare un altro tipo:

process.send(inbox, 42)

ricevi un Type mismatch, con Expected type: String e Found type: Int. È una caratteristica rara: in Erlang e in Elixir si può mandare qualsiasi cosa a qualsiasi processo, e un messaggio inatteso si scopre solo quando il programma gira. In Gleam il Subject porta con sé il tipo dei suoi messaggi, e ogni send viene controllato.

Nelle firme delle funzioni il tipo si scrive Subject(String), e si importa con import gleam/erlang/process.{type Subject}.

La mailbox

Cosa succede ai messaggi che arrivano mentre il processo destinatario sta facendo altro? Non si perdono: ogni processo ha una mailbox, una casella della posta, dove i messaggi si mettono in fila in ordine di arrivo. receive prende il primo messaggio in fila per quel Subject; se la fila è vuota, aspetta.

Una garanzia importante: se un processo A manda due messaggi a un processo B, B li riceve nell’ordine in cui A li ha spediti. Tra mittenti diversi, invece, l’ordine non è garantito, per la solita regola della lezione precedente.

Aspettare tutti, senza orologio

Nella lezione precedente main aspettava gli altri processi con un sleep “abbastanza lungo”. Con i messaggi si può fare meglio: ogni processo avvisa quando ha finito, e main aspetta esattamente il numero di avvisi che si aspetta.

import gleam/erlang/process.{type Subject}
import gleam/io
import gleam/list
import gleam/string

pub fn main() -> Nil {
  let done = process.new_subject()
  list.each([1, 2, 3, 4, 5], fn(n) {
    process.spawn(fn() {
      process.sleep(100 * { 6 - n })
      process.send(done, n)
    })
  })
  let order = wait_for(done, 5, [])
  io.println("Finished in this order: " <> string.inspect(order))
}

fn wait_for(done: Subject(Int), left: Int, arrived: List(Int)) -> List(Int) {
  case left {
    0 -> list.reverse(arrived)
    _ -> {
      let n = process.receive_forever(done)
      wait_for(done, left - 1, [n, ..arrived])
    }
  }
}

Finished in this order: [5, 4, 3, 2, 1]

wait_for è una ricorsione con accumulatore, come quelle del modulo 3: ogni giro riceve un messaggio, lo aggiunge alla lista, e scala di uno il contatore. Il programma ora dura mezzo secondo, il tempo del processo più lento, e non un istante di più.

Qui ho usato process.receive_forever, che aspetta senza limiti di tempo e restituisce direttamente il messaggio (non un Result). È comodo, ma pericoloso: se un processo non manda il suo messaggio, main resta lì per sempre, e il programma non termina più (ricordi Ctrl+C della lezione 2.6?). Nei programmi veri si preferisce receive con un timeout.

Solo il proprietario può ricevere

Un Subject appartiene al processo che l’ha creato, e solo quel processo può ricevere i suoi messaggi. Se un altro processo ci prova:

let inbox = process.new_subject()
process.spawn(fn() {
  let message = process.receive(inbox, within: 1000)
  echo message
})

il processo nuovo si schianta con il messaggio Cannot receive with a subject owned by another process (e trascina con sé anche main: vedremo perché nella lezione sui supervisori). Quindi il verso della comunicazione conta: chi vuole ricevere crea il Subject e lo consegna a chi deve scrivergli. Nella prossima lezione vedremo come fa un processo appena nato a consegnare il suo Subject a chi l’ha avviato.

Centomila processi

Proviamo a esagerare. Avviamo centomila processi, uno per ogni numero da 1 a 100 000: ognuno manda il suo numero a main, che li somma.

import gleam/erlang/process.{type Subject}
import gleam/int
import gleam/io

pub fn main() -> Nil {
  let results = process.new_subject()
  int.range(from: 1, to: 100_001, with: Nil, run: fn(_, n) {
    process.spawn(fn() { process.send(results, n) })
    Nil
  })
  io.println(int.to_string(sum_of(results, 100_000, 0)))
}

fn sum_of(results: Subject(Int), left: Int, total: Int) -> Int {
  case left {
    0 -> total
    _ -> sum_of(results, left - 1, total + process.receive_forever(results))
  }
}

5000050000

Centomila processi nati, che hanno lavorato e sono morti, in meno di un secondo, su un portatile qualunque. int.range è il fold sugli interi della lezione 3.5 (con to escluso): qui l’accumulatore non serve, e passiamo Nil. Con i thread di un sistema operativo, un esperimento del genere metterebbe in ginocchio il computer.

[1, 4, 9, 16, 25, 36, 49, 64, 81, 100]
Sum: 385

import gleam/erlang/process.{type Subject}
import gleam/int
import gleam/io
import gleam/list
import gleam/string

pub fn main() -> Nil {
  let results = process.new_subject()
  list.each([1, 2, 3, 4, 5, 6, 7, 8, 9, 10], fn(n) {
    process.spawn(fn() { process.send(results, n * n) })
  })
  let squares = collect(results, 10, []) |> list.sort(int.compare)
  io.println(string.inspect(squares))
  io.println("Sum: " <> int.to_string(int.sum(squares)))
}

fn collect(results: Subject(Int), left: Int, found: List(Int)) -> List(Int) {
  case left {
    0 -> found
    _ -> collect(results, left - 1, [process.receive_forever(results), ..found])
  }
}

Ricapitolando

• Un Subject(t) è un indirizzo a cui mandare messaggi di tipo t. Lo crea con process.new_subject() il processo che vuole ricevere.
• process.send(subject, valore) spedisce, senza aspettare.
• process.receive(subject, within: ms) aspetta al massimo ms millisecondi e restituisce Ok(messaggio) o Error(Nil); process.receive_forever aspetta senza limiti.
• Il compilatore controlla il tipo di ogni messaggio.
• I messaggi aspettano nella mailbox, in ordine di arrivo; tra due processi l’ordine è garantito.
• Solo il proprietario di un Subject può riceverne i messaggi.
• Per aspettare N processi: ognuno manda un messaggio, e una funzione ricorsiva ne riceve N.

Nella prossima lezione, un processo che non muore dopo un messaggio: resta in vita, riceve richieste, e si ricorda le cose.