Attori

Lo schema, già pronto

Il contatore della lezione precedente aveva tre pezzi: la stretta di mano per avere l’inbox, il ciclo ricorsivo con lo stato, e la funzione che decide cosa fare con ogni messaggio. Solo il terzo pezzo era davvero nostro: gli altri due sarebbero identici in qualsiasi processo con stato.

Il pacchetto gleam_otp li ha già scritti, nel modulo gleam/otp/actor. Un attore è proprio questo: un processo che tiene uno stato e reagisce ai messaggi, uno alla volta. Tu fornisci lo stato iniziale e la funzione che gestisce un messaggio; al resto pensa l’attore.

Il contatore, versione attore

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

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

pub fn main() -> Nil {
  let assert Ok(started) =
    actor.new(0)
    |> actor.on_message(handle_message)
    |> actor.start
  let counter = started.data
  actor.send(counter, Increment)
  actor.send(counter, Increment)
  actor.send(counter, Increment)
  let value = actor.call(counter, waiting: 100, sending: Get)
  io.println("Counter: " <> int.to_string(value))
}

fn handle_message(count: Int, message: Message) -> actor.Next(Int, Message) {
  case message {
    Increment -> actor.continue(count + 1)
    Get(reply_to) -> {
      actor.send(reply_to, count)
      actor.continue(count)
    }
  }
}

Counter: 3

Confrontalo con la versione della lezione precedente: il tipo Message è lo stesso, e handle_message assomiglia molto a loop. Le differenze:

• Niente ricorsione: handle_message gestisce un messaggio e restituisce cosa fare dopo. È l’attore che la richiama a ogni messaggio, con lo stato nuovo.
• Niente receive: il messaggio arriva già come argomento, insieme allo stato attuale.
• Niente stretta di mano: actor.start avvia il processo, aspetta che sia pronto e restituisce il suo Subject.

La costruzione dell’attore è una catena di tubi: actor.new(0) prepara un attore con stato iniziale 0, actor.on_message gli dà la funzione per i messaggi, actor.start lo avvia. actor.send e actor.call sono le stesse process.send e process.call, ripetute nel modulo actor per comodità.

Cosa restituisce actor.start

actor.start restituisce un Result, perché avviare un attore può fallire (vedremo come nella prossima lezione). Se va bene, dentro Ok c’è un record Started con due campi:

• pid: il Pid del processo dell’attore;
• data: il Subject per mandargli i messaggi.

Per questo scriviamo let assert Ok(started) = ... e poi usiamo started.data. Se l’avvio fallisse, il let assert farebbe schiantare main: per un programma che senza il suo contatore non ha senso, è la scelta giusta.

actor.Next

La funzione dei messaggi restituisce un actor.Next(stato, messaggio), che dice all’attore come proseguire. Ci sono due modi principali di costruirlo:

actor.Next è un tipo opaco (lezione 6.4): non puoi costruirlo a mano, solo con queste funzioni.

Un modulo che nasconde i messaggi

Nel programma sopra, chi usa il contatore deve conoscere i suoi messaggi: Increment, Get, e il fatto che Get vuole un Subject. È un dettaglio interno, come i campi di un tipo opaco. Nei programmi Gleam si usa quasi sempre lo stesso schema: l’attore sta in un modulo suo, e il modulo espone delle normali funzioni, start, increment, get, che mandano i messaggi al posto tuo.

import gleam/erlang/process.{type Subject}
import gleam/otp/actor
import gleam/result

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

pub fn start() -> Result(Subject(Message), actor.StartError) {
  actor.new(0)
  |> actor.on_message(handle_message)
  |> actor.start
  |> result.map(fn(started) { started.data })
}

pub fn increment(counter: Subject(Message)) -> Nil {
  actor.send(counter, Increment)
}

pub fn get(counter: Subject(Message)) -> Int {
  actor.call(counter, waiting: 100, sending: Get)
}

fn handle_message(count: Int, message: Message) -> actor.Next(Int, Message) {
  case message {
    Increment -> actor.continue(count + 1)
    Get(reply_to) -> {
      actor.send(reply_to, count)
      actor.continue(count)
    }
  }
}

import counter
import gleam/int
import gleam/io

pub fn main() -> Nil {
  let assert Ok(visits) = counter.start()
  counter.increment(visits)
  counter.increment(visits)
  io.println("Visits: " <> int.to_string(counter.get(visits)))
}

Visits: 2

Il tipo Message è opaco: da fuori si può tenere in mano un Subject(Message), ma non si possono costruire messaggi, quindi l’unico modo di parlare con il contatore sono le funzioni del modulo. Chi scrive counter.increment(visits) non sa nemmeno che dietro c’è un processo. E se un giorno volessi aggiungere un messaggio, o cambiare lo stato, cambieresti solo counter.gleam.

Quando una chiamata scade

Cosa succede se l’attore ci mette più tempo del previsto a rispondere? Qui l’attore impiega un secondo a “pensare”, ma chi chiama ha pazienza solo per 100 millisecondi:

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

pub type Message {
  Think(reply_to: Subject(Int))
}

pub fn main() -> Nil {
  let assert Ok(started) =
    actor.new(42)
    |> actor.on_message(handle_message)
    |> actor.start
  let answer = actor.call(started.data, waiting: 100, sending: Think)
  io.println("The answer is " <> int.to_string(answer))
}

fn handle_message(answer: Int, message: Message) -> actor.Next(Int, Message) {
  case message {
    Think(reply_to) -> {
      process.sleep(1000)
      actor.send(reply_to, answer)
      actor.continue(answer)
    }
  }
}

runtime error: let assert

callee did not send reply before timeout

unmatched value:
  Error(Nil)

stacktrace:
  gleam/erlang/process.perform_call src/gleam/erlang/process.gleam:630
  slow_actor.main src/slow_actor.gleam:19

È l’errore della lezione 5.4: un let assert fallito, dentro process.call, con il messaggio callee did not send reply before timeout (“il chiamato non ha risposto entro il tempo”). È main, il chiamante, a schiantarsi; e come sempre quando main termina, il programma si spegne con tutto il resto.

Quanto aspettare dipende da cosa fa l’attore: per un contatore in memoria, 100 millisecondi sono un’eternità; per un attore che interroga un server dall’altra parte del mondo, possono essere pochi.

Ticket 1
Ticket 2
Ticket 3
Serving 1
Serving 2
Serving 3
Nobody waiting

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

pub type State {
  State(next_ticket: Int, now_serving: Int)
}

pub type Message {
  Take(reply_to: Subject(Int))
  Serve(reply_to: Subject(Result(Int, Nil)))
}

pub fn main() -> Nil {
  let assert Ok(started) =
    actor.new(State(next_ticket: 1, now_serving: 1))
    |> actor.on_message(handle_message)
    |> actor.start
  let machine = started.data
  print_ticket(take(machine))
  print_ticket(take(machine))
  print_ticket(take(machine))
  call_next(machine)
  call_next(machine)
  call_next(machine)
  call_next(machine)
}

fn take(machine: Subject(Message)) -> Int {
  actor.call(machine, waiting: 100, sending: Take)
}

fn serve(machine: Subject(Message)) -> Result(Int, Nil) {
  actor.call(machine, waiting: 100, sending: Serve)
}

fn print_ticket(ticket: Int) -> Nil {
  io.println("Ticket " <> int.to_string(ticket))
}

fn call_next(machine: Subject(Message)) -> Nil {
  case serve(machine) {
    Ok(ticket) -> io.println("Serving " <> int.to_string(ticket))
    Error(Nil) -> io.println("Nobody waiting")
  }
}

fn handle_message(
  state: State,
  message: Message,
) -> actor.Next(State, Message) {
  case message {
    Take(reply_to) -> {
      actor.send(reply_to, state.next_ticket)
      actor.continue(State(..state, next_ticket: state.next_ticket + 1))
    }
    Serve(reply_to) ->
      case state.now_serving < state.next_ticket {
        True -> {
          actor.send(reply_to, Ok(state.now_serving))
          actor.continue(State(..state, now_serving: state.now_serving + 1))
        }
        False -> {
          actor.send(reply_to, Error(Nil))
          actor.continue(state)
        }
      }
  }
}

Ricapitolando

• Un attore (gleam/otp/actor) è un processo con stato già pronto: tu scrivi solo la funzione che gestisce un messaggio.
• actor.new(stato) |> actor.on_message(gestore) |> actor.start avvia l’attore; il risultato è Ok(Started(pid:, data:)), e data è il Subject.
• Il gestore riceve lo stato e il messaggio, e restituisce un actor.Next: actor.continue(nuovo_stato) o actor.stop().
• actor.send e actor.call sono process.send e process.call.
• Lo schema comune: un modulo per l’attore, con pub opaque type Message e funzioni pubbliche che mandano i messaggi.
• Se call non riceve risposta entro il timeout, il chiamante si schianta.

Nella prossima lezione faremo schiantare apposta un attore, e vedremo chi lo rimette in piedi.