Add comprehensive actor model implementation guide
ober
9d3439f7de2dd0fc013f090dfbd415af6e12ad13
new file mode 100644 --- /dev/null +++ b/docs/actor-model.md @@ -0,0 +1,2176 @@ +# Jerboa Actor Model: Complete Implementation Guide + +This document is a step-by-step implementation guide for building a production-quality +actor system on Chez Scheme. Each layer is independently implementable and testable. +A lesser model can implement this by following the layers in order — do not skip ahead. + +--- + +## Design Philosophy + +**Goals** (what makes this better than Gerbil's `:std/actor`): + +1. **Clean layer separation** — each layer is independently importable and testable. + Gerbil mixes local spawn, remote RPC, filesystem deployment, and admin auth into + a single 400-symbol namespace. Here each layer is a separate library. + +2. **No shimming** — built directly on Chez's native OS threads, not green threads. + Every primitive maps directly to a Chez or OS concept. + +3. **Native serialization** — use Chez's built-in `fasl-write`/`fasl-read` for + distributed transport. Any Scheme value is automatically serializable. No separate + serialization library needed. + +4. **OTP-style supervision** — Erlang-proven restart strategies (one-for-one, + one-for-all, rest-for-one) with max-intensity/period restart limiting. + +5. **Location transparency** — `(send actor-ref msg)` works whether `actor-ref` is + local or remote. The caller does not need to know. + +6. **Typed protocols via macros** — `defprotocol` generates message structs and + typed dispatch. Less boilerplate than Gerbil's `defmessage` + `defcall-actor`. + +7. **Gradual complexity** — Layers 2-4 (local actors + supervision) are useful + without Layer 1 (work-stealing) or Layer 7 (distributed). Build and ship each + layer independently. + +**Non-goals**: +- Full Gerbil compatibility (we implement what real programs use, not every symbol) +- Green threads / continuations (real OS threads are simpler and SMP-safe) +- Hot code loading (out of scope) + +--- + +## Architecture Overview + +``` +┌──────────────────────────────────────────────────────┐ +│ Layer 7: Distributed Transport │ +│ lib/std/actor/transport.sls, remote.sls, node.sls │ +│ TCP+TLS, fasl serialization, location transparency │ +├──────────────────────────────────────────────────────┤ +│ Layer 6: Registry │ +│ lib/std/actor/registry.sls │ +│ Named actors, whereis, register, unregister │ +├──────────────────────────────────────────────────────┤ +│ Layer 5: Supervision Trees │ +│ lib/std/actor/supervisor.sls │ +│ OTP strategies, restart intensity, child specs │ +├──────────────────────────────────────────────────────┤ +│ Layer 4: Protocol System │ +│ lib/std/actor/protocol.sls │ +│ defprotocol, ask, tell, call, pattern dispatch │ +├──────────────────────────────────────────────────────┤ +│ Layer 3: Actor Core │ +│ lib/std/actor/core.sls │ +│ spawn-actor, send, receive, self, dead letters │ +├──────────────────────────────────────────────────────┤ +│ Layer 2: Scheduler │ +│ lib/std/actor/scheduler.sls │ +│ Work-stealing thread pool, lightweight tasks │ +├──────────────────────────────────────────────────────┤ +│ Layer 1: Data Structures │ +│ lib/std/actor/mpsc.sls — MPSC queue (mailbox) │ +│ lib/std/actor/deque.sls — Work-stealing deque │ +├──────────────────────────────────────────────────────┤ +│ Foundation (already exists in Jerboa) │ +│ (std misc channel) — bounded channels + select │ +│ (std misc thread) — Gambit thread API │ +│ (std task) — task groups + futures │ +│ (std net ssl) — TCP+TLS via chez-ssl │ +└──────────────────────────────────────────────────────┘ +``` + +**Implementation order**: Layer 1 → Layer 3 → Layer 4 → Layer 5 → Layer 6 → Layer 2 → Layer 7. +Layer 2 (scheduler) can be deferred — Layers 3-6 work fine on 1:1 OS threads initially. + +--- + +## Layer 1A: MPSC Queue (`lib/std/actor/mpsc.sls`) + +### Purpose + +Each actor has a mailbox. Multiple threads (producers) can send messages to it +concurrently. Only the actor's own thread (consumer) reads from it. +This is the Multi-Producer Single-Consumer (MPSC) pattern. + +### Data Structure: Lock-Based Linked List + +We use a two-lock linked list: one lock for the tail (producers) and one lock for +the head (consumer). This minimizes contention because producers never block the +consumer and vice versa, except in the rare empty/single-element cases. + +This is simpler and more practical for Chez than a lock-free Michael-Scott queue +(which would require `compare-and-swap` via FFI C shims). + +```scheme +#!chezscheme +(library (std actor mpsc) + (export + make-mpsc-queue + mpsc-queue? + mpsc-enqueue! ;; producer: add to tail + mpsc-dequeue! ;; consumer: remove from head (blocks if empty) + mpsc-try-dequeue! ;; consumer: remove or return #f immediately + mpsc-empty? ;; peek (approximate — only safe from consumer) + mpsc-close! ;; signal no more messages + mpsc-closed?) + (import (chezscheme)) + + ;; Node in the linked list + (define-record-type mpsc-node + (fields + (mutable value) ;; the message, or 'sentinel for dummy head + (mutable next)) ;; next node or #f + (protocol + (lambda (new) + (lambda (val) (new val #f)))) + (sealed #t)) + + (define-record-type mpsc-queue + (fields + (mutable head) ;; points to dummy node; consumer reads head.next + (mutable tail) ;; points to last real node (or dummy if empty) + (immutable head-mutex) ;; consumer lock + (immutable tail-mutex) ;; producer lock + (immutable not-empty) ;; condition: signaled when item enqueued + (mutable closed?) + (mutable count)) ;; approximate item count + (protocol + (lambda (new) + (lambda () + (let ([dummy (make-mpsc-node 'sentinel)]) + (new dummy dummy + (make-mutex) (make-mutex) + (make-condition) + #f 0))))) + (sealed #t)) + + ;; Producer: enqueue a value + ;; Lock only the tail — does not interfere with consumer reading head + (define (mpsc-enqueue! q val) + (let ([node (make-mpsc-node val)]) + (mutex-acquire (mpsc-queue-tail-mutex q)) + (when (mpsc-queue-closed? q) + (mutex-release (mpsc-queue-tail-mutex q)) + (error 'mpsc-enqueue! "queue is closed")) + (mpsc-node-next-set! (mpsc-queue-tail q) node) + (mpsc-queue-tail-set! q node) + (mpsc-queue-count-set! q (fx+ (mpsc-queue-count q) 1)) + ;; Signal consumer (must hold head-mutex to signal safely) + (mutex-acquire (mpsc-queue-head-mutex q)) + (condition-signal (mpsc-queue-not-empty q)) + (mutex-release (mpsc-queue-head-mutex q)) + (mutex-release (mpsc-queue-tail-mutex q)))) + + ;; Consumer: dequeue, blocking if empty + (define (mpsc-dequeue! q) + (mutex-acquire (mpsc-queue-head-mutex q)) + (let loop () + (let ([next (mpsc-node-next (mpsc-queue-head q))]) + (cond + [next + ;; Advance dummy head to the first real node + ;; The old head is discarded; the real node becomes the new dummy + (let ([val (mpsc-node-value next)]) + (mpsc-queue-head-set! q next) + (mpsc-node-value-set! next 'sentinel) ;; help GC + (mpsc-queue-count-set! q (fx- (mpsc-queue-count q) 1)) + (mutex-release (mpsc-queue-head-mutex q)) + val)] + [(mpsc-queue-closed? q) + (mutex-release (mpsc-queue-head-mutex q)) + (error 'mpsc-dequeue! "queue closed and empty")] + [else + (condition-wait (mpsc-queue-not-empty q) + (mpsc-queue-head-mutex q)) + (loop)])))) + + ;; Consumer: try dequeue without blocking + ;; Returns (values val #t) if successful, (values #f #f) if empty + (define (mpsc-try-dequeue! q) + (mutex-acquire (mpsc-queue-head-mutex q)) + (let ([next (mpsc-node-next (mpsc-queue-head q))]) + (cond + [next + (let ([val (mpsc-node-value next)]) + (mpsc-queue-head-set! q next) + (mpsc-node-value-set! next 'sentinel) + (mpsc-queue-count-set! q (fx- (mpsc-queue-count q) 1)) + (mutex-release (mpsc-queue-head-mutex q)) + (values val #t))] + [else + (mutex-release (mpsc-queue-head-mutex q)) + (values #f #f)]))) + + (define (mpsc-empty? q) + (not (mpsc-node-next (mpsc-queue-head q)))) + + (define (mpsc-close! q) + (mutex-acquire (mpsc-queue-tail-mutex q)) + (mpsc-queue-closed?-set! q #t) + (mutex-acquire (mpsc-queue-head-mutex q)) + (condition-broadcast (mpsc-queue-not-empty q)) + (mutex-release (mpsc-queue-head-mutex q)) + (mutex-release (mpsc-queue-tail-mutex q))) + + ) ;; end library +``` + +**Test file**: `tests/test-actor-mpsc.ss` +- Enqueue from 10 threads simultaneously, dequeue from 1 thread — verify all messages received +- try-dequeue on empty queue returns `#f` +- Close while consumer is blocked — consumer gets error +- Count is approximately correct after concurrent operations + +--- + +## Layer 1B: Work-Stealing Deque (`lib/std/actor/deque.sls`) + +### Purpose + +The scheduler (Layer 2) gives each worker thread its own double-ended queue of tasks. +The owner thread pushes/pops from the bottom. Idle workers steal from the top of +other workers' deques. This is the Chase-Lev work-stealing deque. + +### Implementation Note + +A fully lock-free Chase-Lev deque requires `compare-and-swap` (CAS) on memory +words — an operation Chez does not expose natively. Two options: + +**Option A (Recommended for initial implementation)**: Use a single mutex per deque. +The deque is mostly uncontended (owner push/pop), and stealing is rare. A mutex +is fast enough when not under heavy contention. + +**Option B (For high-throughput production)**: Add a C shim: +```c +// support/atomic.c +#include <stdatomic.h> +#include <stdint.h> + +// Returns 1 if swap succeeded, 0 if not +int cas_int64(int64_t *ptr, int64_t expected, int64_t desired) { + return atomic_compare_exchange_strong( + (atomic_int_fast64_t*)ptr, &expected, desired); +} +``` +Compile as `libjerboa-atomic.so` and load via `(load-shared-object ...)`. +Then use `(foreign-procedure "cas_int64" (void* integer-64 integer-64) integer-32)`. + +The document describes the mutex-based version. Upgrading to lock-free is a drop-in +replacement at the deque level — the scheduler above does not change. + +```scheme +#!chezscheme +(library (std actor deque) + (export + make-work-deque + work-deque? + deque-push-bottom! ;; owner pushes a task + deque-pop-bottom! ;; owner pops (LIFO — locality of reference) + deque-steal-top! ;; thief steals (FIFO — oldest tasks first) + deque-empty? + deque-size) + (import (chezscheme)) + + ;; Circular buffer that grows as needed + (define-record-type work-deque + (fields + (mutable buf) ;; vector of tasks + (mutable bottom) ;; owner's end (push/pop here) + (mutable top) ;; thief's end (steal from here) + (immutable mutex)) + (protocol + (lambda (new) + (lambda () + (new (make-vector 64 #f) 0 0 (make-mutex))))) + (sealed #t)) + + (define (deque-capacity d) (vector-length (work-deque-buf d))) + + (define (deque-size d) + (let ([b (work-deque-bottom d)] + [t (work-deque-top d)]) + (if (fx>= b t) (fx- b t) 0))) + + (define (deque-empty? d) + (fx<= (work-deque-bottom d) (work-deque-top d))) + + ;; Grow buffer when full + (define (deque-grow! d) + (let* ([old (work-deque-buf d)] + [old-cap (vector-length old)] + [new-cap (fx* old-cap 2)] + [new-buf (make-vector new-cap #f)] + [top (work-deque-top d)] + [bottom (work-deque-bottom d)]) + (do ([i top (fx+ i 1)]) + ((fx= i bottom)) + (vector-set! new-buf (fxmod i new-cap) + (vector-ref old (fxmod i old-cap)))) + (work-deque-buf-set! d new-buf))) + + ;; Owner pushes a task to the bottom + (define (deque-push-bottom! d task) + (mutex-acquire (work-deque-mutex d)) + (let ([b (work-deque-bottom d)]) + (when (fx>= (fx- b (work-deque-top d)) (fx- (deque-capacity d) 1)) + (deque-grow! d)) + (vector-set! (work-deque-buf d) (fxmod b (deque-capacity d)) task) + (work-deque-bottom-set! d (fx+ b 1))) + (mutex-release (work-deque-mutex d))) + + ;; Owner pops from the bottom (LIFO — most recently pushed task first) + ;; Returns the task or #f if empty + (define (deque-pop-bottom! d) + (mutex-acquire (work-deque-mutex d)) + (let ([b (fx- (work-deque-bottom d) 1)]) + (work-deque-bottom-set! d b) + (let ([result + (if (fx< (work-deque-top d) b) + ;; Non-empty: take from bottom + (let ([task (vector-ref (work-deque-buf d) + (fxmod b (deque-capacity d)))]) + (vector-set! (work-deque-buf d) (fxmod b (deque-capacity d)) #f) + task) + ;; Empty or contested + (begin + (work-deque-bottom-set! d (fx+ b 1)) + #f))]) + (mutex-release (work-deque-mutex d)) + result))) + + ;; Thief steals from the top (FIFO — oldest tasks first) + ;; Returns (values task #t) or (values #f #f) if empty + (define (deque-steal-top! d) + (mutex-acquire (work-deque-mutex d)) + (let ([t (work-deque-top d)] + [b (work-deque-bottom d)]) + (cond + [(fx>= t b) + (mutex-release (work-deque-mutex d)) + (values #f #f)] + [else + (let ([task (vector-ref (work-deque-buf d) + (fxmod t (deque-capacity d)))]) + (vector-set! (work-deque-buf d) (fxmod t (deque-capacity d)) #f) + (work-deque-top-set! d (fx+ t 1)) + (mutex-release (work-deque-mutex d)) + (values task #t))]))) + + ) ;; end library +``` + +--- + +## Layer 2: Work-Stealing Scheduler (`lib/std/actor/scheduler.sls`) + +### Purpose + +Instead of one OS thread per actor (which limits concurrency to ~1000), the scheduler +maintains a fixed pool of N OS threads (default: `(cpu-count)`) and schedules +lightweight tasks across them. M tasks run on N threads (M >> N). + +### Key Insight + +Actors are NOT OS threads. An actor is a record with a mailbox. When a message arrives, +a **task** (a thunk) is scheduled to run the actor's receive loop for one message. +The task runs on whatever worker thread picks it up. This is the M:N model. + +### Data Structures + +```scheme +#!chezscheme +(library (std actor scheduler) + (export + scheduler-start! ;; create and start the thread pool + scheduler-stop! ;; drain and shut down + scheduler-submit! ;; submit a thunk as a task + scheduler-worker-count + current-scheduler + default-scheduler) + (import (chezscheme) (std actor deque)) + + ;; A task is just a thunk (zero-argument procedure) + ;; The scheduler runs thunks; it doesn't know about actors. + + ;; Per-worker state (one per OS thread in the pool) + (define-record-type worker + (fields + (immutable id) ;; integer index 0..N-1 + (immutable deque) ;; this worker's task deque + (immutable thread-id) ;; Chez thread id (set after start) + (mutable running?)) ;; #f when shutting down + (protocol + (lambda (new) + (lambda (id) + (new id (make-work-deque) #f #t)))) + (sealed #t)) + + ;; The scheduler: a pool of workers + (define-record-type scheduler + (fields + (immutable workers) ;; vector of worker records + (immutable global-queue) ;; overflow queue for load balancing + (immutable mutex) + (immutable work-available) ;; condition: broadcast when new task added + (mutable running?)) + (protocol + (lambda (new) + (lambda (n) + (new (let ([v (make-vector n)]) + (do ([i 0 (fx+ i 1)]) ((fx= i n) v) + (vector-set! v i (make-worker i)))) + (make-vector 0) ;; simple global queue (vector for now) + (make-mutex) + (make-condition) + #f)))) + (sealed #t)) + + ;; Thread-local: which worker is running on this thread + (define current-worker (make-thread-parameter #f)) + (define current-scheduler (make-thread-parameter #f)) + (define default-scheduler (make-parameter #f)) + + ;; Submit a task to the scheduler + ;; If called from a worker thread, push to its local deque (fast path). + ;; Otherwise, distribute round-robin to worker deques. + (define (scheduler-submit! sched thunk) + (let ([w (current-worker)]) + (if w + ;; Fast path: running inside the pool — push to local deque + (begin + (deque-push-bottom! (worker-deque w) thunk) + (mutex-acquire (scheduler-mutex sched)) + (condition-signal (scheduler-work-available sched)) + (mutex-release (scheduler-mutex sched))) + ;; Slow path: external submission — pick a worker round-robin + (let* ([workers (scheduler-workers sched)] + [n (vector-length workers)] + [idx (fxmod (random n) n)] ;; randomized for load balance + [w (vector-ref workers idx)]) + (deque-push-bottom! (worker-deque w) thunk) + (mutex-acquire (scheduler-mutex sched)) + (condition-signal (scheduler-work-available sched)) + (mutex-release (scheduler-mutex sched)))))) + + ;; The main loop for each worker thread + (define (worker-run! sched w) + (current-worker w) + (current-scheduler sched) + (let ([workers (scheduler-workers sched)] + [n (vector-length (scheduler-workers sched))]) + (let loop () + (when (scheduler-running? sched) + ;; 1. Try own deque first + (let ([task (deque-pop-bottom! (worker-deque w))]) + (if task + (begin + (guard (exn [#t (void)]) ;; tasks must not crash the worker + (task)) + (loop)) + ;; 2. Try stealing from a random other worker + (let try-steal ([attempts 0]) + (if (fx= attempts n) + ;; 3. All deques empty — wait for work + (begin + (mutex-acquire (scheduler-mutex sched)) + (condition-wait (scheduler-work-available sched) + (scheduler-mutex sched)) + (mutex-release (scheduler-mutex sched)) + (loop)) + (let* ([victim-idx (fxmod (fx+ (worker-id w) attempts 1) n)] + [victim (vector-ref workers victim-idx)]) + (let-values ([(task ok) (deque-steal-top! (worker-deque victim))]) + (if ok + (begin + (guard (exn [#t (void)]) + (task)) + (loop)) + (try-steal (fx+ attempts 1))))))))))))) + + (define (scheduler-worker-count sched) + (vector-length (scheduler-workers sched))) + + (define (scheduler-start! sched) + (scheduler-running?-set! sched #t) + (let ([workers (scheduler-workers sched)]) + (do ([i 0 (fx+ i 1)]) + ((fx= i (vector-length workers))) + (let ([w (vector-ref workers i)]) + (fork-thread (lambda () (worker-run! sched w)))))) + sched) + + (define (scheduler-stop! sched) + (scheduler-running?-set! sched #f) + (mutex-acquire (scheduler-mutex sched)) + (condition-broadcast (scheduler-work-available sched)) + (mutex-release (scheduler-mutex sched))) + + ) ;; end library +``` + +### Usage Notes + +- `(scheduler-submit! sched thunk)` is the ONLY way tasks enter the pool. +- Tasks must complete quickly (not block indefinitely) for good throughput. + An actor that blocks on `receive` should suspend and re-submit when a message arrives. +- Exception isolation: each task is wrapped in `guard` so a crashing task does not + kill the worker thread. The actor's supervisor handles the crash, not the scheduler. + +### Initialization + +```scheme +;; Typically done once at program start: +(define sched (scheduler-start! (make-scheduler (cpu-count)))) +(default-scheduler sched) +``` + +--- + +## Layer 3: Actor Core (`lib/std/actor/core.sls`) + +### Core Concepts + +- An **actor** is a record containing: an ID, a mailbox (MPSC queue), a behavior + function, and lifecycle state. +- **Spawning** creates the actor record and schedules its initial run. +- **Sending** enqueues a message in the actor's mailbox and schedules a task if + the actor is idle. +- **Receiving** is done inside the behavior function. The behavior processes one + message per scheduling quantum. +- **self** is a thread-local parameter bound to the current actor's reference. + +### Actor States + +``` + IDLE ──────────[message arrives]──────> SCHEDULED + ^ | + | [task runs] + | | + └────────────[mailbox empty again]──── RUNNING + | + [exception] + | + DEAD ──> supervisor notified +``` + +### The Actor Loop Model + +Unlike Erlang where each actor is a persistent process with a `receive` call that +blocks, in our M:N model an actor runs as a **task per message**: + +1. Message arrives in mailbox +2. A task is submitted to the scheduler: `(lambda () (run-actor! actor))` +3. `run-actor!` dequeues one message, calls `(behavior msg)`, then: + - If more messages in mailbox, re-submits itself + - If mailbox empty, marks actor as IDLE + +This avoids blocking a worker thread while waiting for messages. + +```scheme +#!chezscheme +(library (std actor core) + (export + ;; Actor creation and management + spawn-actor ;; (spawn-actor behavior [name]) → actor-ref + spawn-actor/linked ;; (spawn-actor/linked behavior) → actor-ref + ;; links to current actor; if either dies, both die + actor-ref? + actor-ref-id + actor-ref-node ;; #f for local actors + + ;; Sending messages + send ;; (send actor-ref msg) → unspecified (fire and forget) + send/timeout ;; (send actor-ref msg timeout-secs) + + ;; Receiving inside a behavior + ;; Note: receive is only valid inside a spawn-actor behavior + self ;; (self) → current actor's actor-ref + actor-id ;; (actor-id) → current actor's id integer + + ;; Actor lifecycle + actor-alive? ;; (actor-alive? actor-ref) → bool + actor-kill! ;; (actor-kill! actor-ref) → forcibly terminate + actor-wait! ;; (actor-wait! actor-ref) → block until dead + + ;; Dead letter handler + set-dead-letter-handler! ;; (set-dead-letter-handler! proc) + + ;; Low-level: use the default scheduler or provide one + set-actor-scheduler! ;; (set-actor-scheduler! sched) + ) + (import (chezscheme) + (std actor mpsc) + (std actor scheduler)) + + ;; ========== Actor ID generation ========== + ;; Simple monotonic counter; fine for local actors. + + (define *next-actor-id* 0) + (define *actor-id-mutex* (make-mutex)) + + (define (next-actor-id!) + (mutex-acquire *actor-id-mutex*) + (let ([id *next-actor-id*]) + (set! *next-actor-id* (fx+ id 1)) + (mutex-release *actor-id-mutex*) + id)) + + ;; ========== Actor Record ========== + + (define-record-type actor-ref + (fields + (immutable id) ;; unique integer + (immutable node) ;; #f = local; string = remote node id + (immutable mailbox) ;; mpsc-queue + (mutable state) ;; 'idle | 'scheduled | 'running | 'dead + (mutable behavior) ;; current behavior: (lambda (msg) ...) + (mutable links) ;; list of actor-refs to notify on death + (mutable monitors) ;; list of (actor-ref . tag) to notify + (immutable name) ;; symbol or #f + (immutable done-mutex) + (immutable done-cond) ;; signaled when state = 'dead + (mutable exit-reason)) ;; 'normal | exception | 'killed + (protocol + (lambda (new) + (lambda (behavior name) + (new (next-actor-id!) + #f ;; local + (make-mpsc-queue) + 'idle + behavior + '() ;; links + '() ;; monitors + name + (make-mutex) + (make-condition) + #f)))) ;; exit-reason not set yet + (sealed #t)) + + ;; ========== Global actor table ========== + ;; Maps id → actor-ref for local lookups (used by supervisor and registry) + + (define *actor-table* (make-eq-hashtable)) + (define *actor-table-mutex* (make-mutex)) + + (define (register-local-actor! a) + (mutex-acquire *actor-table-mutex*) + (hashtable-set! *actor-table* (actor-ref-id a) a) + (mutex-release *actor-table-mutex*)) + + (define (unregister-local-actor! a) + (mutex-acquire *actor-table-mutex*) + (hashtable-delete! *actor-table* (actor-ref-id a)) + (mutex-release *actor-table-mutex*)) + + (define (lookup-local-actor id) + (mutex-acquire *actor-table-mutex*) + (let ([a (hashtable-ref *actor-table* id #f)]) + (mutex-release *actor-table-mutex*) + a)) + + ;; ========== Thread-local actor context ========== + + (define current-actor (make-thread-parameter #f)) + (define (self) (current-actor)) + (define (actor-id) (and (current-actor) (actor-ref-id (current-actor)))) + + ;; ========== Dead letter handler ========== + + (define *dead-letter-handler* + (lambda (msg dest) + ;; Default: log to stderr + (parameterize ([current-output-port (current-error-port)]) + (display "DEAD LETTER: actor #") + (display (actor-ref-id dest)) + (display " is dead, message dropped: ") + (write msg) + (newline)))) + + (define (set-dead-letter-handler! proc) + (set! *dead-letter-handler* proc)) + + ;; ========== Actor scheduler reference ========== + + (define *actor-scheduler* (make-parameter #f)) + (define (set-actor-scheduler! sched) (*actor-scheduler* sched)) + + ;; ========== Running an actor (internal) ========== + + ;; Process one message from the actor's mailbox. + ;; Called as a task on a worker thread. + (define (run-actor! a) + (parameterize ([current-actor a]) + (actor-ref-state-set! a 'running) + (let-values ([(msg ok) (mpsc-try-dequeue! (actor-ref-mailbox a))]) + (if ok + (begin + ;; Call the behavior with the message + (guard (exn [#t (actor-die! a exn)]) + ((actor-ref-behavior a) msg)) + ;; Check if more messages waiting + (if (not (mpsc-empty? (actor-ref-mailbox a))) + (schedule-actor! a) ;; re-submit + (actor-ref-state-set! a 'idle))) + ;; Spurious wake (shouldn't happen) — go idle + (actor-ref-state-set! a 'idle))))) + + ;; Schedule the actor to run on the scheduler + (define (schedule-actor! a) + (actor-ref-state-set! a 'scheduled) + (let ([sched (or (*actor-scheduler*) (default-scheduler))]) + (if sched + (scheduler-submit! sched (lambda () (run-actor! a))) + ;; No scheduler — fall back to fork-thread (1:1 mode) + (fork-thread (lambda () (run-actor! a)))))) + + ;; Handle actor death + (define (actor-die! a reason) + (actor-ref-state-set! a 'dead) + (actor-ref-exit-reason-set! a reason) + (unregister-local-actor! a) + (mpsc-close! (actor-ref-mailbox a)) + ;; Notify linked actors + (for-each + (lambda (linked) + (when (actor-alive? linked) + (send linked (list 'EXIT (actor-ref-id a) reason)))) + (actor-ref-links a)) + ;; Notify monitors + (for-each + (lambda (mon) + (let ([watcher (car mon)] + [tag (cdr mon)]) + (when (actor-alive? watcher) + (send watcher (list 'DOWN tag (actor-ref-id a) reason))))) + (actor-ref-monitors a)) + ;; Signal anyone waiting on actor-wait! + (mutex-acquire (actor-ref-done-mutex a)) + (condition-broadcast (actor-ref-done-cond a)) + (mutex-release (actor-ref-done-mutex a))) + + ;; ========== Public API ========== + + (define (spawn-actor behavior . rest) + (let* ([name (if (null? rest) #f (car rest))] + [a (make-actor-ref behavior name)]) + (register-local-actor! a) + ;; Submit initial run — actor starts processing immediately when first message arrives + ;; (Don't run until first message; actor is idle until then) + a)) + + (define (spawn-actor/linked behavior . rest) + (let ([parent (current-actor)] + [child (apply spawn-actor behavior rest)]) + (when parent + ;; Bidirectional link + (actor-ref-links-set! parent (cons child (actor-ref-links parent))) + (actor-ref-links-set! child (cons parent (actor-ref-links child)))) + child)) + + (define (send actor msg) + (cond + [(actor-ref? actor) + (if (actor-alive? actor) + (begin + (mpsc-enqueue! (actor-ref-mailbox actor) msg) + ;; Wake the actor if it's idle + (when (eq? (actor-ref-state actor) 'idle) + (schedule-actor! actor))) + ;; Actor is dead — deliver to dead letter handler + (*dead-letter-handler* msg actor))] + ;; Future: handle remote actor-refs here (Layer 7) + [else + (error 'send "not an actor-ref" actor)])) + + (define (send/timeout actor msg timeout-secs) + ;; For local actors, send is synchronous (fire-and-forget) — timeout doesn't apply. + ;; For remote actors, timeout applies to the network send. + (send actor msg)) + + (define (actor-alive? actor) + (not (eq? (actor-ref-state actor) 'dead))) + + (define (actor-kill! actor) + (actor-die! actor 'killed)) + + (define (actor-wait! actor) + (mutex-acquire (actor-ref-done-mutex actor)) + (let loop () + (unless (eq? (actor-ref-state actor) 'dead) + (condition-wait (actor-ref-done-cond actor) + (actor-ref-done-mutex actor)) + (loop))) + (mutex-release (actor-ref-done-mutex actor))) + + ) ;; end library +``` + +### Behavior Function Contract + +A behavior function receives one message at a time: + +```scheme +(define my-actor + (spawn-actor + (lambda (msg) + (match msg + [('ping reply-to) (send reply-to 'pong)] + [('stop) (actor-kill! (self))] + [_ (display "unknown message\n")])))) +``` + +The behavior function may call `(self)` to get its own actor-ref. +It must not block indefinitely — use `ask` (Layer 4) for request-reply patterns, +which suspends via a one-shot future rather than blocking. + +--- + +## Layer 4: Protocol System (`lib/std/actor/protocol.sls`) + +### Purpose + +Define typed message structs with constructor, predicate, and field accessors. +Generate typed send/receive helpers. This replaces Gerbil's `defmessage` + +`defcall-actor` pattern with a cleaner `defprotocol` macro. + +### `defprotocol` Macro + +```scheme +(defprotocol my-service + ;; Each clause: (message-name field ...) or (message-name field ... -> reply-type) + (ping) ;; no fields, no reply expected + (compute value -> result) ;; one field, reply expected + (shutdown reason)) ;; one field, no reply +``` + +Expands to: + +```scheme +;; Message structs +(define-record-type my-service:ping (fields) ...) +(define-record-type my-service:compute (fields value) ...) +(define-record-type my-service:result (fields value) ...) ;; reply type +(define-record-type my-service:shutdown (fields reason) ...) + +;; Constructors +(define (make-my-service:ping) ...) +(define (make-my-service:compute value) ...) +(define (make-my-service:result value) ...) +(define (make-my-service:shutdown reason) ...) + +;; Typed ask (returns future) +(define (my-service:compute! actor value) + (ask actor (make-my-service:compute value))) + +;; Typed tell (fire and forget) +(define (my-service:ping! actor) + (tell actor (make-my-service:ping))) +(define (my-service:shutdown! actor reason) + (tell actor (make-my-service:shutdown reason))) +``` + +### Full Implementation + +```scheme +#!chezscheme +(library (std actor protocol) + (export + defprotocol + + ;; Core ask/tell/call + ask ;; (ask actor-ref msg [timeout-secs]) → future + ask-sync ;; (ask-sync actor-ref msg [timeout-secs]) → value (blocks) + tell ;; (tell actor-ref msg) → void (fire and forget, alias for send) + call ;; (call actor-ref proc [timeout]) → value (RPC shorthand) + + ;; Reply inside a behavior + reply ;; (reply value) → void (must be in ask context) + reply-to ;; (reply-to) → actor-ref of requester or #f + + ;; One-shot reply channels + make-reply-channel + reply-channel? + reply-channel-get ;; blocks until reply + reply-channel-put! ;; sender puts reply + ) + (import (chezscheme) + (std actor core) + (std task)) ;; for futures + + ;; ========== Reply channels ========== + ;; A reply channel is a one-shot future: the asker creates it, + ;; sends it inside the message, and waits on it. + ;; The behavior calls (reply value) to complete it. + + (define-record-type reply-channel + (fields + (immutable future)) + (protocol + (lambda (new) + (lambda () (new (make-future))))) + (sealed #t)) + + (define (reply-channel-get rc) + (future-get (reply-channel-future rc))) + + (define (reply-channel-put! rc value) + (future-complete! (reply-channel-future rc) value)) + + ;; Thread-local: current reply channel (set by ask infrastructure) + (define current-reply-channel (make-thread-parameter #f)) + (define current-sender (make-thread-parameter #f)) + + (define (reply value) + (let ([rc (current-reply-channel)]) + (unless rc + (error 'reply "not in an ask context")) + (reply-channel-put! rc value))) + + (define (reply-to) + (current-sender)) + + ;; ========== ask ========== + ;; Sends msg to actor with an embedded reply channel. + ;; The message is wrapped in an envelope: ('ask reply-channel . original-msg) + ;; The behavior must call (reply value) to complete the future. + + (define (ask actor-ref msg . timeout-args) + (let* ([rc (make-reply-channel)] + [envelope (list 'ask rc msg)]) + (send actor-ref envelope) + (reply-channel-future rc))) ;; return future, caller calls future-get + + (define (ask-sync actor-ref msg . timeout-args) + (let ([fut (apply ask actor-ref msg timeout-args)]) + ;; TODO: apply timeout by racing future-get against a timer + (future-get fut))) + + ;; ========== tell ========== + ;; Simple alias for send — semantic distinction makes code clearer. + (define (tell actor-ref msg) + (send actor-ref msg)) + + ;; ========== call ========== + ;; Sends a lambda to the actor for remote execution. + ;; Useful for actors that expose a mutable state model. + (define (call actor-ref proc . timeout-args) + (apply ask-sync actor-ref (list 'call proc) timeout-args)) + + ;; ========== ask envelope unwrapping ========== + ;; Actors that want to support ask must call this in their behavior: + ;; + ;; (define (my-behavior msg) + ;; (with-ask-context msg + ;; (lambda (actual-msg) + ;; (match actual-msg + ;; [('compute n) (reply (* n n))] + ;; ...)))) + ;; + ;; OR use defprotocol which generates the dispatch automatically. + + (define-syntax with-ask-context + (syntax-rules () + [(_ msg body-thunk) + (if (and (pair? msg) (eq? (car msg) 'ask)) + (let ([rc (cadr msg)] + [actual (caddr msg)]) + (parameterize ([current-reply-channel rc] + [current-sender (self)]) + (body-thunk actual))) + (body-thunk msg))])) + + ;; ========== defprotocol macro ========== + + (define-syntax defprotocol + (lambda (stx) + (syntax-case stx (->) + [(_ proto-name clause ...) + (let* ([proto (syntax->datum #'proto-name)] + [prefix (symbol->string proto)]) + (define (sym . parts) + (string->symbol (apply string-append (map (lambda (p) + (cond [(symbol? p) (symbol->string p)] + [(string? p) p] + [else (error 'defprotocol "bad part" p)])) parts)))) + (define clauses (syntax->datum #'(clause ...)))