Step 1: implement (std actor mpsc) — MPSC mailbox queue
ober
142d9e3f7977bd271bfb7851f484f62733199c39
--- a/docs/actor-model.md +++ b/docs/actor-model.md @@ -3275,23 +3275,23 @@ node-id parsing, cookie hash). Cross-process tests are described as manual steps Implement and test each step before moving to the next. -### Step 1: MPSC Queue +### Step 1: MPSC Queue ✓ COMPLETE **File**: `lib/std/actor/mpsc.sls` **Test**: `tests/test-actor-mpsc.ss` **Dependencies**: `(chezscheme)` only Implementation checklist: -- [ ] `make-mpsc-queue` creates two-lock linked list with dummy head -- [ ] `mpsc-enqueue!` acquires tail-lock, signals after releasing tail-lock -- [ ] `mpsc-dequeue!` acquires head-lock, blocks via `condition-wait` if empty -- [ ] `mpsc-try-dequeue!` returns `(values #f #f)` immediately if empty -- [ ] `mpsc-close!` sets closed flag, broadcasts to wake blocked consumers -- [ ] All locking uses `with-mutex` for exception safety -- [ ] Test: 10 concurrent producers, 1 consumer, verify all messages received -- [ ] Test: `try-dequeue` on empty returns `(values #f #f)` -- [ ] Test: close wakes blocked consumer with error -- [ ] Test: single-producer ordering is preserved (FIFO) +- [x] `make-mpsc-queue` creates two-lock linked list with dummy head +- [x] `mpsc-enqueue!` acquires tail-lock, signals after releasing tail-lock +- [x] `mpsc-dequeue!` acquires head-lock, blocks via `condition-wait` if empty +- [x] `mpsc-try-dequeue!` returns `(values #f #f)` immediately if empty +- [x] `mpsc-close!` sets closed flag, broadcasts to wake blocked consumers +- [x] All locking uses `with-mutex` for exception safety +- [x] Test: 10 concurrent producers, 1 consumer, verify all messages received +- [x] Test: `try-dequeue` on empty returns `(values #f #f)` +- [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) new file mode 100644 --- /dev/null +++ b/lib/std/actor/mpsc.sls @@ -0,0 +1,117 @@ +#!chezscheme +;;; (std actor mpsc) — Multi-Producer Single-Consumer queue (mailbox) +;;; +;;; Two-lock linked list: one lock for the tail (producers), one for the head +;;; (consumer). Producers never block the consumer. Signal the consumer by +;;; briefly acquiring head-mutex AFTER releasing tail-mutex — no nested locking. + +(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 (values #f #f) immediately + mpsc-empty? ;; approximate — safe only from consumer thread + mpsc-close! ;; signal no more messages; wakes blocked consumers + mpsc-closed?) + (import (chezscheme)) + + ;; -------- Linked-list node -------- + + (define-record-type mpsc-node + (fields + (mutable value) ;; message payload, or 'sentinel for dummy head + (mutable next)) ;; next node or #f + (protocol + (lambda (new) + (lambda (val) (new val #f)))) + (sealed #t)) + + ;; -------- Queue record -------- + + (define-record-type mpsc-queue + (fields + (mutable head) ;; dummy node; consumer reads head.next + (mutable tail) ;; last real node (or dummy when empty) + (immutable head-mutex) ;; consumer lock + condition variable + (immutable tail-mutex) ;; producer lock + (immutable not-empty) ;; condition: signaled on enqueue + (mutable closed?)) + (protocol + (lambda (new) + (lambda () + (let ([dummy (make-mpsc-node 'sentinel)]) + (new dummy dummy + (make-mutex) (make-mutex) + (make-condition) + #f))))) + (sealed #t)) + + ;; -------- Producer -------- + + ;; Enqueue a value. Acquires tail-mutex only. + ;; Signals consumer AFTER releasing tail-mutex to avoid nested locking. + (define (mpsc-enqueue! q val) + (when (mpsc-queue-closed? q) + (error 'mpsc-enqueue! "queue is closed")) + (let ([node (make-mpsc-node val)]) + (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 + (with-mutex (mpsc-queue-head-mutex q) + (condition-signal (mpsc-queue-not-empty q)))) + + ;; -------- Consumer -------- + + ;; Dequeue, blocking if the queue is empty. + (define (mpsc-dequeue! q) + (with-mutex (mpsc-queue-head-mutex q) + (let loop () + (let ([next (mpsc-node-next (mpsc-queue-head q))]) + (cond + [next + (let ([val (mpsc-node-value next)]) + ;; Advance dummy head; old head is discarded + (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)]))))) + + ;; Try dequeue without blocking. + ;; Returns (values val #t) on success, (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))]) + (if next + (let ([val (mpsc-node-value next)]) + (mpsc-queue-head-set! q next) + (mpsc-node-value-set! next 'sentinel) + (values val #t)) + (values #f #f))))) + + ;; Approximate empty check — safe only from the consumer thread. + (define (mpsc-empty? q) + (not (mpsc-node-next (mpsc-queue-head q)))) + + ;; -------- Lifecycle -------- + + (define (mpsc-close! 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)))) + + ;; Public predicate — wraps the record field accessor + (define (mpsc-closed? q) (mpsc-queue-closed? q)) + + ) ;; end library new file mode 100644 --- /dev/null +++ b/tests/test-actor-mpsc.ss @@ -0,0 +1,127 @@ +#!chezscheme +;;; Tests for (std actor mpsc) — MPSC mailbox queue + +(import (chezscheme) (std actor mpsc)) + +(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 mpsc) tests ---~%") + +;; Test 1: basic enqueue / dequeue +(let ([q (make-mpsc-queue)]) + (mpsc-enqueue! q 'hello) + (mpsc-enqueue! q 'world) + (test "dequeue-1" (mpsc-dequeue! q) 'hello) + (test "dequeue-2" (mpsc-dequeue! q) 'world)) + +;; Test 2: try-dequeue on empty returns (values #f #f) +(let ([q (make-mpsc-queue)]) + (let-values ([(v ok) (mpsc-try-dequeue! q)]) + (test "try-empty-val" v #f) + (test "try-empty-ok" ok #f))) + +;; Test 3: try-dequeue after enqueue returns value +(let ([q (make-mpsc-queue)]) + (mpsc-enqueue! q 42) + (let-values ([(v ok) (mpsc-try-dequeue! q)]) + (test "try-val" v 42) + (test "try-ok" ok #t))) + +;; Test 4: single-producer FIFO ordering +(let ([q (make-mpsc-queue)]) + (do ([i 0 (+ i 1)]) ((= i 100)) + (mpsc-enqueue! q i)) + (let loop ([i 0] [ok #t]) + (if (= i 100) + (test "fifo-order" ok #t) + (let-values ([(v got) (mpsc-try-dequeue! q)]) + (loop (+ i 1) (and ok got (= v i))))))) + +;; Test 5: 10 concurrent producers, 1 consumer — all messages received +(let ([q (make-mpsc-queue)] + [received (make-eq-hashtable)] + [recv-mutex (make-mutex)] + [prod-done 0] + [prod-mutex (make-mutex)] + [prod-cond (make-condition)]) + (define msgs-per-thread 100) + (define total (* 10 msgs-per-thread)) + ;; Start 10 producers + (do ([t 0 (+ t 1)]) ((= t 10)) + (let ([tid t]) + (fork-thread + (lambda () + (do ([i 0 (+ i 1)]) ((= i msgs-per-thread)) + (mpsc-enqueue! q (+ (* tid msgs-per-thread) i))) + (with-mutex prod-mutex + (set! prod-done (+ prod-done 1)) + (when (= prod-done 10) + (condition-signal prod-cond))))))) + ;; Wait for all producers + (with-mutex prod-mutex + (let loop () (unless (= prod-done 10) (condition-wait prod-cond prod-mutex) (loop)))) + (mpsc-close! q) + ;; Consume all + (let loop ([count 0]) + (let-values ([(v ok) (mpsc-try-dequeue! q)]) + (if ok + (begin (hashtable-set! received v #t) (loop (+ count 1))) + (test "concurrent-producers-count" count total)))) + ;; Verify every expected message was received + (let ([missing 0]) + (do ([i 0 (+ i 1)]) ((= i total)) + (unless (hashtable-ref received i #f) + (set! missing (+ missing 1)))) + (test "concurrent-producers-no-missing" missing 0))) + +;; Test 6: close wakes a blocked consumer with an error +(let ([q (make-mpsc-queue)] + [error-caught #f] + [done-mutex (make-mutex)] + [done-cond (make-condition)]) + (fork-thread + (lambda () + (guard (exn [#t (set! error-caught #t)]) + (mpsc-dequeue! q)) + (with-mutex done-mutex (condition-signal done-cond)))) + (sleep (make-time 'time-duration 50000000 0)) ;; 50ms + (mpsc-close! q) + (with-mutex done-mutex + (let loop () + (unless error-caught + (condition-wait done-cond done-mutex) + (loop)))) + (test "close-wakes-consumer" error-caught #t)) + +;; Test 7: enqueue on closed queue raises error +(let ([q (make-mpsc-queue)]) + (mpsc-close! q) + (let ([raised #f]) + (guard (exn [#t (set! raised #t)]) + (mpsc-enqueue! q 'x)) + (test "enqueue-closed" raised #t))) + +;; Test 8: mpsc-empty? reflects state +(let ([q (make-mpsc-queue)]) + (test "empty-initially" (mpsc-empty? q) #t) + (mpsc-enqueue! q 1) + (test "not-empty-after-enqueue" (mpsc-empty? q) #f) + (mpsc-try-dequeue! q) + (test "empty-after-dequeue" (mpsc-empty? q) #t)) + +(printf "~%Results: ~a passed, ~a failed~%" pass fail) +(when (> fail 0) (exit 1))