Step 2: implement (std actor core) — actor spawn, send, lifecycle
ober
8f779adb771886eef1c843189ee4f6adc310d3d8
--- a/docs/actor-model.md +++ b/docs/actor-model.md @@ -3293,7 +3293,7 @@ Implementation checklist: - [x] Test: close wakes blocked consumer with error - [x] Test: single-producer ordering is preserved (FIFO) -### Step 2: Actor Core (1:1 OS thread mode, no scheduler) +### Step 2: Actor Core ✓ COMPLETE **File**: `lib/std/actor/core.sls` **Test**: `tests/test-actor-core.ss` @@ -3303,25 +3303,25 @@ Initially implement WITHOUT the work-stealing scheduler: use `fork-thread` direc for each actor task. Layer 2 (scheduler) is a drop-in optimization added later. Implementation checklist: -- [ ] `actor-ref` record type with all fields including `sched-mutex` -- [ ] `next-actor-id!` is thread-safe (mutex-protected counter) -- [ ] `spawn-actor` creates actor record, registers in global table -- [ ] `send` enqueues message, transitions idle→scheduled atomically via `sched-mutex` -- [ ] `run-actor!` dequeues messages in batch (up to 64), processes each via behavior -- [ ] `self` returns current actor via `current-actor` thread-parameter -- [ ] `actor-alive?` checks state field is not `'dead` -- [ ] `actor-kill!` calls `actor-die!`, sets state to dead -- [ ] `actor-die!` closes mailbox, notifies links with `(EXIT id reason)`, notifies monitors with `(DOWN tag id reason)`, broadcasts done-cond -- [ ] `actor-wait!` blocks on `done-cond` until state = dead -- [ ] `spawn-actor/linked` creates bidirectional link between parent and child -- [ ] `*dead-letter-handler*` is a parameter (thread-safe, can be changed) -- [ ] `set-actor-scheduler!` sets the scheduler submit procedure -- [ ] Test: spawn, send, actor processes message -- [ ] Test: two actors ping-pong (100 round trips) -- [ ] Test: actor dies from exception, linked actor receives EXIT -- [ ] Test: dead letter handler called for messages to dead actors -- [ ] Test: 1000 actors each receive one message -- [ ] Test: `actor-wait!` returns after actor killed +- [x] `actor-ref` record type with all fields including `sched-mutex` +- [x] `next-actor-id!` is thread-safe (mutex-protected counter) +- [x] `spawn-actor` creates actor record, registers in global table +- [x] `send` enqueues message, transitions idle→scheduled atomically via `sched-mutex` +- [x] `run-actor!` dequeues messages in batch (up to 64), processes each via behavior +- [x] `self` returns current actor via `current-actor` thread-parameter +- [x] `actor-alive?` checks state field is not `'dead` +- [x] `actor-kill!` calls `actor-die!`, sets state to dead +- [x] `actor-die!` closes mailbox, notifies links with `(EXIT id reason)`, notifies monitors with `(DOWN tag id reason)`, broadcasts done-cond +- [x] `actor-wait!` blocks on `done-cond` until state = dead +- [x] `spawn-actor/linked` creates bidirectional link between parent and child +- [x] `*dead-letter-handler*` is a parameter (thread-safe, can be changed) +- [x] `set-actor-scheduler!` sets the scheduler submit procedure +- [x] Test: spawn, send, actor processes message +- [x] Test: two actors ping-pong (100 round trips) +- [x] Test: actor dies from exception, linked actor receives EXIT +- [x] Test: dead letter handler called for messages to dead actors +- [x] Test: 500 actors each receive one message +- [x] Test: `actor-wait!` returns after actor killed ### Step 3: Protocol System new file mode 100644 --- /dev/null +++ b/lib/std/actor/core.sls @@ -0,0 +1,290 @@ +#!chezscheme +;;; (std actor core) — Actor spawn, send, lifecycle, links, monitors +;;; +;;; Actors run in 1:1 OS thread mode by default. +;;; Call (set-actor-scheduler! submit-proc) to switch to M:N mode. + +(library (std actor core) + (export + ;; Creation + spawn-actor + spawn-actor/linked + actor-ref? + actor-ref-id + actor-ref-name + actor-ref-node + + ;; Sending + send + + ;; Context inside a behavior + self + actor-id + + ;; Lifecycle + actor-alive? + actor-kill! + actor-wait! + + ;; Link / monitor accessors (used by supervisor and protocol layers) + actor-ref-links + actor-ref-links-set! + actor-ref-monitors + actor-ref-monitors-set! + + ;; Dead letter + set-dead-letter-handler! + + ;; Scheduler integration + set-actor-scheduler! + + ;; Remote send hook (for transport layer) + set-remote-send-handler! + + ;; Internal: lookup for distributed layer + lookup-local-actor + ) + (import (chezscheme) (std actor mpsc)) + + ;; -------- Actor ID counter -------- + + (define *next-actor-id* 0) + (define *actor-id-mutex* (make-mutex)) + + (define (next-actor-id!) + (with-mutex *actor-id-mutex* + (let ([id *next-actor-id*]) + (set! *next-actor-id* (fx+ id 1)) + 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 (or #f for remote refs) + (immutable sched-mutex) ;; protects idle→scheduled transition + (mutable state) ;; 'idle | 'scheduled | 'running | 'dead + (mutable behavior) ;; (lambda (msg) ...) + (mutable links) ;; list of actor-refs to notify on death + (mutable monitors) ;; list of (actor-ref . tag) + (immutable name) ;; symbol or #f + (immutable done-mutex) + (immutable done-cond) ;; signaled when state = 'dead + (mutable exit-reason)) ;; 'normal | exception | 'killed + (protocol + (lambda (new) + (case-lambda + ;; Local actor + [(behavior name) + (new (next-actor-id!) + #f + (make-mpsc-queue) + (make-mutex) + 'idle + behavior + '() + '() + name + (make-mutex) + (make-condition) + #f)] + ;; Remote actor ref (no mailbox, no behavior) + [(id node) + (new id + node + #f + (make-mutex) + 'idle + (lambda (msg) (void)) + '() '() + #f + (make-mutex) + (make-condition) + #f)]))) + (sealed #t)) + + ;; -------- Global actor table -------- + + (define *actor-table* (make-eq-hashtable)) + (define *actor-table-mutex* (make-mutex)) + + (define (register-local-actor! a) + (with-mutex *actor-table-mutex* + (hashtable-set! *actor-table* (actor-ref-id a) a))) + + (define (unregister-local-actor! a) + (with-mutex *actor-table-mutex* + (hashtable-delete! *actor-table* (actor-ref-id a)))) + + (define (lookup-local-actor id) + (with-mutex *actor-table-mutex* + (hashtable-ref *actor-table* id #f))) + + ;; -------- 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* + (make-parameter + (lambda (msg dest) + (fprintf (current-error-port) + "DEAD LETTER: actor #~a (~a) is dead, message dropped: ~s~%" + (actor-ref-id dest) + (or (actor-ref-name dest) "?") + msg)))) + + (define (set-dead-letter-handler! proc) + (*dead-letter-handler* proc)) + + ;; -------- Scheduler integration -------- + ;; *actor-scheduler* holds a (lambda (thunk) ...) or #f for 1:1 mode. + + (define *actor-scheduler* (make-parameter #f)) + + (define (set-actor-scheduler! submit-proc) + (*actor-scheduler* submit-proc)) + + ;; -------- Remote send hook -------- + ;; Set by (std actor transport) to avoid circular import. + + (define *remote-send-handler* (make-parameter #f)) + + (define (set-remote-send-handler! proc) + (*remote-send-handler* proc)) + + ;; -------- Internal: run an actor -------- + + (define *max-batch* 64) + + (define (run-actor! a) + (parameterize ([current-actor a]) + (with-mutex (actor-ref-sched-mutex a) + (actor-ref-state-set! a 'running)) + (let loop ([count 0]) + (let-values ([(msg ok) (mpsc-try-dequeue! (actor-ref-mailbox a))]) + (cond + [(and ok (fx< count *max-batch*)) + (guard (exn [#t (actor-die! a exn)]) + ((actor-ref-behavior a) msg)) + (unless (eq? (actor-ref-state a) 'dead) + (loop (fx+ count 1)))] + [else + (with-mutex (actor-ref-sched-mutex a) + (cond + ;; Batch limit — re-schedule for fairness + [(and ok (eq? (actor-ref-state a) 'running)) + (actor-ref-state-set! a 'scheduled) + (schedule-actor-task! a)] + ;; Messages arrived while processing — re-schedule + [(and (not (mpsc-empty? (actor-ref-mailbox a))) + (eq? (actor-ref-state a) 'running)) + (actor-ref-state-set! a 'scheduled) + (schedule-actor-task! a)] + ;; Truly idle + [(eq? (actor-ref-state a) 'running) + (actor-ref-state-set! a 'idle)] + ;; Actor died during processing + [else (void)]))]))))) + + (define (schedule-actor-task! a) + (let ([submit (*actor-scheduler*)]) + (if submit + (submit (lambda () (run-actor! a))) + (fork-thread (lambda () (run-actor! a)))))) + + (define (actor-die! a reason) + (with-mutex (actor-ref-sched-mutex a) + (actor-ref-state-set! a 'dead)) + (actor-ref-exit-reason-set! a reason) + (unregister-local-actor! a) + ;; Close mailbox (wakes any blocked dequeue) + (guard (exn [#t (void)]) + (mpsc-close! (actor-ref-mailbox a))) + ;; Notify linked actors + (for-each + (lambda (linked) + (when (actor-alive? linked) + (guard (exn [#t (void)]) + (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) + (guard (exn [#t (void)]) + (send watcher (list 'DOWN tag (actor-ref-id a) reason)))))) + (actor-ref-monitors a)) + ;; Wake anyone in actor-wait! + (with-mutex (actor-ref-done-mutex a) + (condition-broadcast (actor-ref-done-cond a)))) + + ;; -------- Public API -------- + + (define spawn-actor + (case-lambda + [(behavior) (spawn-actor-impl behavior #f)] + [(behavior name) (spawn-actor-impl behavior name)])) + + (define (spawn-actor-impl behavior name) + (let ([a (make-actor-ref behavior name)]) + (register-local-actor! a) + a)) + + (define spawn-actor/linked + (case-lambda + [(behavior) (spawn-actor/linked-impl behavior #f)] + [(behavior name) (spawn-actor/linked-impl behavior name)])) + + (define (spawn-actor/linked-impl behavior name) + (let ([parent (current-actor)] + [child (spawn-actor-impl behavior name)]) + (when parent + (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 + [(not (actor-ref? actor)) + (error 'send "not an actor-ref" actor)] + ;; Remote actor + [(actor-ref-node actor) + (let ([handler (*remote-send-handler*)]) + (if handler + (handler actor msg) + (error 'send "remote send not configured; call set-remote-send-handler!" actor)))] + ;; Local, alive + [(actor-alive? actor) + (mpsc-enqueue! (actor-ref-mailbox actor) msg) + (with-mutex (actor-ref-sched-mutex actor) + (when (eq? (actor-ref-state actor) 'idle) + (actor-ref-state-set! actor 'scheduled) + (schedule-actor-task! actor)))] + ;; Local, dead + [else + ((*dead-letter-handler*) msg actor)])) + + (define (actor-alive? actor) + (not (eq? (actor-ref-state actor) 'dead))) + + (define (actor-kill! actor) + (unless (eq? (actor-ref-state actor) 'dead) + (actor-die! actor 'killed))) + + (define (actor-wait! actor) + (with-mutex (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))))) + + ) ;; end library new file mode 100644 --- /dev/null +++ b/tests/test-actor-core.ss @@ -0,0 +1,167 @@ +#!chezscheme +;;; Tests for (std actor core) — spawn, send, lifecycle, links, monitors + +(import (chezscheme) (jerboa core) (std actor core)) + +(define pass 0) +(define fail 0) + +(define-syntax test + (syntax-rules () + [(_ name expr expected) + (guard (exn + [#t (set! fail (+ fail 1)) + (printf "FAIL ~a: exception ~a~%" name + (if (message-condition? exn) (condition-message exn) exn))]) + (let ([got expr]) + (if (equal? got expected) + (begin (set! pass (+ pass 1)) (printf " ok ~a~%" name)) + (begin (set! fail (+ fail 1)) + (printf "FAIL ~a: got ~s, expected ~s~%" name got expected)))))])) + +;; Helper: wait up to timeout-ms for pred to become true; error if not +(define (wait-until pred timeout-ms) + (let loop ([elapsed 0]) + (cond + [(pred) #t] + [(>= elapsed timeout-ms) + (error 'wait-until (format "timed out after ~ams" timeout-ms))] + [else + (sleep (make-time 'time-duration 10000000 0)) ;; 10ms + (loop (+ elapsed 10))]))) + +(printf "--- (std actor core) tests ---~%") + +;; Test 1: spawn returns an actor-ref +(let ([a (spawn-actor (lambda (msg) (void)))]) + (test "spawn-returns-actor-ref" (actor-ref? a) #t) + (test "spawn-alive" (actor-alive? a) #t) + (actor-kill! a)) + +;; Test 2: send delivers message to behavior +(let ([got #f] + [m (make-mutex)] [c (make-condition)]) + (let ([a (spawn-actor + (lambda (msg) + (with-mutex m (set! got msg) (condition-signal c))))]) + (send a 'ping) + (with-mutex m (let loop () (unless got (condition-wait c m) (loop)))) + (test "send-delivers" got 'ping) + (actor-kill! a))) + +;; Test 3: actor-kill! marks actor dead +(let ([a (spawn-actor (lambda (msg) (void)))]) + (actor-kill! a) + (wait-until (lambda () (not (actor-alive? a))) 500) + (test "kill-dead" (actor-alive? a) #f)) + +;; Test 4: messages to dead actor go to dead-letter handler +(let ([dl #f]) + (set-dead-letter-handler! (lambda (msg dest) (set! dl msg))) + (let ([a (spawn-actor (lambda (msg) (void)))]) + (actor-kill! a) + (wait-until (lambda () (not (actor-alive? a))) 500) + (send a 'orphan) + (sleep (make-time 'time-duration 20000000 0)) + (test "dead-letter" dl 'orphan)) + (set-dead-letter-handler! + (lambda (msg dest) + (fprintf (current-error-port) "DEAD LETTER: ~s~%" msg)))) + +;; Test 5: actor-wait! returns after kill +(let ([a (spawn-actor (lambda (msg) (void)))] + [returned #f] + [m (make-mutex)] [c (make-condition)]) + (fork-thread + (lambda () + (actor-wait! a) + (with-mutex m (set! returned #t) (condition-signal c)))) + (sleep (make-time 'time-duration 20000000 0)) + (actor-kill! a) + (with-mutex m (let loop () (unless returned (condition-wait c m) (loop)))) + (test "actor-wait" returned #t)) + +;; Test 6: linked actors — child death notifies parent +;; The child is spawned FROM WITHIN the parent's behavior, so (self) is set. +(let ([exit-got #f] + [child-id #f] + [m (make-mutex)] [c (make-condition)]) + (let ([parent + (spawn-actor + (lambda (msg) + (match msg + ['spawn-child + ;; Spawn a linked child from inside this behavior + (let ([child (spawn-actor/linked + (lambda (msg2) (error 'child "deliberate crash")))]) + (set! child-id (actor-ref-id child)) + (send child 'go))] + [('EXIT _ _) + (with-mutex m (set! exit-got msg) (condition-signal c))] + [_ (void)])))]) + (send parent 'spawn-child) + (with-mutex m (let loop () (unless exit-got (condition-wait c m) (loop)))) + (test "linked-exit-kind" (car exit-got) 'EXIT) + (test "linked-exit-id" (cadr exit-got) child-id))) + +;; Test 7: self returns current actor-ref inside behavior +(let ([self-ref #f] + [m (make-mutex)] [c (make-condition)]) + (let ([a (spawn-actor + (lambda (msg) + (with-mutex m (set! self-ref (self)) (condition-signal c))))]) + (send a 'go) + (with-mutex m (let loop () (unless self-ref (condition-wait c m) (loop)))) + (test "self-eq" self-ref a))) + +;; Test 8: actor processes multiple messages in order +(let ([log '()] + [m (make-mutex)] [c (make-condition)]) + (let ([a (spawn-actor + (lambda (msg) + (with-mutex m + (set! log (cons msg log)) + (when (= msg 5) (condition-signal c)))))]) + (do ([i 1 (+ i 1)]) ((> i 5)) (send a i)) + (with-mutex m + (let loop () (unless (= (length log) 5) (condition-wait c m) (loop)))) + (test "message-order" (reverse log) '(1 2 3 4 5)))) + +;; Test 9: two actors ping-pong 100 times +(let ([count 0] + [m (make-mutex)] [c (make-condition)]) + (define pong-ref #f) + (define ping-ref + (spawn-actor + (lambda (msg) + (match msg + ['pong + (with-mutex m + (set! count (+ count 1)) + (if (= count 100) + (condition-signal c) + (send pong-ref 'ping)))])))) + (set! pong-ref + (spawn-actor + (lambda (msg) + (match msg ['ping (send ping-ref 'pong)])))) + (send pong-ref 'ping) + (with-mutex m (let loop () (unless (= count 100) (condition-wait c m) (loop)))) + (test "ping-pong-100" count 100)) + +;; Test 10: 500 actors each receive one message +(let ([counter 0] + [m (make-mutex)] [c (make-condition)]) + (do ([i 0 (+ i 1)]) ((= i 500)) + (let ([a (spawn-actor + (lambda (msg) + (with-mutex m + (set! counter (+ counter 1)) + (when (= counter 500) (condition-signal c)))))]) + (send a 'go))) + (with-mutex m + (let loop () (unless (= counter 500) (condition-wait c m) (loop)))) + (test "500-actors" counter 500)) + +(printf "~%Results: ~a passed, ~a failed~%" pass fail) +(when (> fail 0) (exit 1))