Rewrite actor model guide with validated Chez 10.4 primitives
ober
9cec29b1b1cfff8b394d87cdd0c589c47bb1e0a2
--- a/docs/actor-model.md +++ b/docs/actor-model.md @@ -1,8 +1,11 @@ # 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. +actor system on Chez Scheme 10.4+ (threaded build, `ta6le`). Each layer is independently +implementable and testable. A lesser model can implement this by following the layers in +order — do not skip ahead. + +**Validated against**: Chez Scheme 10.4.0, threaded build, Linux x86_64. --- @@ -18,8 +21,9 @@ A lesser model can implement this by following the layers in order — do not sk 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. + distributed transport. Any Scheme value (records, vectors, bytevectors, symbols, + numbers, booleans, pairs) 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. @@ -30,8 +34,8 @@ A lesser model can implement this by following the layers in order — do not sk 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 +7. **Gradual complexity** — Layers 3-6 (local actors + supervision) are useful + without Layer 2 (work-stealing) or Layer 7 (distributed). Build and ship each layer independently. **Non-goals**: @@ -41,12 +45,92 @@ A lesser model can implement this by following the layers in order — do not sk --- +## Chez Scheme Primitives Reference + +Every primitive used in this guide. Know these before implementing. + +### Threading (Chez 10, threaded build) +| Primitive | Description | +|-----------|-------------| +| `(fork-thread thunk)` | Starts a new OS thread immediately, returns thread-id | +| `(get-thread-id)` | Returns current thread's integer id | +| `(make-mutex)` | Creates a new mutex | +| `(mutex-acquire mutex)` | Blocks until mutex acquired | +| `(mutex-release mutex)` | Releases a mutex | +| `(with-mutex mutex body ...)` | Acquires, evaluates body, releases (even on exception) | +| `(make-condition)` | Creates a condition variable | +| `(condition-wait cond mutex)` | Atomically releases mutex and blocks on condition; re-acquires mutex on wake | +| `(condition-wait cond mutex time)` | Same but with timeout; returns `#f` on timeout, `#t` if signaled | +| `(condition-signal cond)` | Wakes one waiting thread | +| `(condition-broadcast cond)` | Wakes all waiting threads | +| `(make-thread-parameter default)` | Creates a thread-local parameter (SMP-safe, no global lock) | + +### Data Structures +| Primitive | Description | +|-----------|-------------| +| `(make-vector n init)` | Creates a vector of size n | +| `(make-eq-hashtable)` | Creates a hashtable with `eq?` comparison | +| `(make-hashtable hash equiv)` | Creates a hashtable with custom hash/equiv | +| `(hashtable-set! ht key val)` | Insert/update | +| `(hashtable-ref ht key default)` | Lookup with default | +| `(hashtable-delete! ht key)` | Remove | +| `(hashtable-keys ht)` | Returns vector of keys | + +### Serialization +| Primitive | Description | +|-----------|-------------| +| `(fasl-write obj port)` | Serializes any Scheme value (records, vectors, etc.) to binary port | +| `(fasl-read port)` | Deserializes from binary port | +| `(open-bytevector-output-port)` | Returns `(values port get-bytevector-proc)` | +| `(open-bytevector-input-port bv)` | Creates input port from bytevector | + +### Time +| Primitive | Description | +|-----------|-------------| +| `(make-time type nanoseconds seconds)` | Creates a time object. Use `'time-duration` for durations | +| `(current-time)` | Returns `time-utc` object | +| `(time-second t)` | Extracts seconds from a time object | +| `(time-nanosecond t)` | Extracts nanoseconds from a time object | + +**WARNING**: `time->seconds` does NOT exist in Chez. It is in `(std srfi srfi-19)`. +To convert a time object to a float without SRFI-19: +```scheme +(define (time->float t) + (+ (time-second t) (/ (time-nanosecond t) 1000000000.0))) +``` + +### Other +| Primitive | Description | +|-----------|-------------| +| `(random n)` | Returns random integer in [0, n) | +| `(filter pred lst)` | Standard R6RS filter | +| `(define-record-type ...)` | R6RS record types with protocol, sealed, etc. | +| `(guard (var [test expr] ...) body ...)` | R6RS exception handling | + +**WARNING**: `match` is NOT built-in Chez. Import from `(jerboa core)`. +**WARNING**: `cpu-count` does NOT exist in Chez. Read from `/proc` or use a constant: +```scheme +(define (cpu-count) + (or (let ([p (open-input-file "/proc/cpuinfo")]) + (let loop ([n 0]) + (let ([line (get-line p)]) + (cond + [(eof-object? line) (close-port p) n] + [(and (>= (string-length line) 9) + (string=? (substring line 0 9) "processor")) + (loop (fx+ n 1))] + [else (loop n)])))) + 4)) ;; fallback +``` + +--- + ## Architecture Overview ``` ┌──────────────────────────────────────────────────────┐ │ Layer 7: Distributed Transport │ -│ lib/std/actor/transport.sls, remote.sls, node.sls │ +│ lib/std/actor/transport.sls │ │ TCP+TLS, fasl serialization, location transparency │ ├──────────────────────────────────────────────────────┤ │ Layer 6: Registry │ @@ -77,7 +161,8 @@ A lesser model can implement this by following the layers in order — do not sk │ (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 │ +│ (std net ssl) — TCP+TLS via chez-ssl │ +│ (jerboa core) — match, def, defstruct │ └──────────────────────────────────────────────────────┘ ``` @@ -94,14 +179,25 @@ 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 +### Data Structure: Two-Lock 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. +We use a Michael-Scott style 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. 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). +(which would require `compare-and-swap` via FFI C shims — Chez does not expose CAS natively). + +### Critical Design: Signaling Without Deadlock + +The original design had a subtle deadlock risk: signaling `not-empty` while holding +`tail-mutex`. If the consumer holds `head-mutex` and tries to signal or if the +producer tries to acquire `head-mutex` while holding `tail-mutex`, you can deadlock +when another thread does the reverse. + +**Solution**: Use a single condition variable protected by `head-mutex` only. The +producer signals by acquiring `head-mutex` briefly AFTER releasing `tail-mutex`. +This ensures no nested lock acquisition. ```scheme #!chezscheme @@ -111,8 +207,8 @@ This is simpler and more practical for Chez than a lock-free Michael-Scott 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-try-dequeue! ;; consumer: remove or return (values #f #f) immediately + mpsc-empty? ;; peek (approximate — only safe from consumer thread) mpsc-close! ;; signal no more messages mpsc-closed?) (import (chezscheme)) @@ -131,11 +227,10 @@ This is simpler and more practical for Chez than a lock-free Michael-Scott 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 head-mutex) ;; consumer lock (also protects condition variable) (immutable tail-mutex) ;; producer lock (immutable not-empty) ;; condition: signaled when item enqueued - (mutable closed?) - (mutable count)) ;; approximate item count + (mutable closed?)) (protocol (lambda (new) (lambda () @@ -143,85 +238,81 @@ This is simpler and more practical for Chez than a lock-free Michael-Scott queue (new dummy dummy (make-mutex) (make-mutex) (make-condition) - #f 0))))) + #f))))) (sealed #t)) ;; Producer: enqueue a value - ;; Lock only the tail — does not interfere with consumer reading head + ;; Lock only the tail — does not interfere with consumer reading head. + ;; Signal the consumer AFTER releasing tail-mutex to avoid nested locking. (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)))) + (with-mutex (mpsc-queue-tail-mutex q) + (when (mpsc-queue-closed? q) + (error 'mpsc-enqueue! "queue is closed")) + (mpsc-node-next-set! (mpsc-queue-tail q) node) + (mpsc-queue-tail-set! q node)) + ;; Signal consumer OUTSIDE tail-lock (head-mutex acquired briefly) + (with-mutex (mpsc-queue-head-mutex q) + (condition-signal (mpsc-queue-not-empty q))))) ;; Consumer: dequeue, blocking if empty (define (mpsc-dequeue! q) - (mutex-acquire (mpsc-queue-head-mutex q)) - (let loop () + (with-mutex (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 + val)] + [(mpsc-queue-closed? 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) + (with-mutex (mpsc-queue-head-mutex q) (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")] + (mpsc-node-value-set! next 'sentinel) + (values val #t))] [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)]))) + (values #f #f)])))) (define (mpsc-empty? q) + ;; Approximate: safe only from the consumer thread. + ;; Reads head.next without lock — may see stale data from producers. (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))) + (with-mutex (mpsc-queue-tail-mutex q) + (mpsc-queue-closed?-set! q #t)) + ;; Wake all blocked consumers + (with-mutex (mpsc-queue-head-mutex q) + (condition-broadcast (mpsc-queue-not-empty q)))) ) ;; end library ``` +**Why `with-mutex` instead of manual acquire/release**: `with-mutex` is a Chez +built-in that uses `dynamic-wind` to guarantee the mutex is released even if an +exception occurs inside the body. Manual acquire/release leaks the lock on exception. + **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` +- `try-dequeue` on empty queue returns `(values #f #f)` - Close while consumer is blocked — consumer gets error -- Count is approximately correct after concurrent operations +- Ordering: messages from a single producer arrive in FIFO order --- @@ -249,13 +340,22 @@ is fast enough when not under heavy contention. #include <stdint.h> // Returns 1 if swap succeeded, 0 if not -int cas_int64(int64_t *ptr, int64_t expected, int64_t desired) { +int jerboa_cas_int64(int64_t *ptr, int64_t expected, int64_t desired) { return atomic_compare_exchange_strong( - (atomic_int_fast64_t*)ptr, &expected, desired); + (_Atomic int64_t*)ptr, &expected, desired); +} + +void jerboa_atomic_store_int64(int64_t *ptr, int64_t val) { + atomic_store((_Atomic int64_t*)ptr, val); +} + +int64_t jerboa_atomic_load_int64(int64_t *ptr) { + return atomic_load((_Atomic int64_t*)ptr); } ``` -Compile as `libjerboa-atomic.so` and load via `(load-shared-object ...)`. -Then use `(foreign-procedure "cas_int64" (void* integer-64 integer-64) integer-32)`. +Compile: `gcc -shared -fPIC -O2 -o libjerboa-atomic.so support/atomic.c` +Load: `(load-shared-object "./libjerboa-atomic.so")` +Use: `(define cas-int64 (foreign-procedure "jerboa_cas_int64" (void* integer-64 integer-64) int))` 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. @@ -289,14 +389,16 @@ replacement at the deque level — the scheduler above does not change. (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))) + (with-mutex (work-deque-mutex 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))) + (with-mutex (work-deque-mutex d) + (fx<= (work-deque-bottom d) (work-deque-top d)))) - ;; Grow buffer when full + ;; Grow buffer when full (called under lock) (define (deque-grow! d) (let* ([old (work-deque-buf d)] [old-cap (vector-length old)] @@ -312,55 +414,54 @@ replacement at the deque level — the scheduler above does not change. ;; 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))) + (with-mutex (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))))) ;; 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))) + (with-mutex (work-deque-mutex d) + (let ([b (work-deque-bottom d)] + [t (work-deque-top d)]) + (if (fx> b t) + (let ([new-b (fx- b 1)]) + (work-deque-bottom-set! d new-b) + (let ([task (vector-ref (work-deque-buf d) + (fxmod new-b (deque-capacity d)))]) + (vector-set! (work-deque-buf d) (fxmod new-b (deque-capacity d)) #f) + task)) + #f)))) ;; 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))]))) + (with-mutex (work-deque-mutex d) + (let ([t (work-deque-top d)] + [b (work-deque-bottom d)]) + (cond + [(fx>= t b) + (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)) + (values task #t))])))) ) ;; end library ``` +**Test file**: `tests/test-actor-deque.ss` +- Push 1000 items, pop all — verify LIFO order +- Push 1000 items, steal all — verify FIFO order +- Concurrent push + steal from different threads +- Grow: push more than initial capacity (64) +- Empty deque: pop returns #f, steal returns (values #f #f) + --- ## Layer 2: Work-Stealing Scheduler (`lib/std/actor/scheduler.sls`) @@ -368,8 +469,8 @@ replacement at the deque level — the scheduler above does not change. ### 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). +maintains a fixed pool of N OS threads and schedules lightweight tasks across them. +M tasks run on N threads (M >> N). ### Key Insight @@ -377,12 +478,21 @@ Actors are NOT OS threads. An actor is a record with a mailbox. When a message a 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. +### How Workers Find Work + +Each worker follows this priority: +1. **Pop own deque** (LIFO — hot cache, best locality) +2. **Steal from random other worker** (FIFO — cold tasks migrate to idle cores) +3. **Sleep on condition variable** (avoid busy-wait; woken when new task submitted) + ### Data Structures ```scheme #!chezscheme (library (std actor scheduler) (export + make-scheduler + scheduler? scheduler-start! ;; create and start the thread pool scheduler-stop! ;; drain and shut down scheduler-submit! ;; submit a thunk as a task @@ -391,7 +501,7 @@ The task runs on whatever worker thread picks it up. This is the M:N model. default-scheduler) (import (chezscheme) (std actor deque)) - ;; A task is just a thunk (zero-argument procedure) + ;; 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) @@ -399,19 +509,17 @@ The task runs on whatever worker thread picks it up. This is the M:N model. (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)))) + (new id (make-work-deque) #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?)) @@ -421,7 +529,6 @@ The task runs on whatever worker thread picks it up. This is the M:N model. (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)))) @@ -432,34 +539,31 @@ The task runs on whatever worker thread picks it up. This is the M:N model. (define current-scheduler (make-thread-parameter #f)) (define default-scheduler (make-parameter #f)) - ;; Submit a task to the scheduler + ;; 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. + ;; Otherwise, distribute to a random worker deque. (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 + (deque-push-bottom! (worker-deque w) thunk) + ;; Slow path: external submission — pick a random worker (let* ([workers (scheduler-workers sched)] [n (vector-length workers)] - [idx (fxmod (random n) n)] ;; randomized for load balance + [idx (random n)] [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)))))) + (deque-push-bottom! (worker-deque w) thunk)))) + ;; Wake one sleeping worker + (with-mutex (scheduler-mutex sched) + (condition-signal (scheduler-work-available 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))]) + [n (vector-length (scheduler-workers sched))] + [my-id (worker-id w)]) (let loop () (when (scheduler-running? sched) ;; 1. Try own deque first @@ -469,17 +573,20 @@ The task runs on whatever worker thread picks it up. This is the M:N model. (guard (exn [#t (void)]) ;; tasks must not crash the worker (task)) (loop)) - ;; 2. Try stealing from a random other worker + ;; 2. Try stealing from other workers (round-robin from own id) (let try-steal ([attempts 0]) - (if (fx= attempts n) + (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)) + ;; Re-check before sleeping (avoid lost wakeup) + (when (and (scheduler-running? sched) + (not (deque-pop-bottom! (worker-deque w)))) + (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)] + (let* ([victim-idx (fxmod (fx+ my-id attempts 1) n)] [victim (vector-ref workers victim-idx)]) (let-values ([(task ok) (deque-steal-top! (worker-deque victim))]) (if ok @@ -492,6 +599,7 @@ The task runs on whatever worker thread picks it up. This is the M:N model. (define (scheduler-worker-count sched) (vector-length (scheduler-workers sched))) + ;; Start the scheduler: fork N worker threads (define (scheduler-start! sched) (scheduler-running?-set! sched #t) (let ([workers (scheduler-workers sched)]) @@ -501,11 +609,11 @@ The task runs on whatever worker thread picks it up. This is the M:N model. (fork-thread (lambda () (worker-run! sched w)))))) sched) + ;; Stop the scheduler: signal all workers to exit (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))) + (with-mutex (scheduler-mutex sched) + (condition-broadcast (scheduler-work-available sched)))) ) ;; end library ``` @@ -517,10 +625,26 @@ The task runs on whatever worker thread picks it up. This is the M:N model. 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. +- The sleep-before-wait pattern (check own deque after acquiring mutex but before + `condition-wait`) prevents the lost-wakeup bug where a task is submitted between + the failed steal attempts and the `condition-wait`. ### Initialization ```scheme +;; Helper: read CPU count from /proc on Linux +(define (cpu-count) + (guard (exn [#t 4]) + (let ([p (open-input-file "/proc/cpuinfo")]) + (let loop ([n 0]) + (let ([line (get-line p)]) + (cond + [(eof-object? line) (close-port p) (fxmax n 1)] + [(and (fx>= (string-length line) 9) + (string=? (substring line 0 9) "processor")) + (loop (fx+ n 1))] + [else (loop n)])))))) + ;; Typically done once at program start: (define sched (scheduler-start! (make-scheduler (cpu-count)))) (default-scheduler sched) @@ -534,7 +658,7 @@ The task runs on whatever worker thread picks it up. This is the M:N model. - 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. +- **Spawning** creates the actor record and registers it in the global table. - **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 @@ -558,15 +682,26 @@ The task runs on whatever worker thread picks it up. This is the M:N model. ### 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**: +blocks, in our M:N model an actor runs as a **task per message batch**: 1. Message arrives in mailbox -2. A task is submitted to the scheduler: `(lambda () (run-actor! actor))` +2. If actor is `idle`, 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 more messages in mailbox, processes the next one immediately (batching) - If mailbox empty, marks actor as IDLE + - On exception, marks actor as DEAD and notifies links/monitors + +### Race Condition: The IDLE→SCHEDULED Transition + +A critical race exists between checking `idle?` and scheduling. Two threads could +both see `idle` and double-schedule the actor. We solve this with a state mutex: + +``` +Thread A: send msg → enqueue → check state → (if idle) → set scheduled → submit task +Thread B: send msg → enqueue → check state → (if idle) → set scheduled → submit task ← BUG! +``` -This avoids blocking a worker thread while waiting for messages. +**Fix**: Use `with-mutex` on a per-actor scheduling mutex when transitioning states. ```scheme #!chezscheme @@ -574,18 +709,16 @@ This avoids blocking a worker thread while waiting for messages. (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 + spawn-actor/linked ;; links to current actor; if either dies, both notified actor-ref? actor-ref-id + actor-ref-name 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 + ;; Context inside a behavior self ;; (self) → current actor's actor-ref actor-id ;; (actor-id) → current actor's id integer @@ -594,28 +727,36 @@ This avoids blocking a worker thread while waiting for messages. actor-kill! ;; (actor-kill! actor-ref) → forcibly terminate actor-wait! ;; (actor-wait! actor-ref) → block until dead + ;; Monitors and links + actor-ref-links ;; accessor for linked actors list + actor-ref-links-set! ;; mutator (used by spawn-actor/linked) + actor-ref-monitors ;; accessor for monitor list + actor-ref-monitors-set! ;; mutator (used by supervisor) + ;; Dead letter handler - set-dead-letter-handler! ;; (set-dead-letter-handler! proc) + set-dead-letter-handler! + + ;; Scheduler integration + set-actor-scheduler! - ;; Low-level: use the default scheduler or provide one - set-actor-scheduler! ;; (set-actor-scheduler! sched) + ;; Internal: lookup for distributed layer + lookup-local-actor ) (import (chezscheme) - (std actor mpsc) - (std actor scheduler)) + (std actor mpsc)) ;; ========== Actor ID generation ========== - ;; Simple monotonic counter; fine for local actors. + ;; Simple monotonic counter protected by mutex. + ;; Thread-safe: multiple threads may spawn actors concurrently. (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)) + (with-mutex *actor-id-mutex* + (let ([id *next-actor-id*]) + (set! *next-actor-id* (fx+ id 1)) + id))) ;; ========== Actor Record ========== @@ -623,7 +764,8 @@ This avoids blocking a worker thread while waiting for messages. (fields (immutable id) ;; unique integer (immutable node) ;; #f = local; string = remote node id - (immutable mailbox) ;; mpsc-queue + (immutable mailbox) ;; mpsc-queue (or #f for remote refs) + (immutable sched-mutex) ;; protects state transitions (idle→scheduled) (mutable state) ;; 'idle | 'scheduled | 'running | 'dead (mutable behavior) ;; current behavior: (lambda (msg) ...) (mutable links) ;; list of actor-refs to notify on death @@ -634,18 +776,34 @@ This avoids blocking a worker thread while waiting for messages. (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 + (case-lambda + ;; Local actor constructor + [(behavior name) + (new (next-actor-id!) + #f ;; local + (make-mpsc-queue) + (make-mutex) ;; sched-mutex + 'idle + behavior + '() ;; links + '() ;; monitors + name + (make-mutex) + (make-condition) + #f)] ;; exit-reason not set yet + ;; Remote actor ref constructor (no mailbox, no behavior) + [(id node) + (new id + node + #f ;; no local mailbox + (make-mutex) + 'idle ;; state unused for remote + (lambda (msg) (void)) + '() '() ;; no links/monitors locally + #f + (make-mutex) + (make-condition) + #f)]))) (sealed #t)) ;; ========== Global actor table ========== @@ -655,20 +813,16 @@ This avoids blocking a worker thread while waiting for messages. (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*)) + (with-mutex *actor-table-mutex* + (hashtable-set! *actor-table* (actor-ref-id a) a))) (define (unregister-local-actor! a) - (mutex-acquire *actor-table-mutex*) - (hashtable-delete! *actor-table* (actor-ref-id a)) - (mutex-release *actor-table-mutex*)) + (with-mutex *actor-table-mutex* + (hashtable-delete! *actor-table* (actor-ref-id a)))) (define (lookup-local-actor id) - (mutex-acquire *actor-table-mutex*) - (let ([a (hashtable-ref *actor-table* id #f)]) - (mutex-release *actor-table-mutex*) - a)) + (with-mutex *actor-table-mutex* + (hashtable-ref *actor-table* id #f))) ;; ========== Thread-local actor context ========== @@ -679,151 +833,213 @@ This avoids blocking a worker thread while waiting for messages. ;; ========== 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)))) + (make-parameter + (lambda (msg dest) + (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)) + (*dead-letter-handler* proc)) ;; ========== Actor scheduler reference ========== + ;; When #f, actors fall back to fork-thread (1:1 mode). (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. + ;; Process messages from the actor's mailbox in a batch. + ;; Called as a task on a worker thread (or a dedicated OS thread in 1:1 mode). + ;; Processes up to max-batch messages before yielding to let other actors run. + (define *max-batch* 64) + (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))]) + (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*)) + ;; Process this message + (guard (exn [#t (actor-die! a exn)]) + ((actor-ref-behavior a) msg)) + ;; If actor died during processing, stop + (unless (eq? (actor-ref-state a) 'dead) + (loop (fx+ count 1)))] + [else + ;; No more messages or batch limit reached + (with-mutex (actor-ref-sched-mutex a) + (cond + ;; Batch limit reached — re-schedule to be fair + [(and ok (eq? (actor-ref-state a) 'running)) + (actor-ref-state-set! a 'scheduled) + (schedule-actor-task! a)] + ;; Check once more if messages arrived while we were processing + [(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, do nothing + [else (void)]))]))))) + + ;; Submit the actor's run-loop as a task to the scheduler + (define (schedule-actor-task! a) + (let ([sched (*actor-scheduler*)]) (if sched - (scheduler-submit! sched (lambda () (run-actor! a))) + ;; Import scheduler-submit! dynamically to avoid circular dependency. + ;; In practice, store the submit procedure in *actor-scheduler*. + ;; For now, *actor-scheduler* holds the submit procedure directly. + (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) + (with-mutex (actor-ref-sched-mutex a) + (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 + ;; Close mailbox (wakes any blocked dequeue) + (guard (exn [#t (void)]) ;; ignore if already closed + (mpsc-close! (actor-ref-mailbox a))) + ;; Notify linked actors (bidirectional links) (for-each (lambda (linked) (when (actor-alive? linked) - (send linked (list 'EXIT (actor-ref-id a) reason)))) + (guard (exn [#t (void)]) ;; don't crash if linked actor is dead + (send linked (list 'EXIT (actor-ref-id a) reason))))) (actor-ref-links a)) - ;; Notify monitors + ;; Notify monitors (one-way) (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))))) + (guard (exn [#t (void)]) + (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)))