Un processo che ricorda

Lo stato senza variabili

In Gleam niente cambia: ogni valore è immutabile, e fino a qui lo “stato” di un programma è sempre passato da una funzione all’altra, come l’accumulatore di un fold. Ma allora come si fa ad avere un contatore che tanti pezzi del programma possono incrementare e leggere, in momenti diversi?

La risposta della BEAM è: con un processo. Un processo che non termina dopo il primo messaggio, ma resta in vita, riceve richieste una dopo l’altra, e tiene lo stato come argomento di una funzione ricorsiva. Ogni messaggio produce un nuovo stato, e la funzione richiama sé stessa con quello.

I messaggi del contatore

Prima di tutto, cosa si può chiedere al contatore? Due cose: incrementarsi, e dire quanto vale. Lo scriviamo con un tipo, come nel modulo 4:

pub type Message {
  Increment
  Get(reply_to: Subject(Int))
}

Increment non porta dati. Get invece porta un Subject(Int): l’indirizzo a cui il contatore deve mandare la risposta. Un processo non ha un modo di “restituire” qualcosa a chi gli scrive, perché i messaggi vanno in una sola direzione; se vuoi una risposta, devi spedire insieme alla domanda l’indirizzo dove riceverla. Come una lettera con la busta preaffrancata dentro.

Il ciclo

Il cuore del processo è una funzione che riceve un messaggio, decide il nuovo stato, e ricomincia:

fn loop(inbox: Subject(Message), count: Int) -> Nil {
  case process.receive_forever(inbox) {
    Increment -> loop(inbox, count + 1)
    Get(reply_to) -> {
      process.send(reply_to, count)
      loop(inbox, count)
    }
  }
}

• Con Increment, il ciclo ricomincia con count + 1: quello è il nuovo stato.
• Con Get, manda il valore attuale a reply_to e ricomincia con lo stesso stato.

Nessuna variabile viene modificata: count è un parametro, e ogni giro ne riceve uno nuovo. Ed è una ricorsione in coda (la chiamata a loop è l’ultima cosa in ogni ramo), quindi, come nella lezione 2.7, il processo può girare per anni senza occupare più memoria. Mentre aspetta in receive_forever non consuma nemmeno processore: gli scheduler lo risvegliano solo quando arriva un messaggio.

La stretta di mano

Resta un problema, quello lasciato aperto nella lezione precedente: solo il proprietario di un Subject può ricevere, quindi l’inbox del contatore deve crearla il contatore stesso, dall’interno del suo processo. Ma allora come fa main ad averla, per mandargli messaggi?

Con una piccola stretta di mano: main crea un Subject temporaneo e lo passa al processo nuovo; il processo crea la sua inbox e la spedisce indietro su quel Subject; main la riceve, e da lì in poi la usa.

fn start() -> Subject(Message) {
  let handshake = process.new_subject()
  process.spawn(fn() {
    let inbox = process.new_subject()
    process.send(handshake, inbox)
    loop(inbox, 0)
  })
  process.receive_forever(handshake)
}

Nota il tipo di handshake: è un Subject(Subject(Message)), un indirizzo a cui si mandano indirizzi. Un Subject è un valore come un altro, e viaggia nei messaggi come un numero.

Tutto insieme

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

pub type Message {
  Increment
  Get(reply_to: Subject(Int))
}

pub fn main() -> Nil {
  let counter = start()
  process.send(counter, Increment)
  process.send(counter, Increment)
  process.send(counter, Increment)
  let reply = process.new_subject()
  process.send(counter, Get(reply))
  let assert Ok(value) = process.receive(reply, within: 100)
  io.println("Counter: " <> int.to_string(value))
}

fn start() -> Subject(Message) {
  let handshake = process.new_subject()
  process.spawn(fn() {
    let inbox = process.new_subject()
    process.send(handshake, inbox)
    loop(inbox, 0)
  })
  process.receive_forever(handshake)
}

fn loop(inbox: Subject(Message), count: Int) -> Nil {
  case process.receive_forever(inbox) {
    Increment -> loop(inbox, count + 1)
    Get(reply_to) -> {
      process.send(reply_to, count)
      loop(inbox, count)
    }
  }
}

Counter: 3

I tre Increment e il Get partono da main, quindi arrivano in quest’ordine (la garanzia della lezione precedente): quando il contatore risponde, ha già contato fino a tre.

process.call

Chiedere qualcosa a un processo e aspettare la risposta è così comune che c’è una funzione apposta. Le tre righe della richiesta:

let reply = process.new_subject()
process.send(counter, Get(reply))
let assert Ok(value) = process.receive(reply, within: 100)

diventano una:

let value = process.call(counter, waiting: 100, sending: Get)

process.call crea da sola il Subject per la risposta, costruisce il messaggio chiamando la funzione sending con quel Subject, lo spedisce, e aspetta la risposta per al massimo waiting millisecondi. Una variante con dati si usa come una funzione (lezione 4.3), e infatti Get è una funzione fn(Subject(Int)) -> Message, ed è esattamente quello che sending vuole. Se il messaggio avesse altri dati, useresti una funzione anonima: sending: fn(reply) { Deposit(amount, reply) }.

E se la risposta non arriva in tempo? process.call non restituisce un Result: fa schiantare il processo che ha chiamato. Può sembrare brutale, ma è voluto: un processo che non risponde è un problema serio, e far finta di niente rischierebbe di lasciare il programma in uno stato confuso. Nella lezione sui supervisori vedremo perché sulla BEAM schiantarsi è una strategia, non una disgrazia.

Fermare il processo

Il contatore gira per sempre, o almeno finché gira main. Per poterlo fermare prima, basta un messaggio in più, in cui il ciclo non richiama sé stesso:

Stop -> Nil

La funzione del processo restituisce Nil, e un processo la cui funzione è finita termina. I messaggi che arrivano dopo finiscono nel vuoto: send a un processo morto non dà errori, semplicemente nessuno li leggerà mai.

pub type Message {
  Write(note: String)
  Read(reply_to: Subject(List(String)))
  Erase
}

Board: buy milk, call Joe
Board: (empty)

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

pub type Message {
  Write(note: String)
  Read(reply_to: Subject(List(String)))
  Erase
}

pub fn main() -> Nil {
  let board = start()
  process.send(board, Write("buy milk"))
  process.send(board, Write("call Joe"))
  show(process.call(board, waiting: 100, sending: Read))
  process.send(board, Erase)
  show(process.call(board, waiting: 100, sending: Read))
}

fn show(notes: List(String)) -> Nil {
  case notes {
    [] -> io.println("Board: (empty)")
    _ -> io.println("Board: " <> string.join(notes, with: ", "))
  }
}

fn start() -> Subject(Message) {
  let handshake = process.new_subject()
  process.spawn(fn() {
    let inbox = process.new_subject()
    process.send(handshake, inbox)
    loop(inbox, [])
  })
  process.receive_forever(handshake)
}

fn loop(inbox: Subject(Message), notes: List(String)) -> Nil {
  case process.receive_forever(inbox) {
    Write(note) -> loop(inbox, [note, ..notes])
    Read(reply_to) -> {
      process.send(reply_to, list.reverse(notes))
      loop(inbox, notes)
    }
    Erase -> loop(inbox, [])
  }
}

Ricapitolando

• Un processo con stato è una funzione ricorsiva che riceve un messaggio, calcola il nuovo stato e richiama sé stessa: lo stato è il suo argomento.
• I messaggi che il processo capisce si descrivono con un tipo; quelli che chiedono una risposta portano un Subject per la risposta.
• La stretta di mano: il processo nuovo crea la sua inbox e la spedisce a chi l’ha avviato.
• process.call(subject, waiting: ms, sending: Costruttore) manda una richiesta e aspetta la risposta; se non arriva in tempo, il chiamante si schianta.
• Il processo gestisce i messaggi uno alla volta: niente corse critiche.
• Un processo termina quando la sua funzione finisce.

Tutto questo funziona, ma è parecchio codice da riscrivere ogni volta: la stretta di mano, il ciclo, la ricorsione. Nella prossima lezione lo farà per noi gleam/otp/actor.