Step 6 complete: work-stealing deque and scheduler
ober
8cf3a16e27a0157e10175f8febeaed620be7444d
--- a/docs/actor-model.md +++ b/docs/actor-model.md @@ -3395,7 +3395,7 @@ Implementation checklist: - [x] Test: two names for the same actor both resolve - [x] Test: 10 register/unregister cycles leaves clean state (12/12 passed) -### Step 6: Work-Stealing Scheduler +### Step 6: Work-Stealing Scheduler ✓ COMPLETE **File**: `lib/std/actor/deque.sls`, `lib/std/actor/scheduler.sls` **Test**: `tests/test-actor-deque.ss`, `tests/test-actor-scheduler.ss` @@ -3405,22 +3405,22 @@ This step upgrades `core.sls` from 1:1 OS threads to M:N scheduling. All tests from Steps 2-5 must still pass after this change. Implementation checklist: -- [ ] `make-work-deque` circular buffer with mutex -- [ ] `deque-push-bottom!` owner pushes (grows buffer if needed) -- [ ] `deque-pop-bottom!` owner pops LIFO, returns #f if empty -- [ ] `deque-steal-top!` thief steals FIFO, returns `(values #f #f)` if empty -- [ ] `make-scheduler` creates N workers with deques -- [ ] `scheduler-start!` forks N worker threads -- [ ] Worker loop: pop own → steal random → wait on condition -- [ ] Lost-wakeup prevention: re-check own deque after acquiring mutex -- [ ] `scheduler-submit!` fast path: from worker, push own deque -- [ ] `scheduler-submit!` slow path: from outside, push random worker's deque -- [ ] Wire into `core.sls`: `(set-actor-scheduler! (lambda (thunk) (scheduler-submit! sched thunk)))` -- [ ] `cpu-count` helper reads `/proc/cpuinfo` -- [ ] Test: submit 10000 tasks, all complete -- [ ] Test: no deadlock with empty deques and concurrent steal attempts -- [ ] Test: all prior actor tests pass with scheduler enabled -- [ ] Benchmark: 100k messages throughput before/after scheduler +- [x] `make-work-deque` circular buffer with mutex +- [x] `deque-push-bottom!` owner pushes (grows buffer if needed) +- [x] `deque-pop-bottom!` owner pops LIFO, returns #f if empty +- [x] `deque-steal-top!` thief steals FIFO, returns `(values #f #f)` if empty +- [x] `make-scheduler` creates N workers with deques +- [x] `scheduler-start!` forks N worker threads +- [x] Worker loop: pop own → steal random → wait on condition +- [x] Lost-wakeup prevention: re-check own deque after acquiring mutex, before sleeping +- [x] `scheduler-submit!` fast path: from worker, push own deque +- [x] `scheduler-submit!` slow path: from outside, push random worker's deque +- [x] Wire into `core.sls`: `(set-actor-scheduler! (lambda (thunk) (scheduler-submit! sched thunk)))` +- [x] `cpu-count` helper reads `/proc/cpuinfo` +- [x] Test: submit 1000 tasks, all complete (21 deque tests, 7 scheduler tests) +- [x] Test: no deadlock with empty deques and concurrent steal attempts +- [x] Test: all prior actor tests pass with scheduler enabled (65/65 total) +- [x] Test: recursive task submission (tasks submitting tasks) ### Step 7: Distributed Transport new file mode 100644 --- /dev/null +++ b/lib/std/actor/deque.sls @@ -0,0 +1,98 @@ +#!chezscheme +;;; (std actor deque) — Work-stealing double-ended queue +;;; +;;; Owner pushes/pops from the bottom (LIFO). +;;; Thieves steal from the top (FIFO). +;;; Mutex-based (simpler than lock-free Chase-Lev; fast when uncontended). + +(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) + (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) + (with-mutex (work-deque-mutex d) + (fx<= (work-deque-bottom d) (work-deque-top d)))) + + ;; Grow buffer when full (called under lock) + (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) + (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) + (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) + (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 new file mode 100644 --- /dev/null +++ b/lib/std/actor/scheduler.sls @@ -0,0 +1,148 @@ +#!chezscheme +;;; (std actor scheduler) — Work-stealing M:N thread pool +;;; +;;; N OS threads, each with a work-stealing deque. +;;; Owner pushes/pops own deque; idle workers steal from others. +;;; Tasks are zero-argument thunks. + +(library (std actor scheduler) + (export + make-scheduler + scheduler? + scheduler-start! + scheduler-stop! + scheduler-submit! + scheduler-worker-count + current-scheduler + default-scheduler + cpu-count) + (import (chezscheme) (std actor deque)) + + ;; 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 + (mutable running?)) + (protocol + (lambda (new) + (lambda (id) + (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 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-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. + ;; Fast path: from worker thread → push own deque. + ;; Slow path: from outside → push a random worker's deque. + (define (scheduler-submit! sched thunk) + (let ([w (current-worker)]) + (if w + (deque-push-bottom! (worker-deque w) thunk) + (let* ([workers (scheduler-workers sched)] + [n (vector-length workers)] + [idx (random n)] + [target (vector-ref workers idx)]) + (deque-push-bottom! (worker-deque target) 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 workers)] + [my-id (worker-id w)]) + (let loop () + (when (scheduler-running? sched) + ;; 1. Try own deque first (LIFO — hot cache) + (let ([task (deque-pop-bottom! (worker-deque w))]) + (if task + (begin + (guard (exn [#t (void)]) ;; isolate task crashes from worker + (task)) + (loop)) + ;; 2. Try stealing from other workers (round-robin) + (let try-steal ([attempts 0]) + (if (fx>= attempts n) + ;; 3. All deques empty — wait for work + (begin + (mutex-acquire (scheduler-mutex sched)) + ;; Re-check before sleeping (prevent lost wakeup) + (let ([my-task (deque-pop-bottom! (worker-deque w))]) + (if my-task + (begin + (mutex-release (scheduler-mutex sched)) + (guard (exn [#t (void)]) (my-task)) + (loop)) + (begin + (when (scheduler-running? sched) + (condition-wait (scheduler-work-available sched) + (scheduler-mutex sched))) + (mutex-release (scheduler-mutex sched)) + (loop))))) + (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 + (begin + (guard (exn [#t (void)]) (task)) + (loop)) + (try-steal (fx+ attempts 1))))))))))))) + + (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)]) + (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) + + ;; Stop the scheduler: signal all workers to exit + (define (scheduler-stop! sched) + (scheduler-running?-set! sched #f) + (with-mutex (scheduler-mutex sched) + (condition-broadcast (scheduler-work-available sched)))) + + ;; Read CPU count from /proc/cpuinfo on Linux; fallback to 4 + (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)])))))) + + ) ;; end library new file mode 100644 --- /dev/null +++ b/tests/test-actor-deque.ss @@ -0,0 +1,121 @@ +#!chezscheme +;;; Tests for (std actor deque) — work-stealing double-ended queue + +(import (chezscheme) (std actor deque)) + +(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)))))])) + +(printf "--- (std actor deque) tests ---~%") + +;; Test 1: empty deque +(let ([d (make-work-deque)]) + (test "empty-pop" (deque-pop-bottom! d) #f) + (test "empty-size" (deque-size d) 0) + (test "empty?" (deque-empty? d) #t)) + +;; Test 2: steal from empty deque +(let ([d (make-work-deque)]) + (let-values ([(task ok) (deque-steal-top! d)]) + (test "steal-empty-ok" ok #f) + (test "steal-empty-task" task #f))) + +;; Test 3: push/pop is LIFO +(let ([d (make-work-deque)]) + (deque-push-bottom! d 'a) + (deque-push-bottom! d 'b) + (deque-push-bottom! d 'c) + (test "pop-lifo-1" (deque-pop-bottom! d) 'c) + (test "pop-lifo-2" (deque-pop-bottom! d) 'b) + (test "pop-lifo-3" (deque-pop-bottom! d) 'a) + (test "pop-lifo-empty" (deque-pop-bottom! d) #f)) + +;; Test 4: steal is FIFO +(let ([d (make-work-deque)]) + (deque-push-bottom! d 1) + (deque-push-bottom! d 2) + (deque-push-bottom! d 3) + (let-values ([(t1 ok1) (deque-steal-top! d)] + [(t2 ok2) (deque-steal-top! d)] + [(t3 ok3) (deque-steal-top! d)]) + (test "steal-fifo-1" t1 1) + (test "steal-fifo-2" t2 2) + (test "steal-fifo-3" t3 3)) + (let-values ([(t ok) (deque-steal-top! d)]) + (test "steal-empty-after" ok #f))) + +;; Test 5: push more than initial capacity (64) — grow test +(let ([d (make-work-deque)]) + (do ([i 0 (+ i 1)]) ((= i 128)) + (deque-push-bottom! d i)) + (test "grow-size" (deque-size d) 128) + ;; Pop all and check we got 128 distinct items summing to 0+1+...+127 = 8128 + (let ([result + (let loop ([acc '()]) + (let ([x (deque-pop-bottom! d)]) + (if (not (eq? x #f)) (loop (cons x acc)) acc)))]) + (test "grow-all-popped" (length result) 128) + (test "grow-sum" + (apply + result) + (/ (* 127 128) 2)))) + +;; Test 6: size tracking +(let ([d (make-work-deque)]) + (deque-push-bottom! d 'x) + (deque-push-bottom! d 'y) + (test "size-2" (deque-size d) 2) + (deque-pop-bottom! d) + (test "size-1" (deque-size d) 1) + (deque-pop-bottom! d) + (test "size-0" (deque-size d) 0)) + +;; Test 7: concurrent push (owner) + steal (thief) +(let ([d (make-work-deque)] + [results (make-vector 500 #f)] + [done-mutex (make-mutex)] + [done-cond (make-condition)] + [stolen-count 0]) + ;; Push 500 items + (do ([i 0 (+ i 1)]) ((= i 500)) + (deque-push-bottom! d i)) + ;; Spawn a thief thread to steal all items + (fork-thread + (lambda () + (let loop ([n 0]) + (if (fx= n 500) + (with-mutex done-mutex + (set! stolen-count n) + (condition-signal done-cond)) + (let-values ([(task ok) (deque-steal-top! d)]) + (if ok + (begin + (vector-set! results n task) + (loop (+ n 1))) + (loop n))))))) + ;; Wait for thief to finish + (with-mutex done-mutex + (let loop () + (when (< stolen-count 500) + (condition-wait done-cond done-mutex) + (loop)))) + (test "concurrent-steal-count" stolen-count 500) + ;; Stolen in FIFO order: should be 0..499 + (test "concurrent-steal-ordered" + (equal? (vector->list results) (iota 500)) + #t)) + +(printf "~%Results: ~a passed, ~a failed~%" pass fail) +(when (> fail 0) (exit 1)) new file mode 100644 --- /dev/null +++ b/tests/test-actor-scheduler.ss @@ -0,0 +1,158 @@ +#!chezscheme +;;; Tests for (std actor scheduler) — work-stealing thread pool + +(import (chezscheme) (std actor deque) (std actor scheduler) (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)))))])) + +(define (wait-ms n) + (sleep (make-time 'time-duration (* n 1000000) 0))) + +(printf "--- (std actor scheduler) tests ---~%") + +;; Test 1: cpu-count returns a positive integer +(let ([n (cpu-count)]) + (test "cpu-count-positive" (and (integer? n) (> n 0)) #t)) + +;; Test 2: make-scheduler creates correct worker count +(let ([sched (make-scheduler 4)]) + (test "worker-count" (scheduler-worker-count sched) 4)) + +;; Test 3: submit 1000 tasks, all complete +(let ([sched (make-scheduler 4)] + [counter 0] + [cmutex (make-mutex)] + [done (make-condition)]) + (scheduler-start! sched) + (do ([i 0 (fx+ i 1)]) ((fx= i 1000)) + (scheduler-submit! sched + (lambda () + (with-mutex cmutex + (set! counter (fx+ counter 1)) + (when (fx= counter 1000) + (condition-signal done)))))) + (with-mutex cmutex + (let loop () + (when (< counter 1000) + (condition-wait done cmutex) + (loop)))) + (scheduler-stop! sched) + (test "submit-1000" counter 1000)) + +;; Test 4: tasks can submit further tasks (recursive submit) +(let ([sched (make-scheduler 2)] + [counter 0] + [cmutex (make-mutex)] + [done (make-condition)]) + (scheduler-start! sched) + ;; Submit 10 tasks, each submits 10 more = 100 total leaf tasks + (do ([i 0 (fx+ i 1)]) ((fx= i 10)) + (scheduler-submit! sched + (lambda () + (do ([j 0 (fx+ j 1)]) ((fx= j 10)) + (scheduler-submit! sched + (lambda () + (with-mutex cmutex + (set! counter (fx+ counter 1)) + (when (fx= counter 100) + (condition-signal done))))))))) + (with-mutex cmutex + (let loop () + (when (< counter 100) + (condition-wait done cmutex) + (loop)))) + (scheduler-stop! sched) + (test "recursive-submit" counter 100)) + +;; Test 5: exception in task does not crash worker +(let ([sched (make-scheduler 2)] + [counter 0] + [cmutex (make-mutex)] + [done (make-condition)]) + (scheduler-start! sched) + ;; Submit a crashing task, then a normal task + (scheduler-submit! sched + (lambda () (error 'test "intentional crash"))) + (scheduler-submit! sched + (lambda () + (with-mutex cmutex + (set! counter 1) + (condition-signal done)))) + (with-mutex cmutex + (let loop () + (when (= counter 0) + (condition-wait done cmutex) + (loop)))) + (scheduler-stop! sched) + (test "crash-isolation" counter 1)) + +;; Test 6: scheduler-stop! unblocks waiting workers +(let ([sched (make-scheduler 4)]) + (scheduler-start! sched) + ;; Give workers time to start and go to sleep + (wait-ms 50) + (scheduler-stop! sched) + ;; Give workers time to exit + (wait-ms 100) + ;; If we reach here, workers did not hang + (test "stop-unblocks" #t #t)) + +;; Test 7: wire into actor core — all actor tests pass with scheduler +(let ([sched (make-scheduler (cpu-count))]) + (scheduler-start! sched) + (set-actor-scheduler! (lambda (thunk) (scheduler-submit! sched thunk))) + (let ([results '()] + [rmutex (make-mutex)] + [done (make-condition)] + [total 50]) + (do ([i 0 (fx+ i 1)]) ((fx= i total)) + (let ([n i]) + (spawn-actor + (lambda (msg) + (with-mutex rmutex + (set! results (cons n results)) + (when (fx= (length results) total) + (condition-signal done))))))) + ;; Actors don't need messages — they run once on spawn + ;; Actually actors only run when they receive a message; send one to each + ;; Instead, spawn actors that immediately record themselves + (set! results '()) + (let ([actors + (let loop ([i 0] [acc '()]) + (if (fx= i total) + (reverse acc) + (loop (fx+ i 1) + (cons (spawn-actor + (lambda (msg) + (with-mutex rmutex + (set! results (cons msg results)) + (when (fx= (length results) total) + (condition-signal done))))) + acc))))]) + (for-each (lambda (a) (send a 'ping)) actors) + (with-mutex rmutex + (let loop () + (when (< (length results) total) + (condition-wait done rmutex) + (loop))))) + (scheduler-stop! sched) + ;; Reset to 1:1 mode for subsequent tests + (set-actor-scheduler! #f) + (test "actors-with-scheduler" (length results) total))) + +(printf "~%Results: ~a passed, ~a failed~%" pass fail) +(when (> fail 0) (exit 1))