464 Advanced

464: Actor Pattern

Functional Programming

Tutorial

The Problem

Shared mutable state with locks leads to deadlocks, priority inversion, and complex reasoning. The actor model offers an alternative: each actor is an isolated entity with private state, communicating only through message passing. No locks, no shared memory — just messages. An actor processes messages sequentially, ensuring its state is never concurrently modified. This model was popularized by Erlang and is the foundation of actix, tokio::actor, and any system requiring encapsulated concurrent state.

Actor systems power chat servers (each user is an actor), game entities (each NPC is an actor), distributed systems (each node is an actor), and the Actix web framework.

🎯 Learning Outcomes

• Understand the actor model: isolated state, message-driven behavior, no shared memory

• Learn how mpsc::Sender<Msg> serves as the actor handle (message address)

• See how an actor loop (for m in rx) processes messages sequentially

• Understand request-response with mpsc::SyncSender<i64> in the message for synchronous queries

• Learn why actors eliminate lock-based concurrency bugs

Code Example

#![allow(clippy::all)]
// 464. Actor model in Rust
use std::sync::mpsc;
use std::thread;

enum Msg {
    Inc(i64),
    Dec(i64),
    Get(mpsc::SyncSender<i64>),
    Reset,
    Stop,
}

struct Actor {
    tx: mpsc::Sender<Msg>,
}

impl Actor {
    fn new() -> Self {
        let (tx, rx) = mpsc::channel::<Msg>();
        thread::spawn(move || {
            let mut s = 0i64;
            for m in rx {
                match m {
                    Msg::Inc(n) => s += n,
                    Msg::Dec(n) => s -= n,
                    Msg::Get(tx) => {
                        let _ = tx.send(s);
                    }
                    Msg::Reset => s = 0,
                    Msg::Stop => break,
                }
            }
        });
        Actor { tx }
    }
    fn inc(&self, n: i64) {
        self.tx.send(Msg::Inc(n)).unwrap();
    }
    fn dec(&self, n: i64) {
        self.tx.send(Msg::Dec(n)).unwrap();
    }
    fn reset(&self) {
        self.tx.send(Msg::Reset).unwrap();
    }
    fn get(&self) -> i64 {
        let (t, r) = mpsc::sync_channel(1);
        self.tx.send(Msg::Get(t)).unwrap();
        r.recv().unwrap()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    #[test]
    fn test_actor() {
        let a = Actor::new();
        a.inc(7);
        a.inc(3);
        assert_eq!(a.get(), 10);
        a.dec(4);
        assert_eq!(a.get(), 6);
        a.reset();
        assert_eq!(a.get(), 0);
    }
}

(* 464. Actor model – OCaml *)
type msg = Inc of int | Get of int Queue.t | Stop

let make_actor () =
  let q=Queue.create () let m=Mutex.create () let c=Condition.create () in
  let send v=Mutex.lock m; Queue.push v q; Condition.signal c; Mutex.unlock m in
  let recv ()=Mutex.lock m; while Queue.is_empty q do Condition.wait c m done;
    let v=Queue.pop q in Mutex.unlock m; v in
  ignore (Thread.create (fun () ->
    let st=ref 0 and go=ref true in
    while !go do match recv () with
    | Inc n -> st := !st+n
    | Get rq -> Queue.push !st rq
    | Stop -> go:=false
    done) ());
  send

let () =
  let send=make_actor () in
  send (Inc 10); send (Inc 5);
  let rq=Queue.create () in send (Get rq);
  (* spin wait *)
  while Queue.is_empty rq do Thread.delay 0.001 done;
  Printf.printf "state=%d\n" (Queue.pop rq);
  send Stop; Thread.delay 0.01

Key Differences

Message enum: Rust uses a Msg enum for all messages to one actor; Erlang/OCaml typically use dynamic dispatch or 'a polymorphic messages.

Request-response: Rust requires a reply channel in the message; Erlang has built-in receive with ! for replies.

Actor identity: Rust actors are Sender<Msg> handles; Erlang has process IDs (PIDs) as first-class values.

Supervision: Erlang has built-in supervision trees; Rust's tokio::actor pattern requires manual error handling.

OCaml Approach

OCaml's actor model is natural with Event channels and the Thread module: each actor is a thread with its own channel. Erlang-style actors in OCaml use the Eio or Mirage effect-based runtimes. The actor library provides Erlang-style process spawning. OCaml's lightweight threads in OCaml 5.x (via Eio.Fiber) make actor systems more efficient than OS-thread-based implementations.

Full Source

#![allow(clippy::all)]
// 464. Actor model in Rust
use std::sync::mpsc;
use std::thread;

enum Msg {
    Inc(i64),
    Dec(i64),
    Get(mpsc::SyncSender<i64>),
    Reset,
    Stop,
}

struct Actor {
    tx: mpsc::Sender<Msg>,
}

impl Actor {
    fn new() -> Self {
        let (tx, rx) = mpsc::channel::<Msg>();
        thread::spawn(move || {
            let mut s = 0i64;
            for m in rx {
                match m {
                    Msg::Inc(n) => s += n,
                    Msg::Dec(n) => s -= n,
                    Msg::Get(tx) => {
                        let _ = tx.send(s);
                    }
                    Msg::Reset => s = 0,
                    Msg::Stop => break,
                }
            }
        });
        Actor { tx }
    }
    fn inc(&self, n: i64) {
        self.tx.send(Msg::Inc(n)).unwrap();
    }
    fn dec(&self, n: i64) {
        self.tx.send(Msg::Dec(n)).unwrap();
    }
    fn reset(&self) {
        self.tx.send(Msg::Reset).unwrap();
    }
    fn get(&self) -> i64 {
        let (t, r) = mpsc::sync_channel(1);
        self.tx.send(Msg::Get(t)).unwrap();
        r.recv().unwrap()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    #[test]
    fn test_actor() {
        let a = Actor::new();
        a.inc(7);
        a.inc(3);
        assert_eq!(a.get(), 10);
        a.dec(4);
        assert_eq!(a.get(), 6);
        a.reset();
        assert_eq!(a.get(), 0);
    }
}

(* 464. Actor model – OCaml *)
type msg = Inc of int | Get of int Queue.t | Stop

let make_actor () =
  let q=Queue.create () let m=Mutex.create () let c=Condition.create () in
  let send v=Mutex.lock m; Queue.push v q; Condition.signal c; Mutex.unlock m in
  let recv ()=Mutex.lock m; while Queue.is_empty q do Condition.wait c m done;
    let v=Queue.pop q in Mutex.unlock m; v in
  ignore (Thread.create (fun () ->
    let st=ref 0 and go=ref true in
    while !go do match recv () with
    | Inc n -> st := !st+n
    | Get rq -> Queue.push !st rq
    | Stop -> go:=false
    done) ());
  send

let () =
  let send=make_actor () in
  send (Inc 10); send (Inc 5);
  let rq=Queue.create () in send (Get rq);
  (* spin wait *)
  while Queue.is_empty rq do Thread.delay 0.001 done;
  Printf.printf "state=%d\n" (Queue.pop rq);
  send Stop; Thread.delay 0.01

✓ Tests Rust test suite

#[cfg(test)]
mod tests {
    use super::*;
    #[test]
    fn test_actor() {
        let a = Actor::new();
        a.inc(7);
        a.inc(3);
        assert_eq!(a.get(), 10);
        a.dec(4);
        assert_eq!(a.get(), 6);
        a.reset();
        assert_eq!(a.get(), 0);
    }
}

Exercises

Bank account: Implement a BankAccount actor with Deposit(Amount), Withdraw(Amount), Balance(SyncSender<i64>) messages. Verify concurrent deposits and withdrawals produce correct balances without locks.

Actor supervision: Create a supervisor that monitors an actor by joining its thread handle. If the actor panics, the supervisor automatically restarts it with fresh state. Implement a counter of restarts.

Actor network: Implement a simple pub-sub system where a Broker actor routes Subscribe(topic, reply_tx) and Publish(topic, payload) messages. Subscribers receive published messages for their topics.

Open Source Repos

functional-rust

View the source for this example on GitHub — OCaml and Rust side by side in the repo.

Rust