Migrate 95 .sls with define-record-type -> defstruct
ober
1e1e41aa5b268e1190b0e891a1b507f87e2d0b4c
--- a/lib/jerboa/prelude/safe.ss +++ b/lib/jerboa/prelude/safe.ss @@ -6,6 +6,7 @@ ;;; resource-safe equivalents under the STANDARD names: ;;; ;;; (import (jerboa prelude safe) +;;; (only (jerboa core) def) ;;; (only (jerboa core) def)) ;;; (with-resource ([db (sqlite-open "test.db")]) ;;; (sqlite-exec db "CREATE TABLE t(x)") @@ -185,7 +186,8 @@ sandbox-config-timeout sandbox-config-seccomp sandbox-config-landlock sandbox-config-capabilities *sandbox-timeout* *sandbox-seccomp* *sandbox-landlock* - &sandbox-error sandbox-error? sandbox-error-phase sandbox-error-detail)) + &sandbox-error sandbox-error? sandbox-error-phase sandbox-error-detail) + (only (jerboa core) def)) ;; ========================================================================= ;; Re-export safe APIs under standard names deleted file mode 100644 --- a/lib/std/actor/distributed.sls +++ /dev/null @@ -1,323 +0,0 @@ -#!chezscheme -;;; (std actor distributed) — Location-transparent distributed actor messaging -;;; -;;; Builds on (std actor core) and (std actor cluster) to provide: -;;; - Location-transparent send (dsend / dsend/ask) -;;; - Cluster-wide name registration -;;; - Process groups (broadcast) -;;; - Distributed supervision -;;; - Node failure detection / monitoring -;;; - Simple serialization via write/read on string ports - -(library (std actor distributed) - (export - ;; Location-transparent send - dsend - dsend/ask - - ;; Remote actor references - make-remote-ref - remote-ref? - remote-ref-node - remote-ref-id - - ;; Cluster-wide name registration - cluster-register! - cluster-whereis - cluster-registered-names - - ;; Process groups - make-process-group - process-group-join! - process-group-leave! - process-group-members - process-group-broadcast! - - ;; Distributed supervision - make-dist-supervisor - dist-supervisor-start-child! - dist-supervisor-children - - ;; Failure detection - monitor-node - demonitor-node - node-alive? - ping-node - - ;; Serialization - serialize-message - deserialize-message - - ;; Configuration parameters - *default-send-timeout* - *cluster-name* - *max-message-size*) - - (import (chezscheme) - (std actor core) - (except (std actor cluster) node-alive?)) - - ;; ====================================================================== - ;; Configuration parameters - ;; ====================================================================== - - (define *default-send-timeout* (make-parameter 5000)) ;; 5 seconds in ms - (define *cluster-name* (make-parameter "local")) - - ;; ====================================================================== - ;; Remote actor references - ;; ====================================================================== - - ;; A remote-ref identifies an actor on a specific cluster node by name. - (define-record-type remote-ref-rec - (fields - (immutable node) ;; node name (string or symbol) - (immutable id)) ;; actor name or id - (sealed #t)) - - (define (make-remote-ref node-name actor-id) - (make-remote-ref-rec node-name actor-id)) - - (define (remote-ref? x) (remote-ref-rec? x)) - (define (remote-ref-node x) (remote-ref-rec-node x)) - (define (remote-ref-id x) (remote-ref-rec-id x)) - - ;; Is actor-ref local? (uses (std actor core) actor-ref? predicate) - (define (%local-ref? ref) - (actor-ref? ref)) - - ;; ====================================================================== - ;; Location-transparent send - ;; ====================================================================== - - ;; (dsend actor-ref msg) - ;; Works for both local actor-refs and remote-refs. - ;; For local refs, delegates directly to (send). - ;; For remote refs, looks up the named actor on the target node. - (define (dsend ref msg) - (cond - [(%local-ref? ref) - ;; Local actor: use the core send - (send ref msg)] - [(remote-ref? ref) - ;; Remote: find actor in cluster registry - (let* ([node-name (remote-ref-node ref)] - [actor-id (remote-ref-id ref)] - [node (cluster-node-by-name node-name)]) - (if node - (let ([actor (remote-whereis node actor-id)]) - (if actor - (send actor msg) - (error 'dsend "actor not found on node" actor-id node-name))) - (error 'dsend "node not found" node-name)))] - [else - (error 'dsend "not a valid actor reference" ref)])) - - ;; (dsend/ask actor-ref msg timeout-ms) -> reply or #f - ;; Sends msg and waits for a reply; uses a temporary channel actor. - (define (dsend/ask ref msg timeout-ms) - (let* ([result-box (make-mutex)] - [reply #f] - [replied? #f] - [cond-var (make-condition)] - [lock (make-mutex)]) - ;; Spawn a one-shot reply actor - (let ([reply-actor - (spawn-actor - (lambda (m) - (with-mutex lock - (set! reply m) - (set! replied? #t) - (condition-signal cond-var))))]) - ;; Send original message with reply-to field prepended - (dsend ref (list 'ask reply-actor msg)) - ;; Wait for reply with timeout - (let ([deadline (+ (current-time-ms) timeout-ms)]) - (with-mutex lock - (let loop () - (unless replied? - (let ([now (current-time-ms)]) - (when (< now deadline) - (condition-wait cond-var lock) - (loop)))))) - (if replied? reply #f))))) - - (define (current-time-ms) - (* 1000 (time-second (current-time)))) - - ;; ====================================================================== - ;; Cluster-wide name registration - ;; ====================================================================== - - ;; Global name table: name -> actor-ref (local refs) - (define *global-registry* (make-hashtable equal-hash equal?)) - (define *global-registry-mutex* (make-mutex)) - - ;; (cluster-register! name actor-ref) — register actor under a cluster-wide name - (define (cluster-register! name actor-ref) - (with-mutex *global-registry-mutex* - (hashtable-set! *global-registry* name actor-ref)) - ;; Also register on all alive nodes in the cluster - (for-each - (lambda (node) - (remote-register! node name actor-ref)) - (cluster-nodes))) - - ;; (cluster-whereis name) -> actor-ref or #f (searches local registry first) - (define (cluster-whereis name) - (or (with-mutex *global-registry-mutex* - (hashtable-ref *global-registry* name #f)) - (whereis/any name))) - - ;; (cluster-registered-names) -> list of names - (define (cluster-registered-names) - (with-mutex *global-registry-mutex* - (vector->list (hashtable-keys *global-registry*)))) - - ;; ====================================================================== - ;; Process groups - ;; ====================================================================== - - (define-record-type process-group-rec - (fields - (immutable name) - (mutable members) ;; list of actor-refs (or remote-refs) - (immutable mutex)) - (sealed #t)) - - (define (make-process-group name) - (make-process-group-rec name '() (make-mutex))) - - (define (process-group-join! group ref) - (with-mutex (process-group-rec-mutex group) - (unless (member ref (process-group-rec-members group)) - (process-group-rec-members-set! group - (cons ref (process-group-rec-members group)))))) - - (define (process-group-leave! group ref) - (with-mutex (process-group-rec-mutex group) - (process-group-rec-members-set! group - (filter (lambda (r) (not (equal? r ref))) - (process-group-rec-members group))))) - - (define (process-group-members group) - (with-mutex (process-group-rec-mutex group) - (list-copy (process-group-rec-members group)))) - - (define (process-group-broadcast! group msg) - (for-each (lambda (ref) (dsend ref msg)) - (process-group-members group))) - - ;; ====================================================================== - ;; Distributed supervision - ;; ====================================================================== - - (define-record-type dist-sup-rec - (fields - (mutable children) ;; list of (id node-hint actor-ref) - (immutable mutex)) - (sealed #t)) - - (define (make-dist-supervisor) - (make-dist-sup-rec '() (make-mutex))) - - ;; (dist-supervisor-start-child! sup id proc [node-hint]) - ;; node-hint: a node name (string) or #f for local - (define (dist-supervisor-start-child! sup id proc . rest) - (let* ([node-hint (if (null? rest) #f (car rest))] - [actor - (if (or (not node-hint) - (equal? node-hint (*cluster-name*))) - ;; Start locally - (spawn-actor proc) - ;; Simulate remote placement: start locally but tag with node - (spawn-actor proc))]) - (with-mutex (dist-sup-rec-mutex sup) - (dist-sup-rec-children-set! sup - (cons (list id node-hint actor) - (filter (lambda (c) (not (equal? (car c) id))) - (dist-sup-rec-children sup))))) - actor)) - - (define (dist-supervisor-children sup) - (with-mutex (dist-sup-rec-mutex sup) - (map (lambda (c) - (list (list-ref c 0) (list-ref c 1) (list-ref c 2))) - (dist-sup-rec-children sup)))) - - ;; ====================================================================== - ;; Node failure detection - ;; ====================================================================== - - ;; Monitor table: node-name -> list of callbacks - (define *node-monitors* (make-hashtable equal-hash equal?)) - (define *node-monitors-mutex* (make-mutex)) - - ;; (monitor-node node-name callback) - ;; callback is called with node-name when node is detected as down - (define (monitor-node node-name callback) - (with-mutex *node-monitors-mutex* - (hashtable-update! *node-monitors* node-name - (lambda (cbs) (cons callback cbs)) '()))) - - ;; (demonitor-node node-name callback) - (define (demonitor-node node-name callback) - (with-mutex *node-monitors-mutex* - (hashtable-update! *node-monitors* node-name - (lambda (cbs) (filter (lambda (c) (not (eq? c callback))) cbs)) - '()))) - - ;; (node-alive? target-name) -> #t/#f - ;; Uses cluster-node-by-name (which searches alive nodes). - ;; Returns #t if the node exists in the cluster and is alive. - (define (node-alive? target-name) - ;; cluster-node-by-name searches (cluster-nodes) which already - ;; filters to only alive nodes. So if found, it's alive. - (if (cluster-node-by-name target-name) #t #f)) - - ;; (ping-node node-name timeout-ms) -> 'ok or 'timeout - ;; Uses cluster membership as a proxy for connectivity. - (define (ping-node node-name timeout-ms) - (if (node-alive? node-name) - 'ok - 'timeout)) - - ;; Internal: notify monitors of a failed node - (define (%notify-node-failure! node-name) - (let ([cbs (with-mutex *node-monitors-mutex* - (hashtable-ref *node-monitors* node-name '()))]) - (for-each (lambda (cb) (cb node-name)) cbs))) - - ;; ====================================================================== - ;; Serialization - ;; ====================================================================== - - ;; Serialize a message to a bytevector using write. - ;; Only works for data that is writable/readable (no procedures, etc.) - (define (serialize-message msg) - (let ([port (open-output-string)]) - (write msg port) - (string->utf8 (get-output-string port)))) - - ;; Maximum allowed message size (bytes) for deserialization. - (define *max-message-size* (make-parameter (* 1 1024 1024))) ;; 1MB default - - ;; Deserialize a message from a bytevector. - ;; HARDENED: Disables read-eval (#. syntax) to prevent code execution - ;; during deserialization, and enforces message size limits. - (define (deserialize-message bv) - (when (> (bytevector-length bv) (*max-message-size*)) - (error 'deserialize-message - "message exceeds maximum allowed size" - (bytevector-length bv) (*max-message-size*))) - (let ([port (open-input-string (utf8->string bv))]) - (parameterize ([read-eval #f]) - (read port)))) - - ;; Hook into cluster leave events so monitors fire automatically. - ;; Must be after all define forms (it's an expression, not a definition). - (on-node-leave - (lambda (node) - (%notify-node-failure! (node-name node)))) - - ) ;; end library new file mode 100644 --- /dev/null +++ b/lib/std/actor/distributed.ss @@ -0,0 +1,311 @@ +#!chezscheme +;;; (std actor distributed) — Location-transparent distributed actor messaging +;;; +;;; Builds on (std actor core) and (std actor cluster) to provide: +;;; - Location-transparent send (dsend / dsend/ask) +;;; - Cluster-wide name registration +;;; - Process groups (broadcast) +;;; - Distributed supervision +;;; - Node failure detection / monitoring +;;; - Simple serialization via write/read on string ports + +(library (std actor distributed) + (export + ;; Location-transparent send + dsend + dsend/ask + + ;; Remote actor references + make-remote-ref + remote-ref? + remote-ref-node + remote-ref-id + + ;; Cluster-wide name registration + cluster-register! + cluster-whereis + cluster-registered-names + + ;; Process groups + make-process-group + process-group-join! + process-group-leave! + process-group-members + process-group-broadcast! + + ;; Distributed supervision + make-dist-supervisor + dist-supervisor-start-child! + dist-supervisor-children + + ;; Failure detection + monitor-node + demonitor-node + node-alive? + ping-node + + ;; Serialization + serialize-message + deserialize-message + + ;; Configuration parameters + *default-send-timeout* + *cluster-name* + *max-message-size*) + + (import (chezscheme) + (std actor core) + (except (std actor cluster) node-alive?) + (only (jerboa core) def defstruct)) + + ;; ====================================================================== + ;; Configuration parameters + ;; ====================================================================== + + (def *default-send-timeout* (make-parameter 5000)) ;; 5 seconds in ms + (def *cluster-name* (make-parameter "local")) + + ;; ====================================================================== + ;; Remote actor references + ;; ====================================================================== + + ;; A remote-ref identifies an actor on a specific cluster node by name. + (defstruct remote-ref-rec (node id)) + + (def (make-remote-ref node-name actor-id) + (make-remote-ref-rec node-name actor-id)) + + (def (remote-ref? x) (remote-ref-rec? x)) + (def (remote-ref-node x) (remote-ref-rec-node x)) + (def (remote-ref-id x) (remote-ref-rec-id x)) + + ;; Is actor-ref local? (uses (std actor core) actor-ref? predicate) + (def (%local-ref? ref) + (actor-ref? ref)) + + ;; ====================================================================== + ;; Location-transparent send + ;; ====================================================================== + + ;; (dsend actor-ref msg) + ;; Works for both local actor-refs and remote-refs. + ;; For local refs, delegates directly to (send). + ;; For remote refs, looks up the named actor on the target node. + (def (dsend ref msg) + (cond + [(%local-ref? ref) + ;; Local actor: use the core send + (send ref msg)] + [(remote-ref? ref) + ;; Remote: find actor in cluster registry + (let* ([node-name (remote-ref-node ref)] + [actor-id (remote-ref-id ref)] + [node (cluster-node-by-name node-name)]) + (if node + (let ([actor (remote-whereis node actor-id)]) + (if actor + (send actor msg) + (error 'dsend "actor not found on node" actor-id node-name))) + (error 'dsend "node not found" node-name)))] + [else + (error 'dsend "not a valid actor reference" ref)])) + + ;; (dsend/ask actor-ref msg timeout-ms) -> reply or #f + ;; Sends msg and waits for a reply; uses a temporary channel actor. + (def (dsend/ask ref msg timeout-ms) + (let* ([result-box (make-mutex)] + [reply #f] + [replied? #f] + [cond-var (make-condition)] + [lock (make-mutex)]) + ;; Spawn a one-shot reply actor + (let ([reply-actor + (spawn-actor + (lambda (m) + (with-mutex lock + (set! reply m) + (set! replied? #t) + (condition-signal cond-var))))]) + ;; Send original message with reply-to field prepended + (dsend ref (list 'ask reply-actor msg)) + ;; Wait for reply with timeout + (let ([deadline (+ (current-time-ms) timeout-ms)]) + (with-mutex lock + (let loop () + (unless replied? + (let ([now (current-time-ms)]) + (when (< now deadline) + (condition-wait cond-var lock) + (loop)))))) + (if replied? reply #f))))) + + (def (current-time-ms) + (* 1000 (time-second (current-time)))) + + ;; ====================================================================== + ;; Cluster-wide name registration + ;; ====================================================================== + + ;; Global name table: name -> actor-ref (local refs) + (def *global-registry* (make-hashtable equal-hash equal?)) + (def *global-registry-mutex* (make-mutex)) + + ;; (cluster-register! name actor-ref) — register actor under a cluster-wide name + (def (cluster-register! name actor-ref) + (with-mutex *global-registry-mutex* + (hashtable-set! *global-registry* name actor-ref)) + ;; Also register on all alive nodes in the cluster + (for-each + (lambda (node) + (remote-register! node name actor-ref)) + (cluster-nodes))) + + ;; (cluster-whereis name) -> actor-ref or #f (searches local registry first) + (def (cluster-whereis name) + (or (with-mutex *global-registry-mutex* + (hashtable-ref *global-registry* name #f)) + (whereis/any name))) + + ;; (cluster-registered-names) -> list of names + (def (cluster-registered-names) + (with-mutex *global-registry-mutex* + (vector->list (hashtable-keys *global-registry*)))) + + ;; ====================================================================== + ;; Process groups + ;; ====================================================================== + + (defstruct process-group-rec (name members mutex)) + + (def (make-process-group name) + (make-process-group-rec name '() (make-mutex))) + + (def (process-group-join! group ref) + (with-mutex (process-group-rec-mutex group) + (unless (member ref (process-group-rec-members group)) + (process-group-rec-members-set! group + (cons ref (process-group-rec-members group)))))) + + (def (process-group-leave! group ref) + (with-mutex (process-group-rec-mutex group) + (process-group-rec-members-set! group + (filter (lambda (r) (not (equal? r ref))) + (process-group-rec-members group))))) + + (def (process-group-members group) + (with-mutex (process-group-rec-mutex group) + (list-copy (process-group-rec-members group)))) + + (def (process-group-broadcast! group msg) + (for-each (lambda (ref) (dsend ref msg)) + (process-group-members group))) + + ;; ====================================================================== + ;; Distributed supervision + ;; ====================================================================== + + (defstruct dist-sup-rec (children mutex)) + + (def (make-dist-supervisor) + (make-dist-sup-rec '() (make-mutex))) + + ;; (dist-supervisor-start-child! sup id proc [node-hint]) + ;; node-hint: a node name (string) or #f for local + (def (dist-supervisor-start-child! sup id proc . rest) + (let* ([node-hint (if (null? rest) #f (car rest))] + [actor + (if (or (not node-hint) + (equal? node-hint (*cluster-name*))) + ;; Start locally + (spawn-actor proc) + ;; Simulate remote placement: start locally but tag with node + (spawn-actor proc))]) + (with-mutex (dist-sup-rec-mutex sup) + (dist-sup-rec-children-set! sup + (cons (list id node-hint actor) + (filter (lambda (c) (not (equal? (car c) id))) + (dist-sup-rec-children sup))))) + actor)) + + (def (dist-supervisor-children sup) + (with-mutex (dist-sup-rec-mutex sup) + (map (lambda (c) + (list (list-ref c 0) (list-ref c 1) (list-ref c 2))) + (dist-sup-rec-children sup)))) + + ;; ====================================================================== + ;; Node failure detection + ;; ====================================================================== + + ;; Monitor table: node-name -> list of callbacks + (def *node-monitors* (make-hashtable equal-hash equal?)) + (def *node-monitors-mutex* (make-mutex)) + + ;; (monitor-node node-name callback) + ;; callback is called with node-name when node is detected as down + (def (monitor-node node-name callback) + (with-mutex *node-monitors-mutex* + (hashtable-update! *node-monitors* node-name + (lambda (cbs) (cons callback cbs)) '()))) + + ;; (demonitor-node node-name callback) + (def (demonitor-node node-name callback) + (with-mutex *node-monitors-mutex* + (hashtable-update! *node-monitors* node-name + (lambda (cbs) (filter (lambda (c) (not (eq? c callback))) cbs)) + '()))) + + ;; (node-alive? target-name) -> #t/#f + ;; Uses cluster-node-by-name (which searches alive nodes). + ;; Returns #t if the node exists in the cluster and is alive. + (def (node-alive? target-name) + ;; cluster-node-by-name searches (cluster-nodes) which already + ;; filters to only alive nodes. So if found, it's alive. + (if (cluster-node-by-name target-name) #t #f)) + + ;; (ping-node node-name timeout-ms) -> 'ok or 'timeout + ;; Uses cluster membership as a proxy for connectivity. + (def (ping-node node-name timeout-ms) + (if (node-alive? node-name) + 'ok + 'timeout)) + + ;; Internal: notify monitors of a failed node + (def (%notify-node-failure! node-name) + (let ([cbs (with-mutex *node-monitors-mutex* + (hashtable-ref *node-monitors* node-name '()))]) + (for-each (lambda (cb) (cb node-name)) cbs))) + + ;; ====================================================================== + ;; Serialization + ;; ====================================================================== + + ;; Serialize a message to a bytevector using write. + ;; Only works for data that is writable/readable (no procedures, etc.) + (def (serialize-message msg) + (let ([port (open-output-string)]) + (write msg port) + (string->utf8 (get-output-string port)))) + + ;; Maximum allowed message size (bytes) for deserialization. + (def *max-message-size* (make-parameter (* 1 1024 1024))) ;; 1MB default + + ;; Deserialize a message from a bytevector. + ;; HARDENED: Disables read-eval (#. syntax) to prevent code execution + ;; during deserialization, and enforces message size limits. + (def (deserialize-message bv) + (when (> (bytevector-length bv) (*max-message-size*)) + (error 'deserialize-message + "message exceeds maximum allowed size" + (bytevector-length bv) (*max-message-size*))) + (let ([port (open-input-string (utf8->string bv))]) + (parameterize ([read-eval #f]) + (read port)))) + + ;; Hook into cluster leave events so monitors fire automatically. + ;; Must be after all define forms (it's an expression, not a definition). + (on-node-leave + (lambda (node) + (%notify-node-failure! (node-name node)))) + + ) ;; end library --- a/lib/std/actor/registry.ss +++ b/lib/std/actor/registry.ss @@ -14,7 +14,7 @@ registry-actor ;; → the registry actor-ref itself ) (import (chezscheme) - (only (jerboa core) match) + (only (jerboa core) match def) (std actor core) (std actor protocol) (only (jerboa core) def match)) deleted file mode 100644 --- a/lib/std/actor/supervisor.sls +++ /dev/null @@ -1,305 +0,0 @@ -#!chezscheme -;;; (std actor supervisor) — OTP-style supervision trees -;;; -;;; Strategies: one-for-one, one-for-all, rest-for-one -;;; Restart policies: permanent, transient, temporary -;;; Monitors child actors; escalates if restart intensity exceeded. - -(library (std actor supervisor) - (export - make-child-spec - child-spec? - child-spec-id - child-spec-start-thunk - child-spec-restart - child-spec-shutdown - child-spec-type - - start-supervisor - - supervisor-which-children - supervisor-count-children - supervisor-terminate-child! - supervisor-restart-child! - supervisor-start-child! - supervisor-delete-child! - ) - (import (chezscheme) - (only (jerboa core) match) - (std actor core) - (std actor protocol)) - - ;; -------- Child spec -------- - - (define-record-type child-spec - (fields - (immutable id) ;; symbol - (immutable start-thunk) ;; (lambda () → actor-ref) - (immutable restart) ;; 'permanent | 'transient | 'temporary - (immutable shutdown) ;; 'brutal-kill | number (seconds) - (immutable type)) ;; 'worker | 'supervisor - (sealed #t)) - - ;; -------- Runtime child entry -------- - - (define-record-type child-entry - (fields - (immutable spec) - (mutable actor-ref) ;; current actor-ref or #f - (mutable status)) ;; 'running | 'stopped | 'dead - (sealed #t)) - - ;; -------- Supervisor state (captured in behavior closure) -------- - - (define-record-type supervisor-state - (fields - (immutable strategy) ;; 'one-for-one | 'one-for-all | 'rest-for-one - (immutable max-restarts) - (immutable period-secs) - (mutable children) ;; ordered list of child-entry - (mutable restart-log)) ;; list of timestamps (floats) - (sealed #t)) - - ;; -------- Time helper (no SRFI-19 needed) -------- - - (define (current-seconds) - (let ([t (current-time)]) - (+ (time-second t) (/ (time-nanosecond t) 1e9)))) - - ;; -------- Supervisor startup -------- - - (define start-supervisor - (case-lambda - [(strategy child-specs) - (start-supervisor-impl strategy child-specs 10 5)] - [(strategy child-specs max-restarts) - (start-supervisor-impl strategy child-specs max-restarts 5)] - [(strategy child-specs max-restarts period-secs) - (start-supervisor-impl strategy child-specs max-restarts period-secs)])) - - (define (start-supervisor-impl strategy child-specs max-restarts period-secs) - (let ([state (make-supervisor-state strategy max-restarts period-secs '() '())]) - (let ([sup (spawn-actor - (lambda (msg) (supervisor-behavior state msg)) - 'supervisor)]) - (for-each (lambda (spec) (start-child! state sup spec)) child-specs) - sup))) - - ;; -------- Start a single child -------- - - (define (start-child! state sup spec) - (let* ([child ((child-spec-start-thunk spec))] - [entry (make-child-entry spec child 'running)]) - ;; Monitor: supervisor gets 'DOWN when child dies (one-way) - (actor-ref-monitors-set! child - (cons (cons sup (child-spec-id spec)) - (actor-ref-monitors child))) - (supervisor-state-children-set! state - (append (supervisor-state-children state) (list entry))) - entry)) - - ;; -------- Supervisor behavior -------- - - (define (supervisor-behavior state msg) - (with-ask-context msg - (lambda (actual) - (match actual - [('DOWN spec-id child-id reason) - (handle-child-exit! state spec-id child-id reason)] - - [('which-children) - (reply (format-children state))] - - [('terminate-child id) - (terminate-child-by-id! state id) - (reply 'ok)] - - [('restart-child id) - (reply (restart-child-by-id! state id))] - - [('start-child spec) - (let ([entry (start-child! state (self) spec)]) - (reply (child-entry-actor-ref entry)))] - - [('delete-child id) - (delete-child-by-id! state id) - (reply 'ok)] - - [_ (void)])))) - - ;; -------- Handle child exit -------- - - (define (handle-child-exit! state spec-id child-id reason) - (let ([entry (find-child-by-id state spec-id)]) - (when entry - (let ([spec (child-entry-spec entry)]) - (let ([should-restart? - (case (child-spec-restart spec) - [(permanent) #t] - [(transient) (not (memq reason '(normal killed)))] - [(temporary) #f] - [else #f])]) - (if should-restart? - (begin - (check-restart-intensity! state) - (case (supervisor-state-strategy state) - [(one-for-one) (restart-one! state entry)] - [(one-for-all) (restart-all! state)] - [(rest-for-one) (restart-rest! state entry)])) - (child-entry-status-set! entry 'dead))))))) - - ;; -------- Restart intensity -------- - - (define (check-restart-intensity! state) - (let* ([now (current-seconds)] - [period (supervisor-state-period-secs state)] - [recent (filter (lambda (t) (> t (- now period))) - (supervisor-state-restart-log state))]) - (supervisor-state-restart-log-set! state (cons now recent)) - (when (>= (length recent) (supervisor-state-max-restarts state)) - (error 'supervisor "restart intensity exceeded" - (supervisor-state-max-restarts state) - (supervisor-state-period-secs state))))) - - ;; -------- Restart strategies -------- - - ;; NOTE: restart-one!, restart-all!, restart-rest! are called only from - ;; handle-child-exit!, which runs inside the supervisor actor's behavior. - ;; (self) correctly returns the supervisor actor-ref in this context. - - (define (restart-one! state entry) - (stop-child-entry! entry) - (let* ([spec (child-entry-spec entry)] - [new-actor ((child-spec-start-thunk spec))]) - (child-entry-actor-ref-set! entry new-actor) - (child-entry-status-set! entry 'running) - (actor-ref-monitors-set! new-actor - (cons (cons (self) (child-spec-id spec)) - (actor-ref-monitors new-actor))))) - - (define (restart-all! state) - (let ([children (supervisor-state-children state)]) - (for-each stop-child-entry! (reverse children)) - (for-each - (lambda (entry) - (let* ([spec (child-entry-spec entry)] - [new-actor ((child-spec-start-thunk spec))]) - (child-entry-actor-ref-set! entry new-actor) - (child-entry-status-set! entry 'running) - (actor-ref-monitors-set! new-actor - (cons (cons (self) (child-spec-id spec)) - (actor-ref-monitors new-actor))))) - children))) - - (define (restart-rest! state failed-entry) - (let* ([children (supervisor-state-children state)] - [pos (let loop ([cs children] [i 0]) - (cond [(null? cs) -1] - [(eq? (car cs) failed-entry) i] - [else (loop (cdr cs) (fx+ i 1))]))] - [rest (if (fx>= pos 0) (list-tail children pos) '())]) - (for-each stop-child-entry! (reverse rest)) - (for-each - (lambda (entry) - (let* ([spec (child-entry-spec entry)] - [new-actor ((child-spec-start-thunk spec))]) - (child-entry-actor-ref-set! entry new-actor) - (child-entry-status-set! entry 'running) - (actor-ref-monitors-set! new-actor - (cons (cons (self) (child-spec-id spec)) - (actor-ref-monitors new-actor))))) - rest))) - - ;; -------- Stop a child -------- - - (define (stop-child-entry! entry) - (let ([a (child-entry-actor-ref entry)] - [shutdown (child-spec-shutdown (child-entry-spec entry))]) - (when (and a (actor-alive? a)) - ;; Remove our monitor BEFORE killing so the forced stop does not - ;; deliver a DOWN message back to this supervisor and trigger a restart. - (actor-ref-monitors-set! a - (filter (lambda (mon) (not (eq? (car mon) (self)))) - (actor-ref-monitors a))) - (cond - [(eq? shutdown 'brutal-kill) - (actor-kill! a)] - [(number? shutdown) - ;; Graceful: send 'shutdown, wait, then force-kill - (guard (exn [#t (void)]) - (send a '(shutdown))) - (let ([deadline (+ (current-seconds) shutdown)]) - (let loop () - (cond - [(not (actor-alive? a)) (void)] - [(>= (current-seconds) deadline) (actor-kill! a)] - [else - (sleep (make-time 'time-duration 20000000 0)) ;; 20ms - (loop)])))])) - (child-entry-actor-ref-set! entry #f) - (child-entry-status-set! entry 'stopped))) - - ;; -------- Dynamic child management -------- - - (define (terminate-child-by-id! state id) - (let ([entry (find-child-by-id state id)]) - (when entry (stop-child-entry! entry)))) - - (define (restart-child-by-id! state id) - (let ([entry (find-child-by-id state id)]) - (if (and entry (eq? (child-entry-status entry) 'stopped)) - (begin (restart-one! state entry) 'ok) - 'not-found))) - - (define (delete-child-by-id! state id) - (let ([entry (find-child-by-id state id)]) - (when entry - (stop-child-entry! entry) - (supervisor-state-children-set! state - (filter (lambda (e) (not (eq? e entry))) - (supervisor-state-children state)))))) - - ;; -------- Public management API -------- - ;; Called from outside the supervisor actor via ask-sync. - - (define (supervisor-which-children sup) - (ask-sync sup '(which-children))) - - (define (supervisor-count-children sup) - (let ([children (supervisor-which-children sup)]) - (let loop ([cs children] [total 0] [active 0]) - (if (null? cs) - (values total active) - (loop (cdr cs) - (fx+ total 1) - (if (eq? (cadr (car cs)) 'running) (fx+ active 1) active)))))) - - (define (supervisor-terminate-child! sup id) - (ask-sync sup (list 'terminate-child id))) - - (define (supervisor-restart-child! sup id) - (ask-sync sup (list 'restart-child id))) - - (define (supervisor-start-child! sup spec) - (ask-sync sup (list 'start-child spec))) - - (define (supervisor-delete-child! sup id) - (ask-sync sup (list 'delete-child id))) - - ;; -------- Helpers -------- - - (define (find-child-by-id state id) - (let loop ([cs (supervisor-state-children state)]) - (cond - [(null? cs) #f] - [(eq? (child-spec-id (child-entry-spec (car cs))) id) (car cs)] - [else (loop (cdr cs))]))) - - (define (format-children state) - (map (lambda (entry) - (list (child-spec-id (child-entry-spec entry)) - (child-entry-status entry) - (child-entry-actor-ref entry))) - (supervisor-state-children state))) - - ) ;; end library new file mode 100644 --- /dev/null +++ b/lib/std/actor/supervisor.ss @@ -0,0 +1,286 @@ +#!chezscheme +;;; (std actor supervisor) — OTP-style supervision trees +;;; +;;; Strategies: one-for-one, one-for-all, rest-for-one +;;; Restart policies: permanent, transient, temporary +;;; Monitors child actors; escalates if restart intensity exceeded. + +(library (std actor supervisor) + (export + make-child-spec + child-spec? + child-spec-id + child-spec-start-thunk + child-spec-restart + child-spec-shutdown + child-spec-type + + start-supervisor + + supervisor-which-children + supervisor-count-children + supervisor-terminate-child! + supervisor-restart-child!