Migrate 3 more protocol-using .sls files (complex bodies)
ober
2bcb2c0c5649efa90c8d82e28b726687720ee7dc
deleted file mode 100644 --- a/lib/std/actor/engine.sls +++ /dev/null @@ -1,231 +0,0 @@ -#!chezscheme -;;; (std actor engine) — Engine-based preemptive actor scheduling -;;; -;;; Chez Scheme's engine API provides preemptible computations via -;;; "fuel" (instruction quanta). This library builds an actor pool -;;; where each actor is run inside an engine so that long-running -;;; behaviors are automatically time-sliced. -;;; -;;; API: -;;; (make-engine-pool #:workers n #:fuel f) -> engine-pool -;;; (engine-pool? x) -;;; (spawn-engine-actor pool behavior) -> actor-ref -;;; (engine-pool-submit! pool thunk) -;;; (engine-pool-stop! pool) -;;; (engine-pool-worker-count pool) -;;; (default-fuel) -> 10000 -;;; -;;; How it works: -;;; Each worker OS thread runs a tight loop that dequeues thunks and -;;; wraps them in Chez engines. If a thunk's engine runs out of fuel -;;; (the computation is still in progress) the remaining engine is -;;; re-queued so another worker can eventually run it. When the -;;; engine completes the result is discarded (fire-and-forget -;;; semantics, matching the actor model). - -(library (std actor engine) - (export - make-engine-pool - engine-pool? - spawn-engine-actor - engine-pool-submit! - engine-pool-stop! - engine-pool-worker-count - default-fuel) - - (import (chezscheme) (std actor core)) - - ;; -------- Default fuel quanta -------- - - (define (default-fuel) 10000) - - ;; -------- Shared task queue -------- - ;; A simple mutex-protected FIFO of thunks / pending engines. - ;; Each item is either: - ;; (cons 'thunk thunk) — not yet started, wrap in make-engine - ;; (cons 'engine engine-fn) — partially run, resume with more fuel - - (define-record-type eng-task-queue - (fields - (immutable mutex) - (immutable not-empty) ;; condition variable - (mutable head) ;; list: items ready to dequeue - (mutable tail)) ;; list: newly enqueued items (reversed) - (protocol - (lambda (new) - (lambda () - (new (make-mutex) - (make-condition) - '() - '())))) - (sealed #t)) - - (define (tq-enqueue! tq item) - (with-mutex (eng-task-queue-mutex tq) - (eng-task-queue-tail-set! tq (cons item (eng-task-queue-tail tq))) - (condition-signal (eng-task-queue-not-empty tq)))) - - ;; Blocking dequeue. Returns an item, or #f when the pool is stopping. - (define (tq-dequeue! tq running-thunk) - (mutex-acquire (eng-task-queue-mutex tq)) - (let loop () - (cond - ;; Head has items — take from front - [(pair? (eng-task-queue-head tq)) - (let ([item (car (eng-task-queue-head tq))]) - (eng-task-queue-head-set! tq (cdr (eng-task-queue-head tq))) - (mutex-release (eng-task-queue-mutex tq)) - item)] - ;; Promote tail into head - [(pair? (eng-task-queue-tail tq)) - (eng-task-queue-head-set! tq (reverse (eng-task-queue-tail tq))) - (eng-task-queue-tail-set! tq '()) - (loop)] - ;; Empty — wait if still running - [(running-thunk) - (condition-wait (eng-task-queue-not-empty tq) - (eng-task-queue-mutex tq)) - (loop)] - ;; Stopping — release and signal shutdown - [else - (mutex-release (eng-task-queue-mutex tq)) - #f]))) - - ;; -------- Engine pool record -------- - ;; Use a distinct record name (eng-pool-rec) so we can provide a - ;; user-facing make-engine-pool procedure with keyword parsing. - - (define-record-type eng-pool-rec - (fields - (immutable queue) ;; eng-task-queue - (immutable fuel) ;; integer: ticks per engine slice - (immutable nworkers) ;; integer: number of OS threads - (mutable running?)) ;; boolean - (protocol - (lambda (new) - (lambda (nworkers fuel) - (new (make-eng-task-queue) fuel nworkers #f)))) - (sealed #t)) - - (define engine-pool? eng-pool-rec?) - - ;; -------- Worker loop -------- - ;; - ;; NOTE: In Chez Scheme 10.x the engine API uses INVERTED semantics - ;; compared to the traditional (Dybvig) documentation: - ;; - ;; expire-proc — called when the computation FINISHES within the fuel - ;; budget: (expire-proc remaining-fuel result) - ;; complete-proc — called when the computation is PREEMPTED (fuel - ;; exhausted): (complete-proc new-engine) - ;; - ;; We rename the parameters accordingly: done-proc / preempt-proc. - - (define (worker-loop pool) - (let ([q (eng-pool-rec-queue pool)] - [fuel (eng-pool-rec-fuel pool)]) - (let loop () - (let ([item (tq-dequeue! q (lambda () (eng-pool-rec-running? pool)))]) - (when item - ;; Build or retrieve the engine - (let ([eng (case (car item) - [(thunk) (make-engine (cdr item))] - [(engine) (cdr item)] - [else - (error 'engine-pool-worker - "unknown task type" (car item))])]) - ;; Run the engine for one fuel slice. - (eng fuel - ;; done-proc (called "expire" in Chez): computation finished - ;; (remaining-fuel result) — result is discarded (fire-and-forget) - (lambda (remaining result) (void)) - ;; preempt-proc (called "complete" in Chez): fuel exhausted - ;; (new-engine) — re-enqueue the continuation - (lambda (new-engine) - (tq-enqueue! q (cons 'engine new-engine))))) - (loop)))))) - - ;; -------- Public API -------- - - ;; Keyword predicate helpers. - ;; In Chez Scheme, the #:foo syntax at a call site evaluates the symbol - ;; as a variable, so callers must quote keyword symbols: '#:workers. - ;; We compare by symbol name string (stripping a leading "#:" if present). - (define (kw=? sym name) - (and (symbol? sym) - (let ([s (symbol->string sym)]) - (or (string=? s name) - ;; Accept symbol with literal "#:" prefix in case the reader - ;; is configured to preserve it (some Chez versions / modes). - (and (fx>= (string-length s) 2) - (char=? (string-ref s 0) #\#) - (char=? (string-ref s 1) #\:) - (string=? (substring s 2 (string-length s)) name)))))) - - ;; (make-engine-pool '#:workers n '#:fuel f) - ;; OR (make-engine-pool) for defaults (4 workers, default-fuel ticks). - ;; OR (make-engine-pool n) for n workers with default fuel. - ;; OR (make-engine-pool n f) for n workers with f fuel. - (define (make-engine-pool . args) - (define (start! workers fuel) - (let ([pool (make-eng-pool-rec workers fuel)]) - (eng-pool-rec-running?-set! pool #t) - (do ([i 0 (fx+ i 1)]) - ((fx= i workers)) - (fork-thread (lambda () (worker-loop pool)))) - pool)) - (cond - ;; No args — defaults - [(null? args) - (start! 4 (default-fuel))] - ;; First arg is a number — positional: (workers) or (workers fuel) - [(and (number? (car args)) (null? (cdr args))) - (start! (car args) (default-fuel))] - [(and (number? (car args)) (pair? (cdr args)) (number? (cadr args)) - (null? (cddr args))) - (start! (car args) (cadr args))] - ;; Keyword-style: '#:workers n '#:fuel f (args are quoted symbols) - [else - (let parse ([rest args] [workers 4] [fuel (default-fuel)]) - (cond - [(null? rest) (start! workers fuel)] - [(and (kw=? (car rest) "workers") (pair? (cdr rest))) - (parse (cddr rest) (cadr rest) fuel)] - [(and (kw=? (car rest) "fuel") (pair? (cdr rest))) - (parse (cddr rest) workers (cadr rest))] - [else - (error 'make-engine-pool - "unexpected argument (use '#:workers n or '#:fuel f)" - (car rest))]))])) - - (define (engine-pool-submit! pool thunk) - (unless (eng-pool-rec-running? pool) - (error 'engine-pool-submit! "pool has been stopped" pool)) - (tq-enqueue! (eng-pool-rec-queue pool) (cons 'thunk thunk))) - - (define (engine-pool-stop! pool) - (eng-pool-rec-running?-set! pool #f) - ;; Broadcast to wake all sleeping workers so they exit their loops - (with-mutex (eng-task-queue-mutex (eng-pool-rec-queue pool)) - (condition-broadcast - (eng-task-queue-not-empty (eng-pool-rec-queue pool))))) - - (define (engine-pool-worker-count pool) - (eng-pool-rec-nworkers pool)) - - ;; Spawn an actor that runs preemptively inside the engine pool. - ;; - ;; The pool is installed as the global actor scheduler so that all - ;; subsequent scheduling decisions for this actor land in the pool's - ;; engine queue, giving preemptive time-slicing. - ;; - ;; Callers that want multiple pools should set the scheduler themselves - ;; before spawning; this convenience wrapper sets it once and leaves it. - (define (spawn-engine-actor pool behavior) - ;; Build a submit procedure matching set-actor-scheduler!'s contract: - ;; it receives a zero-argument thunk and submits it to the pool. - (set-actor-scheduler! - (lambda (thunk) (engine-pool-submit! pool thunk))) - (spawn-actor behavior)) - - ) ;; end library new file mode 100644 --- /dev/null +++ b/lib/std/actor/engine.ss @@ -0,0 +1,227 @@ +#!chezscheme +;;; (std actor engine) — Engine-based preemptive actor scheduling +;;; +;;; Chez Scheme's engine API provides preemptible computations via +;;; "fuel" (instruction quanta). This library builds an actor pool +;;; where each actor is run inside an engine so that long-running +;;; behaviors are automatically time-sliced. +;;; +;;; API: +;;; (make-engine-pool #:workers n #:fuel f) -> engine-pool +;;; (engine-pool? x) +;;; (spawn-engine-actor pool behavior) -> actor-ref +;;; (engine-pool-submit! pool thunk) +;;; (engine-pool-stop! pool) +;;; (engine-pool-worker-count pool) +;;; (default-fuel) -> 10000 +;;; +;;; How it works: +;;; Each worker OS thread runs a tight loop that dequeues thunks and +;;; wraps them in Chez engines. If a thunk's engine runs out of fuel +;;; (the computation is still in progress) the remaining engine is +;;; re-queued so another worker can eventually run it. When the +;;; engine completes the result is discarded (fire-and-forget +;;; semantics, matching the actor model). + +(library (std actor engine) + (export + make-engine-pool + engine-pool? + spawn-engine-actor + engine-pool-submit! + engine-pool-stop! + engine-pool-worker-count + default-fuel) + + (import (chezscheme) (std actor core) + (only (jerboa core) def defstruct)) + + ;; -------- Default fuel quanta -------- + + (def (default-fuel) 10000) + + ;; -------- Shared task queue -------- + ;; A simple mutex-protected FIFO of thunks / pending engines. + ;; Each item is either: + ;; (cons 'thunk thunk) — not yet started, wrap in make-engine + ;; (cons 'engine engine-fn) — partially run, resume with more fuel + + (defstruct eng-task-queue-raw (mutex not-empty head tail)) + (def (make-eng-task-queue) (make-eng-task-queue-raw (make-mutex) + (make-condition) + '() + '())) + (def eng-task-queue? eng-task-queue-raw?) + (def eng-task-queue-mutex eng-task-queue-raw-mutex) + (def eng-task-queue-not-empty eng-task-queue-raw-not-empty) + (def eng-task-queue-head eng-task-queue-raw-head) + (def eng-task-queue-head-set! eng-task-queue-raw-head-set!) + (def eng-task-queue-tail eng-task-queue-raw-tail) + (def eng-task-queue-tail-set! eng-task-queue-raw-tail-set!) + + (def (tq-enqueue! tq item) + (with-mutex (eng-task-queue-mutex tq) + (eng-task-queue-tail-set! tq (cons item (eng-task-queue-tail tq))) + (condition-signal (eng-task-queue-not-empty tq)))) + + ;; Blocking dequeue. Returns an item, or #f when the pool is stopping. + (def (tq-dequeue! tq running-thunk) + (mutex-acquire (eng-task-queue-mutex tq)) + (let loop () + (cond + ;; Head has items — take from front + [(pair? (eng-task-queue-head tq)) + (let ([item (car (eng-task-queue-head tq))]) + (eng-task-queue-head-set! tq (cdr (eng-task-queue-head tq))) + (mutex-release (eng-task-queue-mutex tq)) + item)] + ;; Promote tail into head + [(pair? (eng-task-queue-tail tq)) + (eng-task-queue-head-set! tq (reverse (eng-task-queue-tail tq))) + (eng-task-queue-tail-set! tq '()) + (loop)] + ;; Empty — wait if still running + [(running-thunk) + (condition-wait (eng-task-queue-not-empty tq) + (eng-task-queue-mutex tq)) + (loop)] + ;; Stopping — release and signal shutdown + [else + (mutex-release (eng-task-queue-mutex tq)) + #f]))) + + ;; -------- Engine pool record -------- + ;; Use a distinct record name (eng-pool-rec) so we can provide a + ;; user-facing make-engine-pool procedure with keyword parsing. + + (defstruct eng-pool-rec-raw (queue fuel nworkers running?)) + (def (make-eng-pool-rec nworkers fuel) (make-eng-pool-rec-raw (make-eng-task-queue) fuel nworkers #f)) + (def eng-pool-rec? eng-pool-rec-raw?) + (def eng-pool-rec-queue eng-pool-rec-raw-queue) + (def eng-pool-rec-fuel eng-pool-rec-raw-fuel) + (def eng-pool-rec-nworkers eng-pool-rec-raw-nworkers) + (def eng-pool-rec-running? eng-pool-rec-raw-running?) + (def eng-pool-rec-running?-set! eng-pool-rec-raw-running?-set!) + + (def engine-pool? eng-pool-rec?) + + ;; -------- Worker loop -------- + ;; + ;; NOTE: In Chez Scheme 10.x the engine API uses INVERTED semantics + ;; compared to the traditional (Dybvig) documentation: + ;; + ;; expire-proc — called when the computation FINISHES within the fuel + ;; budget: (expire-proc remaining-fuel result) + ;; complete-proc — called when the computation is PREEMPTED (fuel + ;; exhausted): (complete-proc new-engine) + ;; + ;; We rename the parameters accordingly: done-proc / preempt-proc. + + (def (worker-loop pool) + (let ([q (eng-pool-rec-queue pool)] + [fuel (eng-pool-rec-fuel pool)]) + (let loop () + (let ([item (tq-dequeue! q (lambda () (eng-pool-rec-running? pool)))]) + (when item + ;; Build or retrieve the engine + (let ([eng (case (car item) + [(thunk) (make-engine (cdr item))] + [(engine) (cdr item)] + [else + (error 'engine-pool-worker + "unknown task type" (car item))])]) + ;; Run the engine for one fuel slice. + (eng fuel + ;; done-proc (called "expire" in Chez): computation finished + ;; (remaining-fuel result) — result is discarded (fire-and-forget) + (lambda (remaining result) (void)) + ;; preempt-proc (called "complete" in Chez): fuel exhausted + ;; (new-engine) — re-enqueue the continuation + (lambda (new-engine) + (tq-enqueue! q (cons 'engine new-engine))))) + (loop)))))) + + ;; -------- Public API -------- + + ;; Keyword predicate helpers. + ;; In Chez Scheme, the #:foo syntax at a call site evaluates the symbol + ;; as a variable, so callers must quote keyword symbols: '#:workers. + ;; We compare by symbol name string (stripping a leading "#:" if present). + (def (kw=? sym name) + (and (symbol? sym) + (let ([s (symbol->string sym)]) + (or (string=? s name) + ;; Accept symbol with literal "#:" prefix in case the reader + ;; is configured to preserve it (some Chez versions / modes). + (and (fx>= (string-length s) 2) + (char=? (string-ref s 0) #\#) + (char=? (string-ref s 1) #\:) + (string=? (substring s 2 (string-length s)) name)))))) + + ;; (make-engine-pool '#:workers n '#:fuel f) + ;; OR (make-engine-pool) for defaults (4 workers, default-fuel ticks). + ;; OR (make-engine-pool n) for n workers with default fuel. + ;; OR (make-engine-pool n f) for n workers with f fuel. + (def (make-engine-pool . args) + (def (start! workers fuel) + (let ([pool (make-eng-pool-rec workers fuel)]) + (eng-pool-rec-running?-set! pool #t) + (do ([i 0 (fx+ i 1)]) + ((fx= i workers)) + (fork-thread (lambda () (worker-loop pool)))) + pool)) + (cond + ;; No args — defaults + [(null? args) + (start! 4 (default-fuel))] + ;; First arg is a number — positional: (workers) or (workers fuel) + [(and (number? (car args)) (null? (cdr args))) + (start! (car args) (default-fuel))] + [(and (number? (car args)) (pair? (cdr args)) (number? (cadr args)) + (null? (cddr args))) + (start! (car args) (cadr args))] + ;; Keyword-style: '#:workers n '#:fuel f (args are quoted symbols) + [else + (let parse ([rest args] [workers 4] [fuel (default-fuel)]) + (cond + [(null? rest) (start! workers fuel)] + [(and (kw=? (car rest) "workers") (pair? (cdr rest))) + (parse (cddr rest) (cadr rest) fuel)] + [(and (kw=? (car rest) "fuel") (pair? (cdr rest))) + (parse (cddr rest) workers (cadr rest))] + [else + (error 'make-engine-pool + "unexpected argument (use '#:workers n or '#:fuel f)" + (car rest))]))])) + + (def (engine-pool-submit! pool thunk) + (unless (eng-pool-rec-running? pool) + (error 'engine-pool-submit! "pool has been stopped" pool)) + (tq-enqueue! (eng-pool-rec-queue pool) (cons 'thunk thunk))) + + (def (engine-pool-stop! pool) + (eng-pool-rec-running?-set! pool #f) + ;; Broadcast to wake all sleeping workers so they exit their loops + (with-mutex (eng-task-queue-mutex (eng-pool-rec-queue pool)) + (condition-broadcast + (eng-task-queue-not-empty (eng-pool-rec-queue pool))))) + + (def (engine-pool-worker-count pool) + (eng-pool-rec-nworkers pool)) + + ;; Spawn an actor that runs preemptively inside the engine pool. + ;; + ;; The pool is installed as the global actor scheduler so that all + ;; subsequent scheduling decisions for this actor land in the pool's + ;; engine queue, giving preemptive time-slicing. + ;; + ;; Callers that want multiple pools should set the scheduler themselves + ;; before spawning; this convenience wrapper sets it once and leaves it. + (def (spawn-engine-actor pool behavior) + ;; Build a submit procedure matching set-actor-scheduler!'s contract: + ;; it receives a zero-argument thunk and submits it to the pool. + (set-actor-scheduler! + (lambda (thunk) (engine-pool-submit! pool thunk))) + (spawn-actor behavior)) + + ) ;; end library deleted file mode 100644 --- a/lib/std/concur/stm.sls +++ /dev/null @@ -1,198 +0,0 @@ -;;; Software Transactional Memory — Phase 5d (Track 17.1) -;;; -;;; Optimistic concurrency with transactional variables (TVars). -;;; Supports nested transactions, retry, and or-else. - -(library (std concur stm) - (export - make-tvar tvar? tvar-get tvar-set! - atomically retry or-else) - (import (chezscheme)) - - ;; ----------------------------------------------------------------------- - ;; TVar — a versioned mutable cell - ;; ----------------------------------------------------------------------- - - (define-record-type tvar - (fields (mutable val tvar-val set-tvar-val!) - (mutable ver tvar-ver set-tvar-ver!)) - (protocol (lambda (new) (lambda (init) (new init 0))))) - - ;; Global version clock - (define *global-version* 0) - - ;; ----------------------------------------------------------------------- - ;; Transaction context (thread-local) - ;; ----------------------------------------------------------------------- - ;; Each transaction has: - ;; read-set: eq-hashtable tvar → observed-version - ;; write-set: eq-hashtable tvar → new-value - ;; parent: outer transaction or #f - - (define-record-type txn - (fields (immutable read-set txn-read-set) - (immutable write-set txn-write-set) - (mutable parent txn-parent set-txn-parent!)) - (protocol - (lambda (new) - (lambda (parent) - (new (make-eq-hashtable) (make-eq-hashtable) parent))))) - - (define *current-txn* (make-thread-parameter #f)) - - ;; ----------------------------------------------------------------------- - ;; Global commit mutex - ;; ----------------------------------------------------------------------- - - (define *stm-mutex* (make-mutex)) - (define *stm-cond* (make-condition)) - - ;; ----------------------------------------------------------------------- - ;; tvar-get — read TVar in transaction context - ;; ----------------------------------------------------------------------- - - (define (tvar-get tv) - (let ([txn (*current-txn*)]) - (if txn - ;; Inside transaction - (let ([ws (txn-write-set txn)] - [rs (txn-read-set txn)]) - (cond - ;; Written in this txn → return pending value - [(hashtable-contains? ws tv) - (hashtable-ref ws tv #f)] - ;; First read → record version + return current value - [else - (let ([v (tvar-val tv)] - [ver (tvar-ver tv)]) - (hashtable-set! rs tv ver) - v)])) - ;; Outside transaction → direct read - (tvar-val tv)))) - - ;; ----------------------------------------------------------------------- - ;; tvar-set! — write TVar in transaction context - ;; ----------------------------------------------------------------------- - - (define (tvar-set! tv val) - (let ([txn (*current-txn*)]) - (if txn - (hashtable-set! (txn-write-set txn) tv val) - ;; Direct write outside transaction - (begin - (mutex-acquire *stm-mutex*) - (set-tvar-val! tv val) - (set-tvar-ver! tv (+ *global-version* 1)) - (set! *global-version* (+ *global-version* 1)) - (condition-broadcast *stm-cond*) - (mutex-release *stm-mutex*))))) - - ;; ----------------------------------------------------------------------- - ;; Validation — check all read TVars are still current - ;; ----------------------------------------------------------------------- - - (define (validate-read-set! txn) - "Return #t if all read TVars still have observed versions" - (let-values ([(tvs vers) (hashtable-entries (txn-read-set txn))]) - (let loop ([i 0]) - (or (= i (vector-length tvs)) - (and (= (tvar-ver (vector-ref tvs i)) (vector-ref vers i)) - (loop (+ i 1))))))) - - ;; ----------------------------------------------------------------------- - ;; Commit — write all write-set values atomically - ;; ----------------------------------------------------------------------- - - (define (commit-txn! txn) - "Attempt to commit TXN; return #t on success, #f on conflict" - (mutex-acquire *stm-mutex*) - (let ([ok (validate-read-set! txn)]) - (when ok - (let-values ([(tvs vals) (hashtable-entries (txn-write-set txn))]) - (let ([new-ver (+ *global-version* 1)]) - (set! *global-version* new-ver) - (vector-for-each - (lambda (tv val) - (set-tvar-val! tv val) - (set-tvar-ver! tv new-ver)) - tvs vals))) - (condition-broadcast *stm-cond*)) - (mutex-release *stm-mutex*) - ok)) - - ;; ----------------------------------------------------------------------- - ;; atomically — run a transaction - ;; ----------------------------------------------------------------------- - - ;; Retry sentinel - (define *retry-tag* (list 'retry)) - - (define (atomically thunk) - (let ([outer (*current-txn*)]) - (if outer - ;; Nested: create child txn, merge into outer on success - (let ([child (make-txn outer)]) - (parameterize ([*current-txn* child]) - (let ([result (thunk)]) - ;; Merge child write-set into outer - (let-values ([(tvs vals) (hashtable-entries (txn-write-set child))]) - (vector-for-each - (lambda (tv v) (hashtable-set! (txn-write-set outer) tv v)) - tvs vals)) - result))) - ;; Top-level: run with retry loop - (call-with-current-continuation - (lambda (k-done) - (let loop () - (let ([txn (make-txn #f)]) - (parameterize ([*current-txn* txn]) - (call-with-current-continuation - (lambda (k-escape) - (with-exception-handler - (lambda (e) - (if (eq? e *retry-tag*) - ;; retry: block until some TVar changes - (begin - (mutex-acquire *stm-mutex*) - (condition-wait *stm-cond* *stm-mutex*) - (mutex-release *stm-mutex*) - (k-escape 'retry)) - (raise e))) - (lambda () - (let ([result (thunk)]) - (when (commit-txn! txn) - (k-done result))))))))) - ;; retry or conflict: run again - (loop))))))) - - ;; ----------------------------------------------------------------------- - ;; retry — abort and wait for change - ;; ----------------------------------------------------------------------- - - (define (retry) - (raise *retry-tag*)) - - ;; ----------------------------------------------------------------------- - ;; or-else — try first transaction; if it retries, try second - ;; ----------------------------------------------------------------------- - - (define-syntax or-else - (syntax-rules () - [(_ thunk1 thunk2) - (let ([succeeded #f] - [result #f]) - (call-with-current-continuation - (lambda (k) - (with-exception-handler - (lambda (e) - (if (eq? e *retry-tag*) - (k 'retry-first) - (raise e))) - (lambda () - (set! result (thunk1)) - (set! succeeded #t))))) - (if succeeded - result - (thunk2)))])) - -) new file mode 100644 --- /dev/null +++ b/lib/std/concur/stm.ss @@ -0,0 +1,201 @@ +;;; Software Transactional Memory — Phase 5d (Track 17.1) +;;; +;;; Optimistic concurrency with transactional variables (TVars). +;;; Supports nested transactions, retry, and or-else. + +(library (std concur stm) + (export + make-tvar tvar? tvar-get tvar-set! + atomically retry or-else) + (import (chezscheme) + (only (jerboa core) def defstruct)) + + ;; ----------------------------------------------------------------------- + ;; TVar — a versioned mutable cell + ;; ----------------------------------------------------------------------- + + (defstruct tvar-raw (val ver)) + (def (make-tvar init) (make-tvar-raw init 0)) + (def tvar? tvar-raw?) + (def tvar-val tvar-raw-val) + (def set-tvar-val! tvar-raw-val-set!) + (def tvar-ver tvar-raw-ver) + (def set-tvar-ver! tvar-raw-ver-set!) + + ;; Global version clock + (def *global-version* 0) + + ;; ----------------------------------------------------------------------- + ;; Transaction context (thread-local) + ;; ----------------------------------------------------------------------- + ;; Each transaction has: + ;; read-set: eq-hashtable tvar → observed-version + ;; write-set: eq-hashtable tvar → new-value + ;; parent: outer transaction or #f + + (defstruct txn-raw (read-set write-set parent)) + (def (make-txn parent) (make-txn-raw (make-eq-hashtable) (make-eq-hashtable) parent)) + (def txn? txn-raw?) + (def txn-read-set txn-raw-read-set) + (def txn-write-set txn-raw-write-set) + (def txn-parent txn-raw-parent) + (def set-txn-parent! txn-raw-parent-set!) + + (def *current-txn* (make-thread-parameter #f)) + + ;; ----------------------------------------------------------------------- + ;; Global commit mutex + ;; ----------------------------------------------------------------------- + + (def *stm-mutex* (make-mutex)) + (def *stm-cond* (make-condition)) + + ;; ----------------------------------------------------------------------- + ;; tvar-get — read TVar in transaction context + ;; ----------------------------------------------------------------------- + + (def (tvar-get tv) + (let ([txn (*current-txn*)]) + (if txn + ;; Inside transaction + (let ([ws (txn-write-set txn)] + [rs (txn-read-set txn)]) + (cond + ;; Written in this txn → return pending value + [(hashtable-contains? ws tv) + (hashtable-ref ws tv #f)] + ;; First read → record version + return current value + [else + (let ([v (tvar-val tv)] + [ver (tvar-ver tv)]) + (hashtable-set! rs tv ver) + v)])) + ;; Outside transaction → direct read + (tvar-val tv)))) + + ;; ----------------------------------------------------------------------- + ;; tvar-set! — write TVar in transaction context + ;; ----------------------------------------------------------------------- + + (def (tvar-set! tv val) + (let ([txn (*current-txn*)]) + (if txn + (hashtable-set! (txn-write-set txn) tv val) + ;; Direct write outside transaction + (begin + (mutex-acquire *stm-mutex*) + (set-tvar-val! tv val) + (set-tvar-ver! tv (+ *global-version* 1)) + (set! *global-version* (+ *global-version* 1)) + (condition-broadcast *stm-cond*) + (mutex-release *stm-mutex*))))) + + ;; ----------------------------------------------------------------------- + ;; Validation — check all read TVars are still current + ;; ----------------------------------------------------------------------- + + (def (validate-read-set! txn) + "Return #t if all read TVars still have observed versions" + (let-values ([(tvs vers) (hashtable-entries (txn-read-set txn))]) + (let loop ([i 0]) + (or (= i (vector-length tvs)) + (and (= (tvar-ver (vector-ref tvs i)) (vector-ref vers i)) + (loop (+ i 1))))))) + + ;; ----------------------------------------------------------------------- + ;; Commit — write all write-set values atomically + ;; ----------------------------------------------------------------------- + + (def (commit-txn! txn) + "Attempt to commit TXN; return #t on success, #f on conflict" + (mutex-acquire *stm-mutex*) + (let ([ok (validate-read-set! txn)]) + (when ok + (let-values ([(tvs vals) (hashtable-entries (txn-write-set txn))]) + (let ([new-ver (+ *global-version* 1)]) + (set! *global-version* new-ver) + (vector-for-each + (lambda (tv val) + (set-tvar-val! tv val) + (set-tvar-ver! tv new-ver)) + tvs vals))) + (condition-broadcast *stm-cond*)) + (mutex-release *stm-mutex*) + ok)) + + ;; ----------------------------------------------------------------------- + ;; atomically — run a transaction + ;; ----------------------------------------------------------------------- + + ;; Retry sentinel + (def *retry-tag* (list 'retry)) + + (def (atomically thunk) + (let ([outer (*current-txn*)]) + (if outer + ;; Nested: create child txn, merge into outer on success + (let ([child (make-txn outer)]) + (parameterize ([*current-txn* child]) + (let ([result (thunk)]) + ;; Merge child write-set into outer + (let-values ([(tvs vals) (hashtable-entries (txn-write-set child))]) + (vector-for-each + (lambda (tv v) (hashtable-set! (txn-write-set outer) tv v)) + tvs vals)) + result))) + ;; Top-level: run with retry loop + (call-with-current-continuation + (lambda (k-done) + (let loop () + (let ([txn (make-txn #f)]) + (parameterize ([*current-txn* txn]) + (call-with-current-continuation + (lambda (k-escape) + (with-exception-handler + (lambda (e) + (if (eq? e *retry-tag*) + ;; retry: block until some TVar changes + (begin + (mutex-acquire *stm-mutex*) + (condition-wait *stm-cond* *stm-mutex*) + (mutex-release *stm-mutex*) + (k-escape 'retry)) + (raise e))) + (lambda () + (let ([result (thunk)]) + (when (commit-txn! txn) + (k-done result))))))))) + ;; retry or conflict: run again + (loop))))))) + + ;; ----------------------------------------------------------------------- + ;; retry — abort and wait for change + ;; ----------------------------------------------------------------------- + + (def (retry) + (raise *retry-tag*)) + + ;; ----------------------------------------------------------------------- + ;; or-else — try first transaction; if it retries, try second + ;; ----------------------------------------------------------------------- + + (define-syntax or-else + (syntax-rules () + [(_ thunk1 thunk2) + (let ([succeeded #f] + [result #f]) + (call-with-current-continuation + (lambda (k) + (with-exception-handler + (lambda (e) + (if (eq? e *retry-tag*) + (k 'retry-first) + (raise e))) + (lambda () + (set! result (thunk1)) + (set! succeeded #t))))) + (if succeeded + result + (thunk2)))])) + +) deleted file mode 100644 --- a/lib/std/misc/custodian.sls +++ /dev/null @@ -1,151 +0,0 @@ -#!chezscheme -;;; (std misc custodian) --- Hierarchical custodians for resource management -;;; -;;; Custodians are hierarchical resource groups that can be shut down atomically. -;;; Every managed resource (ports, custom handles) belongs to a custodian. -;;; Shutting down a parent recursively shuts down all children and their resources. -;;; -;;; Usage: -;;; (with-custodian -;;; (let ([p (custodian-open-input-file "data.txt")]) -;;; (read p))) -;;; ;; port is automatically closed when with-custodian exits - -(library (std misc custodian) - (export make-custodian - current-custodian - custodian? - custodian-shutdown-all - custodian-managed-list - custodian-register! - custodian-open-input-file - custodian-open-output-file - with-custodian) - (import (chezscheme)) - - ;; A custodian holds: - ;; parent - parent custodian or #f for root - ;; children - list of child custodians - ;; resources - list of (resource . shutdown-proc) pairs - ;; alive? - #t until shutdown - (define-record-type cust - (fields - (immutable parent) - (mutable children) - (mutable resources) - (mutable alive?)) - (protocol - (lambda (new) - (lambda (parent) - (new parent '() '() #t))))) - - (define (custodian? x) - (cust? x)) - - ;; Root custodian has no parent - (define root-custodian (make-cust #f)) - - ;; Parameter for the current custodian - (define current-custodian (make-parameter root-custodian)) - - ;; Create a new custodian. If parent is not given, uses current-custodian. - (define make-custodian - (case-lambda - [() - (make-custodian (current-custodian))] - [(parent) - (unless (cust? parent) - (error 'make-custodian "expected a custodian" parent)) - (unless (cust-alive? parent) - (error 'make-custodian "parent custodian is shut down" parent)) - (let ([c (make-cust parent)]) - (cust-children-set! parent - (cons c (cust-children parent))) - c)])) - - ;; Register a resource with a custodian. - ;; shutdown-proc is a thunk called to release the resource. - ;; Returns the resource for convenience. - (define custodian-register! - (case-lambda - [(resource shutdown-proc) - (custodian-register! (current-custodian) resource shutdown-proc)] - [(custodian resource shutdown-proc) - (unless (cust? custodian) - (error 'custodian-register! "expected a custodian" custodian)) - (unless (cust-alive? custodian) - (error 'custodian-register! "custodian is shut down" custodian)) - (unless (procedure? shutdown-proc) - (error 'custodian-register! "expected a procedure for shutdown" shutdown-proc)) - (cust-resources-set! custodian - (cons (cons resource shutdown-proc) (cust-resources custodian))) - resource])) - - ;; Shut down a custodian: close all resources, recursively shut down children, - ;; and remove self from parent's child list. - (define (custodian-shutdown-all c) - (unless (cust? c) - (error 'custodian-shutdown-all "expected a custodian" c)) - (when (cust-alive? c) - ;; First, recursively shut down children (copy the list since shutdown mutates it) - (for-each custodian-shutdown-all (list-copy (cust-children c))) - ;; Then close all resources, catching errors so one bad resource - ;; doesn't prevent others from being cleaned up - (for-each - (lambda (pair) - (guard (e [#t (void)]) ;; swallow errors during shutdown - ((cdr pair)))) - (cust-resources c)) - ;; Mark as dead and clear - (cust-alive?-set! c #f) - (cust-resources-set! c '()) - (cust-children-set! c '()) - ;; Remove self from parent's children list - (let ([parent (cust-parent c)]) - (when parent - (cust-children-set! parent - (remq c (cust-children parent))))))) - - ;; List managed resources (not shutdown procs) for a custodian - (define (custodian-managed-list c) - (unless (cust? c) - (error 'custodian-managed-list "expected a custodian" c)) - (append - (map car (cust-resources c)) - (list-copy (cust-children c)))) - - ;; Open an input file port registered with the current custodian - (define custodian-open-input-file - (case-lambda - [(path) - (custodian-open-input-file path (current-custodian))] - [(path custodian) - (let ([p (open-input-file path)]) - (custodian-register! custodian p (lambda () (close-input-port p))) - p)])) - - ;; Open an output file port registered with the current custodian - (define custodian-open-output-file - (case-lambda - [(path) - (custodian-open-output-file path (current-custodian))] - [(path custodian)