Phase 2d complete: Systems & Distributed (6 libraries, 105 tests passing)
ober
71b0c43fd678e830d4f330d9fd5bcdff854b1f17
new file mode 100644 --- /dev/null +++ b/lib/std/net/pool.sls @@ -0,0 +1,289 @@ +#!chezscheme +;;; (std net pool) — Generic Connection Pool +;;; +;;; Manages a pool of connection objects with: +;;; - Configurable min/max pool size +;;; - Idle timeout and health checking +;;; - Blocking acquire with timeout +;;; - Statistics tracking +;;; +;;; A "connection" is any opaque object. The pool is parameterized by: +;;; factory — (lambda () conn) — creates a new connection +;;; closer — (lambda (conn) ...) — destroys a connection +;;; checker — (lambda (conn) bool) — health check (returns #t if healthy) +;;; +;;; API: +;;; (make-connection-pool factory closer checker [min-size [max-size]]) +;;; (pool-acquire! pool [timeout-ms]) → connection +;;; (pool-release! pool conn) +;;; (pool-close! pool) +;;; (pool-size pool) → total connections (available + in-use) +;;; (pool-available pool) → available connection count +;;; (pool-stats pool) → alist of statistics +;;; (with-connection pool proc) → calls proc with conn, releases after +;;; (pool-health-check! pool) → validate all idle connections + +(library (std net pool) + (export + make-connection-pool + connection-pool? + pool-acquire! + pool-release! + pool-close! + pool-size + pool-available + pool-stats + with-connection + pool-health-check!) + + (import (chezscheme)) + + ;; ========== Pool Entry ========== + + (define-record-type pool-entry + (fields + conn ;; the connection object + (mutable healthy?) ;; last known health + (mutable created-at) ;; creation timestamp + (mutable used-at)) ;; last use timestamp + (protocol + (lambda (new) + (lambda (conn) + (let ([now (time-second (current-time 'time-monotonic))]) + (new conn #t now now)))))) + + ;; ========== Connection Pool ========== + + (define-record-type (connection-pool %make-connection-pool connection-pool?) + (fields + factory ;; (lambda () conn) + closer ;; (lambda (conn) ...) + checker ;; (lambda (conn) bool) or #f + min-size ;; minimum idle connections + max-size ;; maximum total connections + (mutable idle) ;; list of pool-entry (available) + (mutable in-use) ;; list of pool-entry (checked out) + (mutable total) ;; current total count + (mutable closed?) + mutex + not-empty ;; condition: conn returned to pool + ;; Stats + (mutable stats-acquired) + (mutable stats-released) + (mutable stats-created) + (mutable stats-destroyed) + (mutable stats-health-failures) + (mutable stats-waits))) + + (define make-connection-pool + (case-lambda + [(factory closer checker) + (make-connection-pool factory closer checker 1 10)] + [(factory closer checker min-size max-size) + (let ([pool (%make-connection-pool + factory closer checker + min-size max-size + '() '() 0 #f + (make-mutex) (make-condition) + 0 0 0 0 0 0)]) + ;; Pre-warm with min-size connections + (let warm ([i 0]) + (when (< i min-size) + (%create-connection! pool) + (warm (+ i 1)))) + pool)])) + + ;; Internal: create a new connection and add to idle + (define (%create-connection! pool) + (let ([conn ((connection-pool-factory pool))]) + (let ([entry (make-pool-entry conn)]) + (connection-pool-idle-set! pool + (cons entry (connection-pool-idle pool))) + (connection-pool-total-set! pool + (+ (connection-pool-total pool) 1)) + (connection-pool-stats-created-set! pool + (+ (connection-pool-stats-created pool) 1)) + entry))) + + (define pool-acquire! + (case-lambda + [(pool) (pool-acquire! pool #f)] + [(pool timeout-ms) + (mutex-acquire (connection-pool-mutex pool)) + (when (connection-pool-closed? pool) + (mutex-release (connection-pool-mutex pool)) + (error 'pool-acquire! "pool is closed")) + (let try () + (cond + ;; Idle connection available + [(pair? (connection-pool-idle pool)) + (let ([entry (car (connection-pool-idle pool))]) + (connection-pool-idle-set! pool + (cdr (connection-pool-idle pool))) + (connection-pool-in-use-set! pool + (cons entry (connection-pool-in-use pool))) + (pool-entry-used-at-set! entry + (time-second (current-time 'time-monotonic))) + (connection-pool-stats-acquired-set! pool + (+ (connection-pool-stats-acquired pool) 1)) + (mutex-release (connection-pool-mutex pool)) + (pool-entry-conn entry))] + ;; Can create new connection + [(< (connection-pool-total pool) (connection-pool-max-size pool)) + (let ([entry (%create-connection! pool)]) + (connection-pool-idle-set! pool + (cdr (connection-pool-idle pool))) ;; remove from idle + (connection-pool-in-use-set! pool + (cons entry (connection-pool-in-use pool))) + (connection-pool-stats-acquired-set! pool + (+ (connection-pool-stats-acquired pool) 1)) + (mutex-release (connection-pool-mutex pool)) + (pool-entry-conn entry))] + ;; Must wait + [else + (connection-pool-stats-waits-set! pool + (+ (connection-pool-stats-waits pool) 1)) + (if timeout-ms + (let* ([ns (* timeout-ms 1000000)] + [s (quotient ns 1000000000)] + [ns-part (remainder ns 1000000000)]) + (condition-wait (connection-pool-not-empty pool) + (connection-pool-mutex pool) + (make-time 'time-duration ns-part s)) + ;; Check if still nothing available → timeout + (if (and (null? (connection-pool-idle pool)) + (>= (connection-pool-total pool) + (connection-pool-max-size pool))) + (begin + (mutex-release (connection-pool-mutex pool)) + (error 'pool-acquire! "timeout waiting for connection")) + (try))) + (begin + (condition-wait (connection-pool-not-empty pool) + (connection-pool-mutex pool)) + (try)))]))])) + + (define (pool-release! pool conn) + (mutex-acquire (connection-pool-mutex pool)) + ;; Find the entry + (let loop ([entries (connection-pool-in-use pool)] [rest '()]) + (cond + [(null? entries) + ;; Not found — ignore + (mutex-release (connection-pool-mutex pool))] + [(eq? (pool-entry-conn (car entries)) conn) + ;; Found: move back to idle (or close if pool is full/closed) + (let ([entry (car entries)]) + (connection-pool-in-use-set! pool + (append (reverse rest) (cdr entries))) + (cond + [(connection-pool-closed? pool) + ;; Pool closed: destroy connection + (guard (exn [#t (void)]) + ((connection-pool-closer pool) conn)) + (connection-pool-total-set! pool + (- (connection-pool-total pool) 1)) + (connection-pool-stats-destroyed-set! pool + (+ (connection-pool-stats-destroyed pool) 1))] + [else + ;; Return to idle + (connection-pool-idle-set! pool + (cons entry (connection-pool-idle pool))) + (connection-pool-stats-released-set! pool + (+ (connection-pool-stats-released pool) 1)) + (condition-signal (connection-pool-not-empty pool))])) + (mutex-release (connection-pool-mutex pool))] + [else + (loop (cdr entries) (cons (car entries) rest))]))) + + (define (pool-size pool) + (mutex-acquire (connection-pool-mutex pool)) + (let ([n (connection-pool-total pool)]) + (mutex-release (connection-pool-mutex pool)) + n)) + + (define (pool-available pool) + (mutex-acquire (connection-pool-mutex pool)) + (let ([n (length (connection-pool-idle pool))]) + (mutex-release (connection-pool-mutex pool)) + n)) + + (define (pool-stats pool) + (mutex-acquire (connection-pool-mutex pool)) + (let ([stats + (list + (cons 'total (connection-pool-total pool)) + (cons 'idle (length (connection-pool-idle pool))) + (cons 'in-use (length (connection-pool-in-use pool))) + (cons 'acquired (connection-pool-stats-acquired pool)) + (cons 'released (connection-pool-stats-released pool)) + (cons 'created (connection-pool-stats-created pool)) + (cons 'destroyed (connection-pool-stats-destroyed pool)) + (cons 'health-failures (connection-pool-stats-health-failures pool)) + (cons 'waits (connection-pool-stats-waits pool)))]) + (mutex-release (connection-pool-mutex pool)) + stats)) + + (define (pool-close! pool) + (mutex-acquire (connection-pool-mutex pool)) + (connection-pool-closed?-set! pool #t) + ;; Close all idle connections + (for-each + (lambda (entry) + (guard (exn [#t (void)]) + ((connection-pool-closer pool) (pool-entry-conn entry))) + (connection-pool-stats-destroyed-set! pool + (+ (connection-pool-stats-destroyed pool) 1))) + (connection-pool-idle pool)) + (connection-pool-idle-set! pool '()) + (connection-pool-total-set! pool + (length (connection-pool-in-use pool))) + ;; Signal any waiters so they can get the error + (condition-broadcast (connection-pool-not-empty pool)) + (mutex-release (connection-pool-mutex pool))) + + (define (pool-health-check! pool) + ;; Check all idle connections; remove unhealthy ones + (let ([checker (connection-pool-checker pool)]) + (when checker + (mutex-acquire (connection-pool-mutex pool)) + (let ([healthy '()] [bad '()]) + (for-each + (lambda (entry) + (let ([ok? + (guard (exn [#t #f]) + (checker (pool-entry-conn entry)))]) + (if ok? + (begin + (pool-entry-healthy?-set! entry #t) + (set! healthy (cons entry healthy))) + (begin + (pool-entry-healthy?-set! entry #f) + (set! bad (cons entry bad)) + (connection-pool-stats-health-failures-set! pool + (+ (connection-pool-stats-health-failures pool) 1)))))) + (connection-pool-idle pool)) + ;; Close unhealthy + (for-each + (lambda (entry) + (guard (exn [#t (void)]) + ((connection-pool-closer pool) (pool-entry-conn entry))) + (connection-pool-total-set! pool + (- (connection-pool-total pool) 1)) + (connection-pool-stats-destroyed-set! pool + (+ (connection-pool-stats-destroyed pool) 1))) + bad) + ;; Keep healthy + (connection-pool-idle-set! pool (reverse healthy))) + (mutex-release (connection-pool-mutex pool))))) + + (define (with-connection pool proc) + (let ([conn (pool-acquire! pool)]) + (dynamic-wind + (lambda () (void)) + (lambda () (proc conn)) + (lambda () + (guard (exn [#t (void)]) + (pool-release! pool conn)))))) + +) ;; end library new file mode 100644 --- /dev/null +++ b/lib/std/net/zero-copy.sls @@ -0,0 +1,180 @@ +#!chezscheme +;;; (std net zero-copy) — Zero-Copy Buffer Management +;;; +;;; Pool of pre-allocated bytevectors. Slices are views into those buffers +;;; (offset + length), avoiding data copies. Reference counting tracks +;;; live slices so buffers can be safely returned to the pool. +;;; +;;; API: +;;; (make-buffer-pool size count) — pool of `count` buffers of `size` bytes +;;; (pool-acquire! pool) — get a buffer (blocks if none available) +;;; (pool-release! pool buf-id) — return buffer to pool +;;; (make-buffer-slice buf-id offset length pool) — create a slice view +;;; (slice-data slice) — the underlying bytevector +;;; (slice-offset slice) — offset into buffer +;;; (slice-length slice) — length of this slice +;;; (slice-copy! dst dst-offset slice) — copy slice into another bytevector +;;; (buffer-pool-stats pool) — returns alist of stats +;;; (with-buffer pool proc) — acquire, call proc with buffer, release +;;; (slice->bytevector slice) — copy slice contents to fresh bytevector + +(library (std net zero-copy) + (export + make-buffer-pool + buffer-pool? + pool-acquire! + pool-release! + make-buffer-slice + buffer-slice? + slice-data + slice-offset + slice-length + slice-copy! + buffer-pool-stats + with-buffer + slice->bytevector) + + (import (chezscheme)) + + ;; ========== Buffer Entry ========== + ;; Each slot in the pool is tracked with a refcount. + + (define-record-type buffer-entry + (fields + id ;; integer id + data ;; bytevector + (mutable refcount) ;; number of live slices + (mutable in-use?)) ;; currently acquired by a consumer + (protocol + (lambda (new) + (lambda (id data) + (new id data 0 #f))))) + + ;; ========== Buffer Pool ========== + + (define-record-type (buffer-pool %make-buffer-pool buffer-pool?) + (fields + buf-size ;; bytes per buffer + entries ;; vector of buffer-entry + (mutable available) ;; list of free entry ids + mutex + not-empty ;; condition: a buffer became available + (mutable stats-acquired) + (mutable stats-released) + (mutable stats-waits))) + + (define (make-buffer-pool buf-size count) + (let* ([entries (let build ([i 0] [acc '()]) + (if (= i count) + (list->vector (reverse acc)) + (build (+ i 1) + (cons (make-buffer-entry i (make-bytevector buf-size 0)) + acc))))] + [available (let build ([i 0] [acc '()]) + (if (= i count) (reverse acc) (build (+ i 1) (cons i acc))))]) + (%make-buffer-pool + buf-size entries available + (make-mutex) (make-condition) + 0 0 0))) + + (define (pool-acquire! pool) + ;; Returns (values buf-id bytevector) + (mutex-acquire (buffer-pool-mutex pool)) + (let loop () + (cond + [(pair? (buffer-pool-available pool)) + (let* ([id (car (buffer-pool-available pool))] + [entry (vector-ref (buffer-pool-entries pool) id)]) + (buffer-pool-available-set! pool (cdr (buffer-pool-available pool))) + (buffer-entry-in-use?-set! entry #t) + (buffer-pool-stats-acquired-set! pool + (+ (buffer-pool-stats-acquired pool) 1)) + (mutex-release (buffer-pool-mutex pool)) + (values id (buffer-entry-data entry)))] + [else + (buffer-pool-stats-waits-set! pool + (+ (buffer-pool-stats-waits pool) 1)) + (condition-wait (buffer-pool-not-empty pool) + (buffer-pool-mutex pool)) + (loop)]))) + + (define (pool-release! pool buf-id) + (mutex-acquire (buffer-pool-mutex pool)) + (let ([entry (vector-ref (buffer-pool-entries pool) buf-id)]) + (when (= (buffer-entry-refcount entry) 0) + (buffer-entry-in-use?-set! entry #f) + (buffer-pool-available-set! pool + (cons buf-id (buffer-pool-available pool))) + (buffer-pool-stats-released-set! pool + (+ (buffer-pool-stats-released pool) 1)) + (condition-signal (buffer-pool-not-empty pool)))) + (mutex-release (buffer-pool-mutex pool))) + + (define (buffer-pool-stats pool) + (mutex-acquire (buffer-pool-mutex pool)) + (let ([stats + (list + (cons 'buf-size (buffer-pool-buf-size pool)) + (cons 'total (vector-length (buffer-pool-entries pool))) + (cons 'available (length (buffer-pool-available pool))) + (cons 'acquired (buffer-pool-stats-acquired pool)) + (cons 'released (buffer-pool-stats-released pool)) + (cons 'waits (buffer-pool-stats-waits pool)))]) + (mutex-release (buffer-pool-mutex pool)) + stats)) + + ;; ========== Buffer Slice ========== + ;; A slice is a view into a buffer: no data copied. + + (define-record-type (buffer-slice %make-buffer-slice buffer-slice?) + (fields + buf-id ;; which buffer + data ;; direct reference to bytevector + offset ;; starting offset + length ;; number of bytes in this slice + pool)) ;; owning pool (for release) + + (define (make-buffer-slice buf-id offset length pool) + ;; Increment refcount + (let ([entry (vector-ref (buffer-pool-entries pool) buf-id)]) + (mutex-acquire (buffer-pool-mutex pool)) + (buffer-entry-refcount-set! entry (+ (buffer-entry-refcount entry) 1)) + (mutex-release (buffer-pool-mutex pool))) + (%make-buffer-slice buf-id + (buffer-entry-data + (vector-ref (buffer-pool-entries pool) buf-id)) + offset length pool)) + + ;; Accessors with renamed field names to avoid confusion + (define (slice-data slice) (buffer-slice-data slice)) + (define (slice-offset slice) (buffer-slice-offset slice)) + (define (slice-length slice) (buffer-slice-length slice)) + + (define (slice-copy! dst dst-offset slice) + ;; Copy slice contents into dst bytevector at dst-offset + (let ([src (buffer-slice-data slice)] + [src-off (buffer-slice-offset slice)] + [len (buffer-slice-length slice)]) + (let loop ([i 0]) + (when (< i len) + (bytevector-u8-set! dst (+ dst-offset i) + (bytevector-u8-ref src (+ src-off i))) + (loop (+ i 1)))))) + + (define (slice->bytevector slice) + (let* ([len (buffer-slice-length slice)] + [bv (make-bytevector len)]) + (slice-copy! bv 0 slice) + bv)) + + ;; ========== with-buffer ========== + + (define (with-buffer pool proc) + ;; Acquire a buffer, call proc with (buf-id bv), always release + (let-values ([(buf-id bv) (pool-acquire! pool)]) + (dynamic-wind + (lambda () (void)) + (lambda () (proc buf-id bv)) + (lambda () (pool-release! pool buf-id))))) + +) ;; end library new file mode 100644 --- /dev/null +++ b/lib/std/proc/supervisor.sls @@ -0,0 +1,282 @@ +#!chezscheme +;;; (std proc supervisor) — OTP-Style Process Supervisor +;;; +;;; Monitors child threads. When a child dies unexpectedly, applies a +;;; restart strategy. Three strategies: +;;; one-for-one — restart only the failed child +;;; one-for-all — restart all children when any fails +;;; rest-for-one — restart the failed child + all started after it +;;; +;;; Child specs describe how to start/restart a child. +;;; +;;; API: +;;; (child-spec id thunk [restart-type [max-restarts [restart-window]]]) +;;; (make-supervisor strategy [max-restarts [restart-window]]) +;;; (supervisor-start-child! sup spec) +;;; (supervisor-stop-child! sup id) +;;; (supervisor-restart-child! sup id) +;;; (supervisor-children sup) → list of child-info +;;; (supervisor-run! sup) → starts supervisor monitoring loop +;;; (supervisor-stop! sup) → stops supervisor and all children +;;; (one-for-one) (one-for-all) (rest-for-one) → strategy constants + +(library (std proc supervisor) + (export + make-supervisor + supervisor? + supervisor-running? + supervisor-start-child! + supervisor-stop-child! + supervisor-restart-child! + supervisor-children + supervisor-run! + supervisor-stop! + child-spec + child-spec? + child-spec-id + child-spec-thunk + child-spec-restart-type + child-spec-max-restarts + child-spec-restart-window + one-for-one + one-for-all + rest-for-one) + + (import (chezscheme) (std misc channel)) + + ;; ========== Strategy Constants ========== + + (define one-for-one 'one-for-one) + (define one-for-all 'one-for-all) + (define rest-for-one 'rest-for-one) + + ;; ========== Child Spec ========== + + (define-record-type (%child-spec %make-child-spec child-spec?) + (fields + (immutable id child-spec-id) + (immutable thunk child-spec-thunk) + (immutable restart-type child-spec-restart-type) + (immutable max-restarts child-spec-max-restarts) + (immutable restart-window child-spec-restart-window))) + + ;; Public constructor with defaults + (define child-spec + (case-lambda + [(id thunk) + (%make-child-spec id thunk 'permanent 3 60)] + [(id thunk restart-type) + (%make-child-spec id thunk restart-type 3 60)] + [(id thunk restart-type max-restarts restart-window) + (%make-child-spec id thunk restart-type max-restarts restart-window)])) + + ;; ========== Child Info (runtime state) ========== + + (define-record-type child-info + (fields + spec ;; child-spec-rec + (mutable thread-id) ;; Chez thread id (from fork-thread) + (mutable status) ;; 'running | 'stopped | 'failed + (mutable restart-count) ;; how many times restarted + (mutable last-restart) ;; timestamp of last restart + done-mutex ;; for join-like behavior + done-cond + (mutable exit-value) ;; #f or exn on failure + (mutable alive?)) ;; thread still running? + (protocol + (lambda (new) + (lambda (spec) + (new spec #f 'stopped 0 0 + (make-mutex) (make-condition) + #f #f))))) + + ;; ========== Supervisor ========== + + (define-record-type (supervisor %make-supervisor supervisor?) + (fields + strategy ;; 'one-for-one | 'one-for-all | 'rest-for-one + max-restarts ;; max restarts in window before crash + restart-window ;; window in seconds + (mutable child-list) ;; list of child-info in start order + (mutable running?) + mutex ;; protects children list + monitor-ch)) ;; channel for death notifications + + (define (make-supervisor strategy . opts) + (let ([max-r (if (pair? opts) (car opts) 10)] + [window (if (and (pair? opts) (pair? (cdr opts))) (cadr opts) 60)]) + (%make-supervisor strategy max-r window + '() #f (make-mutex) + ;; We use a simple channel for exit notifications + (make-channel 64)))) + + ;; ========== Starting Children ========== + + (define (start-child-thread! sup info) + (let* ([spec (child-info-spec info)] + [thunk (child-spec-thunk spec)] + [mon-ch (supervisor-monitor-ch sup)]) + (child-info-status-set! info 'running) + (child-info-alive?-set! info #t) + (fork-thread + (lambda () + (let ([result + (call-with-current-continuation + (lambda (k) + (with-exception-handler + (lambda (exn) + (k (cons 'failed exn))) + (lambda () + (thunk) + (cons 'exited #f)))))]) + (child-info-alive?-set! info #f) + (child-info-exit-value-set! info (cdr result)) + ;; Notify supervisor + (channel-put mon-ch (cons info (car result)))))))) + + (define (supervisor-start-child! sup spec) + (let ([info (make-child-info spec)]) + (mutex-acquire (supervisor-mutex sup)) + (supervisor-child-list-set! sup + (append (supervisor-child-list sup) (list info))) + (mutex-release (supervisor-mutex sup)) + (start-child-thread! sup info) + info)) + + ;; ========== Stopping Children ========== + + (define (stop-child-thread! info) + ;; We can't kill threads in Chez, but we mark them stopped. + ;; The thread will eventually finish on its own. + ;; In production, use thread interrupts or a stop channel per child. + (child-info-status-set! info 'stopped) + (child-info-alive?-set! info #f)) + + (define (supervisor-stop-child! sup id) + (mutex-acquire (supervisor-mutex sup)) + (let ([info (find-child sup id)]) + (when info (stop-child-thread! info))) + (mutex-release (supervisor-mutex sup))) + + (define (find-child sup id) + (let loop ([children (supervisor-child-list sup)]) + (cond + [(null? children) #f] + [(equal? (child-spec-id (child-info-spec (car children))) id) + (car children)] + [else (loop (cdr children))]))) + + ;; ========== Restarting ========== + + (define (maybe-restart! sup info exit-type) + (let* ([spec (child-info-spec info)] + [restart-type (child-spec-restart-type spec)]) + (cond + ;; temporary: never restart + [(eq? restart-type 'temporary) + (child-info-status-set! info 'stopped)] + ;; transient: only restart on failure + [(and (eq? restart-type 'transient) (eq? exit-type 'exited)) + (child-info-status-set! info 'stopped)] + ;; permanent or transient+failed: restart + [else + (let ([count (child-info-restart-count info)]) + (if (>= count (child-spec-max-restarts spec)) + ;; Too many restarts — give up + (begin + (child-info-status-set! info 'failed) + (when (supervisor-running? sup) + (display + (string-append "supervisor: child " + (if (symbol? (child-spec-id spec)) + (symbol->string (child-spec-id spec)) + (child-spec-id spec)) + " exceeded max restarts, not restarting\n")))) + ;; Restart + (begin + (child-info-restart-count-set! info (+ count 1)) + (child-info-last-restart-set! info + (time-second (current-time 'time-monotonic))) + (start-child-thread! sup info))))]))) + + (define (apply-strategy! sup failed-info exit-type) + (let ([strategy (supervisor-strategy sup)]) + (cond + [(eq? strategy one-for-one) + (maybe-restart! sup failed-info exit-type)] + [(eq? strategy one-for-all) + ;; Stop all, restart all + (for-each + (lambda (info) + (unless (eq? info failed-info) + (stop-child-thread! info))) + (supervisor-child-list sup)) + (for-each + (lambda (info) + (maybe-restart! sup info exit-type)) + (supervisor-child-list sup))] + [(eq? strategy rest-for-one) + ;; Find position of failed child, restart it + all after it + (let ([children (supervisor-child-list sup)]) + (let loop ([remaining children] [found? #f]) + (unless (null? remaining) + (let ([info (car remaining)]) + (if (or found? (eq? info failed-info)) + (begin + (unless (eq? info failed-info) + (stop-child-thread! info)) + (maybe-restart! sup info exit-type) + (loop (cdr remaining) #t)) + (loop (cdr remaining) #f))))))]))) + + ;; ========== Restart ========== + + (define (supervisor-restart-child! sup id) + (mutex-acquire (supervisor-mutex sup)) + (let ([info (find-child sup id)]) + (when info + (start-child-thread! sup info))) + (mutex-release (supervisor-mutex sup))) + + ;; ========== Children Inspection ========== + + (define (supervisor-children sup) + (supervisor-child-list sup)) + + ;; ========== Supervisor Lifecycle ========== + + (define (supervisor-run! sup) + (supervisor-running?-set! sup #t) + ;; Monitor loop runs in a thread + (fork-thread + (lambda () + (let loop () + (when (supervisor-running? sup) + (let ([notification + ;; Wait for a child exit notification + (let wait () + (let-values ([(val ok) + (channel-try-get (supervisor-monitor-ch sup))]) + (if ok + val + (begin + ;; Poll with small sleep + (sleep (make-time 'time-duration 10000000 0)) + (wait)))))]) + (when notification + (let ([info (car notification)] + [exit-type (cdr notification)]) + (when (supervisor-running? sup) + (mutex-acquire (supervisor-mutex sup)) + (apply-strategy! sup info exit-type) + (mutex-release (supervisor-mutex sup)))))) + (loop))))) + sup) + + (define (supervisor-stop! sup) + (supervisor-running?-set! sup #f) + (mutex-acquire (supervisor-mutex sup)) + (for-each stop-child-thread! (supervisor-child-list sup)) + (mutex-release (supervisor-mutex sup))) + +) ;; end library new file mode 100644 --- /dev/null +++ b/lib/std/raft.sls @@ -0,0 +1,544 @@ +#!chezscheme +;;; (std raft) — Raft Consensus Algorithm (In-Memory Simulation) +;;; +;;; Full Raft state machine: follower/candidate/leader states. +;;; Nodes communicate via Scheme channels (in-process simulation). +;;; No network I/O — designed to be connected to a transport layer. +;;; +;;; Raft summary: +;;; - Nodes start as followers with random election timeouts +;;; - Follower → Candidate when election timeout fires +;;; - Candidate requests votes; wins if majority +;;; - Leader sends heartbeats to prevent election timeouts +;;; - Log entries replicated from leader to followers +;;; - Commit when majority acknowledge + +(library (std raft) + (export + make-raft-node + raft-start! + raft-stop! + raft-propose! + raft-leader? + raft-term + raft-state + raft-log + raft-commit-index + make-raft-cluster + raft-cluster-nodes + raft-cluster-leader) + + (import (chezscheme) (std misc channel)) + + ;; ========== Log Entry ========== + + (define-record-type log-entry + (fields index term command) + (protocol + (lambda (new) + (lambda (index term command) + (new index term command))))) + + ;; ========== Message Types ========== + ;; Messages sent between nodes via channels. + + ;; RequestVote RPC: candidate → all nodes + (define (make-vote-request term candidate-id last-log-index last-log-term) + (vector 'request-vote term candidate-id last-log-index last-log-term)) + (define (vote-request? msg) (and (vector? msg) (eq? (vector-ref msg 0) 'request-vote))) + (define (vote-request-term msg) (vector-ref msg 1)) + (define (vote-request-candidate-id msg) (vector-ref msg 2)) + (define (vote-request-last-log-index msg) (vector-ref msg 3)) + (define (vote-request-last-log-term msg) (vector-ref msg 4)) + + ;; VoteResponse: node → candidate + (define (make-vote-response term granted? voter-id) + (vector 'vote-response term granted? voter-id)) + (define (vote-response? msg) (and (vector? msg) (eq? (vector-ref msg 0) 'vote-response))) + (define (vote-response-term msg) (vector-ref msg 1)) + (define (vote-response-granted? msg) (vector-ref msg 2)) + + ;; AppendEntries RPC: leader → followers (also used as heartbeat) + (define (make-append-entries term leader-id prev-log-index prev-log-term entries commit-index) + (vector 'append-entries term leader-id prev-log-index prev-log-term entries commit-index)) + (define (append-entries? msg) (and (vector? msg) (eq? (vector-ref msg 0) 'append-entries))) + (define (append-entries-term msg) (vector-ref msg 1)) + (define (append-entries-leader-id msg) (vector-ref msg 2)) + (define (append-entries-prev-log-index msg) (vector-ref msg 3)) + (define (append-entries-prev-log-term msg) (vector-ref msg 4)) + (define (append-entries-entries msg) (vector-ref msg 5)) + (define (append-entries-commit-index msg) (vector-ref msg 6)) + + ;; AppendEntriesResponse + (define (make-append-response term success? follower-id match-index) + (vector 'append-response term success? follower-id match-index)) + (define (append-response? msg) (and (vector? msg) (eq? (vector-ref msg 0) 'append-response))) + (define (append-response-term msg) (vector-ref msg 1)) + (define (append-response-success? msg) (vector-ref msg 2)) + (define (append-response-follower-id msg) (vector-ref msg 3)) + (define (append-response-match-index msg) (vector-ref msg 4)) + + ;; ClientPropose: client → leader + (define (make-client-propose command reply-ch) + (vector 'client-propose command reply-ch)) + (define (client-propose? msg) (and (vector? msg) (eq? (vector-ref msg 0) 'client-propose))) + (define (client-propose-command msg) (vector-ref msg 1)) + (define (client-propose-reply-ch msg) (vector-ref msg 2)) + + ;; Stop signal + (define %stop-signal (list 'stop)) + + ;; ========== Raft Node ========== + + (define-record-type (raft-node %make-raft-node raft-node?) + (fields + id ;; node identifier (symbol/number) + (mutable current-term) ;; current term + (mutable voted-for) ;; candidate voted for in current term (#f if none) + (mutable log) ;; list of log-entry (index 1-based, stored in order) + (mutable commit-index) ;; highest committed log index + (mutable last-applied) ;; highest applied log index + (mutable state) ;; 'follower | 'candidate | 'leader + inbox ;; channel for receiving messages + (mutable peers) ;; list of (id . channel) for other nodes + (mutable votes-received) ;; set of voter IDs in current election + (mutable next-index) ;; per-follower: next log index to send (leader only) + (mutable match-index) ;; per-follower: highest replicated index (leader only) + (mutable running?) ;; is this node running? + mutex ;; protects mutable fields + (mutable election-timer) ;; thread for election timeout + (mutable heartbeat-timer) ;; thread for heartbeat + (mutable committed-log) ;; applied entries (for inspection) + )) + + (define (make-raft-node id) + (%make-raft-node + id + 0 ;; current-term + #f ;; voted-for + '() ;; log + 0 ;; commit-index + 0 ;; last-applied + 'follower ;; state + (make-channel 64) ;; inbox + '() ;; peers + '() ;; votes-received + '() ;; next-index + '() ;; match-index + #f ;; running? + (make-mutex) + #f ;; election-timer thread + #f ;; heartbeat-timer thread + '() ;; committed-log + )) + + ;; Accessors for exported API + (define (raft-term node) (raft-node-current-term node)) + (define (raft-state node) (raft-node-state node)) + (define (raft-log node) (raft-node-log node)) + (define (raft-commit-index node) (raft-node-commit-index node)) + (define (raft-leader? node) (eq? (raft-node-state node) 'leader)) + + ;; ========== Utility Helpers ========== + + (define (log-last-index node) + (let ([log (raft-node-log node)]) + (if (null? log) 0 (log-entry-index (car (last-pair log)))))) + + (define (log-last-term node) + (let ([log (raft-node-log node)]) + (if (null? log) 0 (log-entry-term (car (last-pair log)))))) + + (define (log-entry-at node index) + (let loop ([log (raft-node-log node)]) + (cond + [(null? log) #f] + [(= (log-entry-index (car log)) index) (car log)] + [else (loop (cdr log))]))) + + (define (majority count) + (+ (quotient count 2) 1)) + + ;; Random election timeout: 150–300ms + (define (election-timeout-ms) + (+ 150 (random 150))) + + ;; Heartbeat interval: 50ms + (define heartbeat-interval-ms 50) + + ;; Send to a peer by looking up their channel + (define (send-to-peer node peer-id msg) + (let ([entry (assv peer-id (raft-node-peers node))]) + (when entry + (guard (exn [#t (void)]) + (channel-put (cdr entry) msg))))) + + ;; Broadcast to all peers + (define (broadcast-to-peers node msg) + (for-each + (lambda (entry) (send-to-peer node (car entry) msg)) + (raft-node-peers node))) + + ;; ========== State Transitions ========== + + (define (become-follower! node term) + (raft-node-current-term-set! node term) + (raft-node-voted-for-set! node #f) + (raft-node-state-set! node 'follower) + (raft-node-votes-received-set! node '()) + (stop-heartbeat! node) + (reset-election-timer! node)) + + (define (become-candidate! node) + (let ([new-term (+ (raft-node-current-term node) 1)]) + (raft-node-current-term-set! node new-term) + (raft-node-state-set! node 'candidate) + (raft-node-voted-for-set! node (raft-node-id node)) + (raft-node-votes-received-set! node (list (raft-node-id node))) + ;; Send RequestVote to all peers + (let ([req (make-vote-request + new-term + (raft-node-id node) + (log-last-index node) + (log-last-term node))]) + (broadcast-to-peers node req)) + ;; Check if we already have majority (single node cluster) + (check-election-won! node) + (reset-election-timer! node))) + + (define (become-leader! node) + (raft-node-state-set! node 'leader) + (stop-election-timer! node) + ;; Initialize next-index and match-index for followers + (let ([next-idx (+ (log-last-index node) 1)]) + (raft-node-next-index-set! node + (map (lambda (entry) (cons (car entry) next-idx)) + (raft-node-peers node))) + (raft-node-match-index-set! node + (map (lambda (entry) (cons (car entry) 0)) + (raft-node-peers node)))) + ;; Send immediate heartbeat + (send-heartbeats! node) + (start-heartbeat! node)) + + ;; ========== Election Timeout ========== + + (define (reset-election-timer! node) + (stop-election-timer! node) + (when (raft-node-running? node) + (let ([t (fork-thread + (lambda () + (let ([timeout-ms (election-timeout-ms)]) + (sleep (make-time 'time-duration + (* timeout-ms 1000000) 0)) + (when (and (raft-node-running? node) + (not (eq? (raft-node-state node) 'leader))) + (channel-put (raft-node-inbox node) + (vector 'election-timeout))))))]) + (raft-node-election-timer-set! node t)))) + + (define (stop-election-timer! node)