Step 3: implement (std actor protocol) — ask/tell/reply, defprotocol
ober
db8bbf1e5acf97f2fd790b6d3c64d4a6c8677efe
--- a/docs/actor-model.md +++ b/docs/actor-model.md @@ -3323,25 +3323,25 @@ Implementation checklist: - [x] Test: 500 actors each receive one message - [x] Test: `actor-wait!` returns after actor killed -### Step 3: Protocol System +### Step 3: Protocol System ✓ COMPLETE **File**: `lib/std/actor/protocol.sls` **Test**: `tests/test-actor-protocol.ss` **Dependencies**: `core.sls`, `(std task)` (for futures) Implementation checklist: -- [ ] `reply-channel` wraps a future from `(std task)` -- [ ] `ask` wraps message in `('$ask rc sender msg)` envelope, returns future -- [ ] `ask-sync` calls `ask` then `future-get` -- [ ] `with-ask-context` macro detects `$ask` envelope, binds reply channel -- [ ] `reply` completes the current reply channel (error if not in ask context) -- [ ] `defprotocol` generates: record types, tell helpers (`!` suffix), ask helpers (`?!` suffix) -- [ ] `tell` is alias for `send` -- [ ] Test: ask/reply round-trip returns correct value -- [ ] Test: defprotocol generates correct struct predicates -- [ ] Test: typed ask helper `?!` blocks and returns value -- [ ] Test: `reply` in non-ask context raises error -- [ ] Test: multiple concurrent asks to same actor +- [x] `reply-channel` wraps a future from `(std task)` +- [x] `ask` wraps message in `('$ask rc sender msg)` envelope, returns future +- [x] `ask-sync` calls `ask` then `future-get` +- [x] `with-ask-context` macro detects `$ask` envelope, binds reply channel +- [x] `reply` completes the current reply channel (error if not in ask context) +- [x] `defprotocol` generates: record types, tell helpers (`!` suffix), ask helpers (`?!` suffix) +- [x] `tell` is alias for `send` +- [x] Test: ask/reply round-trip returns correct value +- [x] Test: defprotocol generates correct struct predicates +- [x] Test: typed ask helper `?!` blocks and returns value +- [x] Test: `reply` in non-ask context raises error +- [x] Test: multiple concurrent asks to same actor ### Step 4: Supervision Trees new file mode 100644 --- /dev/null +++ b/lib/std/actor/protocol.sls @@ -0,0 +1,168 @@ +#!chezscheme +;;; (std actor protocol) — ask/tell/reply, defprotocol macro +;;; +;;; ask wraps the message in a ('$ask reply-channel sender msg) envelope. +;;; The behavior unwraps it via (with-ask-context msg (lambda (actual) ...)) +;;; and calls (reply value) to complete the future. + +(library (std actor protocol) + (export + defprotocol + + ;; Core ask/tell/call + ask ;; (ask actor-ref msg) → future + ask-sync ;; (ask-sync actor-ref msg [timeout-secs]) → value + tell ;; alias for send + + ;; Reply inside a behavior + reply ;; (reply value) — must be in ask context + reply-to ;; (reply-to) → sender actor-ref or #f + + ;; ask envelope unwrapping + with-ask-context + + ;; One-shot reply channels (exposed for advanced use) + make-reply-channel + reply-channel? + reply-channel-get + reply-channel-put! + ) + (import (chezscheme) + (std actor core) + (std task)) + + ;; -------- Reply channels -------- + ;; A thin wrapper around (std task) futures. + + (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 context set by with-ask-context + (define current-reply-channel (make-thread-parameter #f)) + (define current-sender-ref (make-thread-parameter #f)) + + (define (reply value) + (let ([rc (current-reply-channel)]) + (unless rc + (error 'reply "not in an ask context — no reply channel present")) + (reply-channel-put! rc value))) + + (define (reply-to) (current-sender-ref)) + + ;; -------- ask -------- + + (define *ask-tag* '$ask) ;; private envelope tag + + (define (ask actor-ref msg) + (let* ([rc (make-reply-channel)] + [env (list *ask-tag* rc (self) msg)]) + (send actor-ref env) + (reply-channel-future rc))) ;; caller calls future-get + + (define ask-sync + (case-lambda + [(actor-ref msg) + (future-get (ask actor-ref msg))] + [(actor-ref msg timeout-secs) + ;; Polling timeout — 10ms intervals. + ;; For finer granularity, submit a delayed cancellation to the scheduler. + (let ([fut (ask actor-ref msg)]) + (let loop ([remaining timeout-secs]) + (if (future-done? fut) + (future-get fut) + (if (<= remaining 0) + (error 'ask-sync "timeout waiting for reply" actor-ref msg) + (begin + (sleep (make-time 'time-duration 10000000 0)) ;; 10ms + (loop (- remaining 0.01)))))))])) + + ;; -------- tell -------- + + (define (tell actor-ref msg) + (send actor-ref msg)) + + ;; -------- with-ask-context -------- + ;; Unwraps a '$ask envelope and binds the reply channel. + ;; If msg is not an ask envelope, calls body-proc with msg as-is. + + (define-syntax with-ask-context + (syntax-rules () + [(_ msg body-proc) + (if (and (pair? msg) (eq? (car msg) '$ask)) + (let ([rc (cadr msg)] + [sender (caddr msg)] + [actual (cadddr msg)]) + (parameterize ([current-reply-channel rc] + [current-sender-ref sender]) + (body-proc actual))) + (body-proc msg))])) + + ;; -------- defprotocol macro -------- + ;; + ;; (defprotocol service-name + ;; (msg-name field ... [-> result]) + ;; ...) + ;; + ;; Generates for each clause: + ;; - define-record-type service-name:msg-name + ;; - tell helper service-name:msg-name! (always) + ;; - ask helper service-name:msg-name?! (only if -> present) + + (define-syntax defprotocol + (lambda (stx) + (syntax-case stx () + [(_ proto-name clause ...) + (let* ([proto (syntax->datum #'proto-name)] + [prefix (symbol->string proto)]) + + (define (parse-clause datum) + ;; Returns (values name fields has-reply?) + (let loop ([rest (cdr datum)] [fields '()]) + (cond + [(null? rest) + (values (car datum) (reverse fields) #f)] + [(eq? (car rest) '->) + (values (car datum) (reverse fields) #t)] + [else + (loop (cdr rest) (cons (car rest) fields))]))) + + (define (sym . parts) + (string->symbol + (apply string-append + (map (lambda (p) + (if (symbol? p) (symbol->string p) p)) + parts)))) + + (with-syntax + ([(expanded ...) + (map (lambda (c) + (let ([datum (syntax->datum c)]) + (let-values ([(name fields has-reply?) (parse-clause datum)]) + (let* ([struct-name (sym prefix ":" name)] + [make-name (sym "make-" prefix ":" name)] + [pred-name (sym prefix ":" name "?")] + [tell-name (sym prefix ":" name "!")] + [ask-name (sym prefix ":" name "?!")]) + (datum->syntax #'proto-name + `(begin + (define-record-type ,struct-name + (fields ,@(map (lambda (f) `(immutable ,f)) fields)) + (sealed #t)) + (define (,tell-name actor ,@fields) + (tell actor (,make-name ,@fields))) + ,@(if has-reply? + `((define (,ask-name actor ,@fields) + (ask-sync actor (,make-name ,@fields)))) + '()))))))) + (syntax->list #'(clause ...)))]) + #'(begin expanded ...)))]))) + + ) ;; end library new file mode 100644 --- /dev/null +++ b/tests/test-actor-protocol.ss @@ -0,0 +1,142 @@ +#!chezscheme +;;; Tests for (std actor protocol) — ask/tell/reply, defprotocol + +(import (chezscheme) (jerboa core) + (std actor core) (std actor protocol)) + +(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 for set comparison +(define (list->set lst) (list-sort < lst)) + +(printf "--- (std actor protocol) tests ---~%") + +;; Test 1: ask/reply round-trip +(let ([a (spawn-actor + (lambda (msg) + (with-ask-context msg + (lambda (actual) + (match actual + [('add x y) (reply (+ x y))] + [_ (void)])))))]) + (test "ask-reply" (ask-sync a '(add 3 4)) 7) + (actor-kill! a)) + +;; Test 2: reply in non-ask context raises error +(let ([raised #f] + [m (make-mutex)] [c (make-condition)]) + (let ([a (spawn-actor + (lambda (msg) + (guard (exn [#t (set! raised #t)]) + (reply 99)) + (with-mutex m (condition-signal c))))]) + (send a 'trigger) + (with-mutex m (let loop () (unless raised (condition-wait c m) (loop)))) + (test "reply-no-context" raised #t) + (actor-kill! a))) + +;; Test 3: tell is fire-and-forget (does not block) +(let ([got #f] + [m (make-mutex)] [c (make-condition)]) + (let ([a (spawn-actor + (lambda (msg) + (with-mutex m (set! got msg) (condition-signal c))))]) + (tell a 'notification) + (with-mutex m (let loop () (unless got (condition-wait c m) (loop)))) + (test "tell" got 'notification) + (actor-kill! a))) + +;; Test 4: ask-sync with timeout raises error on expire +(let ([a (spawn-actor (lambda (msg) (void)))]) ;; never replies + (let ([timed-out #f]) + (guard (exn [#t (set! timed-out #t)]) + (ask-sync a 'anything 0.05)) ;; 50ms timeout + (test "ask-timeout" timed-out #t)) + (actor-kill! a)) + +;; Test 5: defprotocol generates correct types and helpers +(defprotocol math + (square x -> result) + (log-msg text)) + +(let ([a (spawn-actor + (lambda (msg) + (with-ask-context msg + (lambda (actual) + (cond + [(math:square? actual) + (reply (* (math:square-x actual) (math:square-x actual)))] + [(math:log-msg? actual) + (void)] ;; fire-and-forget + [else (void)])))))]) + ;; Record predicates + (test "struct-pred-square" (math:square? (make-math:square 5)) #t) + (test "struct-pred-log" (math:log-msg? (make-math:log-msg "hi")) #t) + ;; ask helper + (test "defprotocol-ask?!" (math:square?! a 7) 49) + ;; tell helper (no return value check needed — fire-and-forget) + (math:log-msg! a "hello") + (actor-kill! a)) + +;; Test 6: reply-to returns sender actor-ref +(let ([sender-got #f] + [m (make-mutex)] [c (make-condition)]) + (let* ([responder + (spawn-actor + (lambda (msg) + (with-ask-context msg + (lambda (actual) + (with-mutex m + (set! sender-got (reply-to)) + (condition-signal c)) + (reply 'ok)))))] + [requester + (spawn-actor + (lambda (msg) (void)))]) + ;; ask from the test thread — reply-to will be #f (no actor context) + (ask-sync responder 'check) + ;; sender-got may be #f since we called from non-actor thread + (test "reply-to-outside-actor" sender-got #f) + (actor-kill! responder) + (actor-kill! requester))) + +;; Test 7: multiple concurrent asks to same actor +(let ([a (spawn-actor + (lambda (msg) + (with-ask-context msg + (lambda (actual) + (match actual + [('echo v) (reply v)] + [_ (void)])))))] + [results '()] + [m (make-mutex)] [c (make-condition)]) + (do ([i 0 (+ i 1)]) ((= i 10)) + (let ([n i]) + (fork-thread + (lambda () + (let ([v (ask-sync a (list 'echo n))]) + (with-mutex m + (set! results (cons v results)) + (when (= (length results) 10) (condition-signal c)))))))) + (with-mutex m + (let loop () (unless (= (length results) 10) (condition-wait c m) (loop)))) + (test "concurrent-asks-count" (length results) 10) + (test "concurrent-asks-values" (list->set results) (list->set '(0 1 2 3 4 5 6 7 8 9))) + (actor-kill! a)) + +(printf "~%Results: ~a passed, ~a failed~%" pass fail) +(when (> fail 0) (exit 1))