Add core.async-style CSP layer: sliding/dropping buffers, alts!, ops, clj surface
ober
60dd3e490f1c728994b743b01b94a938975754b2
--- a/lib/std/csp.sls +++ b/lib/std/csp.sls @@ -1,74 +1,196 @@ #!chezscheme -;;; (std csp) — Communicating Sequential Processes with green threads +;;; (std csp) — Communicating Sequential Processes over OS threads ;;; -;;; True CSP with typed channels, select with timeout, and backpressure. -;;; Green threads scheduled via Chez's engine system. +;;; Channels with blocking put/get and non-blocking try variants, +;;; buffered or unbuffered, safe for multi-producer/multi-consumer +;;; use. Built on Chez Scheme's mutex + condition variables. +;;; +;;; NOTE ON THREADS: `go` and `go-named` are thin wrappers around +;;; `fork-thread` — each `go` spawns a real OS thread. There is no +;;; CPS transform and no green-thread scheduler. For lightweight +;;; fan-out up to a few thousand threads this is fine; don't expect +;;; to scale to millions of concurrent go-blocks the way Clojure's +;;; core.async does. +;;; +;;; NOTE ON SELECT: multi-channel select / alts! is NOT provided by +;;; this module. Build it on top of (std event), or use explicit +;;; polling with `chan-try-get`. +;;; +;;; NOTE ON BUFFERS: three policies are supported: +;;; - fixed : `make-channel/buf` — blocks the writer when full +;;; - sliding : `make-channel/sliding` — drops the oldest queued value +;;; - dropping: `make-channel/dropping` — drops the incoming value ;;; ;;; API: -;;; (make-channel) — unbuffered channel -;;; (make-channel/buf n) — buffered channel with capacity n -;;; (chan-put! ch val) — send value (blocks if full) -;;; (chan-get! ch) — receive value (blocks if empty) -;;; (chan-try-get ch) — non-blocking receive (#f if empty) -;;; (chan-close! ch) — close channel -;;; (chan-closed? ch) — test if closed -;;; (go thunk) — spawn green thread -;;; (go-named name thunk) — spawn named green thread -;;; (yield) — voluntarily yield -;;; (select clause ...) — multi-channel select -;;; (chan->list ch) — drain channel to list +;;; (make-channel) — unbuffered channel +;;; (make-channel n) — buffered channel with capacity n +;;; (make-channel/buf n) — alias for (make-channel n) +;;; (make-channel/sliding n) — capacity n, drop-oldest policy +;;; (make-channel/dropping n) — capacity n, drop-incoming policy +;;; (chan-put! ch val) — blocking send (errors on closed) +;;; (chan-get! ch) — blocking receive (eof on closed+drained) +;;; (chan-try-put! ch val) — non-blocking send; #t success, #f full/closed +;;; (chan-try-get ch) — non-blocking receive; val or #f if empty +;;; (chan-close! ch) — mark channel closed +;;; (chan-closed? ch) — test if closed +;;; (chan-empty? ch) — queue is empty (not the same as closed) +;;; (chan-kind ch) — buffer kind: 'fixed / 'sliding / 'dropping +;;; (go thunk) — spawn an OS thread (= fork-thread) +;;; (go-named name thunk) — spawn an OS thread; name is documentation only +;;; (yield) — 1ms sleep; no scheduler to yield to +;;; (csp-run thunk) — run thunk (no-op wrapper) +;;; (chan->list ch) — drain a closed channel to a list +;;; (chan-pipe from to proc) — spawn a pipeline stage, closes `to` on EOF +;;; (chan-map ch proc) — lazy map onto a new channel +;;; (chan-filter ch pred) — lazy filter onto a new channel +;;; +;;; For the Clojure-style `core.async` surface (chan, >!!, <!!, alts!!, +;;; go macro with body forms, timeout, merge, mult, tap, pipeline, ...) +;;; see `(std csp clj)`. (library (std csp) (export make-channel make-channel/buf - chan-put! chan-get! chan-try-get chan-close! chan-closed? + make-channel/sliding make-channel/dropping + channel? chan-empty? chan-kind + chan-put! chan-get! chan-try-put! chan-try-get + chan-close! chan-closed? go go-named yield csp-run chan->list chan-pipe chan-map chan-filter) (import (chezscheme)) ;; ========== Channel ========== + ;; + ;; FIFO queue is represented in-place with head/tail pointers and an + ;; explicit count. Empty: head = '(), tail = '(), count = 0. Enqueue + ;; and dequeue are O(1). The previous implementation used `append` on + ;; every put — O(n) per put and O(n^2) over n puts. (define-record-type channel (fields - (mutable buffer) ;; list (queue) - (immutable capacity) ;; max items (0 = unbuffered) + (mutable head) ;; first pair of the queue, or '() + (mutable tail) ;; last pair of the queue, or '() + (mutable count) ;; number of items currently buffered + (immutable capacity) ;; max items (0 = unbuffered) + (immutable kind) ;; 'fixed / 'sliding / 'dropping (mutable closed?) (immutable mutex) - (immutable not-empty) ;; condition: buffer has items - (immutable not-full)) ;; condition: buffer has space + (immutable not-empty) ;; signaled on put or close + (immutable not-full)) ;; signaled on get or close (protocol (lambda (new) (case-lambda - [() (new '() 0 #f (make-mutex) (make-condition) (make-condition))] - [(cap) (new '() cap #f (make-mutex) (make-condition) (make-condition))])))) + [() (new '() '() 0 0 'fixed #f + (make-mutex) (make-condition) (make-condition))] + [(cap) (new '() '() 0 cap 'fixed #f + (make-mutex) (make-condition) (make-condition))] + [(cap kind) (new '() '() 0 cap kind #f + (make-mutex) (make-condition) (make-condition))])))) (define (make-channel/buf n) (make-channel n)) + (define (make-channel/sliding n) + (when (or (not (integer? n)) (<= n 0)) + (error 'make-channel/sliding "sliding buffer needs positive capacity" n)) + (make-channel n 'sliding)) + + (define (make-channel/dropping n) + (when (or (not (integer? n)) (<= n 0)) + (error 'make-channel/dropping "dropping buffer needs positive capacity" n)) + (make-channel n 'dropping)) + (define (chan-closed? ch) (channel-closed? ch)) + (define (chan-kind ch) (channel-kind ch)) + (define (chan-empty? ch) + (with-mutex (channel-mutex ch) + (zero? (channel-count ch)))) + + ;; ---- queue helpers (must be called with channel-mutex held) ---- + + (define (q-empty? ch) (zero? (channel-count ch))) + + (define (q-full? ch) + (and (> (channel-capacity ch) 0) + (>= (channel-count ch) (channel-capacity ch)))) - (define (buffer-count ch) (length (channel-buffer ch))) + (define (q-enqueue! ch val) + (let ([cell (cons val '())]) + (cond + [(null? (channel-head ch)) + (channel-head-set! ch cell) + (channel-tail-set! ch cell)] + [else + (set-cdr! (channel-tail ch) cell) + (channel-tail-set! ch cell)]) + (channel-count-set! ch (+ (channel-count ch) 1)))) + + (define (q-dequeue! ch) + (let ([val (car (channel-head ch))]) + (channel-head-set! ch (cdr (channel-head ch))) + (when (null? (channel-head ch)) + (channel-tail-set! ch '())) + (channel-count-set! ch (- (channel-count ch) 1)) + val)) + + ;; ---- put ---- + ;; + ;; Three buffer policies: + ;; 'fixed — block the writer when the queue is at capacity + ;; 'sliding — drop the oldest item to make room (writer never blocks) + ;; 'dropping — silently drop the incoming value (writer never blocks) + ;; + ;; For unbuffered channels (capacity=0), q-full? is always true + ;; because count>=0>=0 is the "no slot" condition. We treat the + ;; unbuffered case identically to fixed — the writer blocks until + ;; a reader shows up. (define (chan-put! ch val) (with-mutex (channel-mutex ch) - (when (channel-closed? ch) - (error 'chan-put! "channel is closed")) - ;; Wait until buffer has space (or unbuffered: wait for receiver) (let loop () - (when (and (> (channel-capacity ch) 0) - (>= (buffer-count ch) (channel-capacity ch))) - (condition-wait (channel-not-full ch) (channel-mutex ch)) - (loop))) - ;; Enqueue - (channel-buffer-set! ch (append (channel-buffer ch) (list val))) - (condition-broadcast (channel-not-empty ch)))) + (cond + [(channel-closed? ch) + (error 'chan-put! "channel is closed")] + [(not (q-full? ch)) + (q-enqueue! ch val) + (condition-broadcast (channel-not-empty ch))] + [(eq? 'sliding (channel-kind ch)) + ;; Drop the oldest value to make room, then enqueue. + (q-dequeue! ch) + (q-enqueue! ch val) + (condition-broadcast (channel-not-empty ch))] + [(eq? 'dropping (channel-kind ch)) + ;; Silently drop the incoming value. + (void)] + [else ;; 'fixed (and the unbuffered degenerate case) + (condition-wait (channel-not-full ch) (channel-mutex ch)) + (loop)])))) + + (define (chan-try-put! ch val) + (with-mutex (channel-mutex ch) + (cond + [(channel-closed? ch) #f] + [(not (q-full? ch)) + (q-enqueue! ch val) + (condition-broadcast (channel-not-empty ch)) + #t] + [(eq? 'sliding (channel-kind ch)) + (q-dequeue! ch) + (q-enqueue! ch val) + (condition-broadcast (channel-not-empty ch)) + #t] + [(eq? 'dropping (channel-kind ch)) + ;; Drop silently but report success (no block). + #t] + [else #f]))) + + ;; ---- get ---- (define (chan-get! ch) (with-mutex (channel-mutex ch) (let loop () (cond - [(pair? (channel-buffer ch)) - (let ([val (car (channel-buffer ch))]) - (channel-buffer-set! ch (cdr (channel-buffer ch))) + [(not (q-empty? ch)) + (let ([val (q-dequeue! ch)]) (condition-broadcast (channel-not-full ch)) val)] [(channel-closed? ch) @@ -79,12 +201,14 @@ (define (chan-try-get ch) (with-mutex (channel-mutex ch) - (if (pair? (channel-buffer ch)) - (let ([val (car (channel-buffer ch))]) - (channel-buffer-set! ch (cdr (channel-buffer ch))) - (condition-broadcast (channel-not-full ch)) - val) - #f))) + (cond + [(not (q-empty? ch)) + (let ([val (q-dequeue! ch)]) + (condition-broadcast (channel-not-full ch)) + val)] + [else #f]))) + + ;; ---- close ---- (define (chan-close! ch) (with-mutex (channel-mutex ch) @@ -92,19 +216,25 @@ (condition-broadcast (channel-not-empty ch)) (condition-broadcast (channel-not-full ch)))) - ;; ========== Green threads ========== - ;; Simple thread-based implementation (engines for cooperative scheduling) + ;; ========== Process spawning ========== + ;; `go` and `go-named` are thin wrappers over `fork-thread`. Each + ;; spawn creates a real OS thread. There is no scheduler. (define (go thunk) (fork-thread thunk)) (define (go-named name thunk) + ;; Chez fork-thread takes no name argument; `name` is accepted + ;; for API compatibility and ignored at runtime. (fork-thread thunk)) (define (yield) + ;; Chez has no cooperative scheduler, so there is nothing to + ;; yield *to*. This is a 1ms sleep that relinquishes the current + ;; OS timeslice — useful for backoff in spin loops, not a + ;; replacement for coroutine yielding. (sleep (make-time 'time-duration 1000000 0))) - ;; Run a CSP system: execute thunk, return result (define (csp-run thunk) (thunk)) new file mode 100644 --- /dev/null +++ b/lib/std/csp/clj.sls @@ -0,0 +1,174 @@ +#!chezscheme +;;; (std csp clj) — Clojure `core.async`-compatible surface +;;; +;;; A thin renaming layer over `(std csp)`, `(std csp select)`, and +;;; `(std csp ops)` that exposes Clojure's short operator names: +;;; chan, >!, <!, >!!, <!!, close!, poll!, offer!, alts!, alts!!, +;;; alt!, alt!!, timeout, go, go-loop, to-chan, onto-chan, merge, +;;; split, pipe, mult, tap, untap, untap-all, pub, sub, unsub, +;;; unsub-all, pipeline, pipeline-blocking, pipeline-async, +;;; promise-chan, sliding-buffer, dropping-buffer. +;;; +;;; PARKING VS BLOCKING +;;; ------------------- +;;; In Clojure core.async, `>!` and `<!` "park" a go-block on a +;;; lightweight scheduler and `>!!` / `<!!` block an OS thread. In +;;; Jerboa every `go` is a real OS thread (there is no CPS transform +;;; and no green-thread scheduler), so parking and blocking collapse +;;; to the same operation. Both name pairs are provided for +;;; compatibility — they all reduce to `chan-put!` / `chan-get!`. +;;; +;;; GO AND THREAD +;;; ------------- +;;; `(go body ...)` spawns a thread, evaluates the body, puts the +;;; result onto a freshly made size-1 channel, and returns it. If the +;;; body raises an exception, the result channel is closed and any +;;; taker gets `(eof-object)`. `(go-loop ((var init) ...) body ...)` +;;; is `(go (let loop ([var init] ...) body ...))` with `loop` bound +;;; hygienically in the user's scope so bodies can tail-recur via +;;; `(loop ...)`. +;;; +;;; BUFFER FACTORIES +;;; ---------------- +;;; `(sliding-buffer n)` and `(dropping-buffer n)` return opaque +;;; buffer-spec records that `chan` recognizes as a request for a +;;; sliding or dropping channel. Passing an integer keeps the +;;; default (fixed) policy. + +(library (std csp clj) + (export + ;; Constructors + chan sliding-buffer dropping-buffer buffer-spec? + ;; Put / take / close / non-blocking + >! <! >!! <!! close! poll! offer! + ;; Select / timeout (re-exported from std csp select) + alts! alts!! alt! alt!! default timeout + ;; Go / thread + go go-loop clj-thread + ;; Collection bridges + to-chan onto-chan + ;; Composition + merge split pipe + ;; Broadcast + mult tap untap untap-all + ;; Topic + pub sub unsub unsub-all + ;; Pipelines + pipeline pipeline-blocking pipeline-async + ;; Promise + promise-chan) + + (import (except (chezscheme) merge) + (except (std csp) go go-named) + (std csp select) + (std csp ops)) + + ;; ====================================================== + ;; Buffer specs — opaque tags for chan to dispatch on + ;; ====================================================== + + (define-record-type buffer-spec + (fields (immutable kind) (immutable size))) + + (define (sliding-buffer n) (make-buffer-spec 'sliding n)) + (define (dropping-buffer n) (make-buffer-spec 'dropping n)) + + ;; ====================================================== + ;; chan — Clojure-style constructor + ;; ====================================================== + ;; + ;; Clojure supports (chan), (chan n), (chan buf), (chan n xform), + ;; (chan buf xform). Transducers aren't implemented yet — passing a + ;; third arg raises an error. + + (define chan + (case-lambda + [() (make-channel)] + [(x) + (cond + [(buffer-spec? x) + (case (buffer-spec-kind x) + [(sliding) (make-channel/sliding (buffer-spec-size x))] + [(dropping) (make-channel/dropping (buffer-spec-size x))] + [else (make-channel (buffer-spec-size x))])] + [(and (integer? x) (zero? x)) (make-channel)] + [(integer? x) (make-channel x)] + [else (error 'chan "expected buffer spec or non-negative integer" x)])] + [(_n _xform) + (error 'chan "transducers are not supported yet")])) + + ;; ====================================================== + ;; Put / take / close / poll / offer + ;; ====================================================== + + (define (>!! ch v) (chan-put! ch v)) + (define (<!! ch) (chan-get! ch)) + ;; parking vs blocking collapse in Jerboa + (define >! >!!) + (define <! <!!) + + (define (close! ch) (chan-close! ch)) + (define (poll! ch) (chan-try-get ch)) + (define (offer! ch v) (chan-try-put! ch v)) + + ;; ====================================================== + ;; go / go-loop / clj-thread + ;; ====================================================== + ;; + ;; `go` returns a size-1 channel with the body's result, closing + ;; on completion or exception. `clj-thread` is an alias — in + ;; Clojure `thread` spawns a blocking OS thread while `go` spawns a + ;; parked goroutine, but Jerboa has only OS threads so the + ;; distinction is moot. + + (define-syntax go + (syntax-rules () + [(_ body ...) + (let ([%result-ch (make-channel 1)]) + (fork-thread + (lambda () + (guard (exn [else (chan-close! %result-ch)]) + (let ([%v (let () body ...)]) + (chan-put! %result-ch %v) + (chan-close! %result-ch))))) + %result-ch)])) + + ;; go-loop expands to (go (let loop ([var init] ...) body ...)) + ;; with `loop` bound in the caller's scope (standard hygienic trick + ;; via datum->syntax) so the user can write (loop ...) to recur. + (define-syntax go-loop + (lambda (stx) + (syntax-case stx () + [(k ((var init) ...) body ...) + (with-syntax ([loop-id (datum->syntax #'k 'loop)]) + #'(go (let loop-id ([var init] ...) body ...)))]))) + + (define-syntax clj-thread + (syntax-rules () + [(_ body ...) (go body ...)])) + + ;; ====================================================== + ;; Clojure-named aliases for (std csp ops) procedures + ;; ====================================================== + + (define merge chan-merge) + (define split chan-split) + (define pipe chan-pipe-to) + + (define mult make-mult) + (define tap tap!) + (define untap untap!) + (define untap-all untap-all!) + + (define pub make-pub) + (define sub sub!) + (define unsub unsub!) + (define unsub-all unsub-all!) + + (define pipeline chan-pipeline) + (define pipeline-blocking chan-pipeline) + (define pipeline-async chan-pipeline-async) + + (define promise-chan make-promise-channel) + +) ;; end library new file mode 100644 --- /dev/null +++ b/lib/std/csp/ops.sls @@ -0,0 +1,457 @@ +#!chezscheme +;;; (std csp ops) — channel combinators (Phase 2 / Phase 3 of core-async.md) +;;; +;;; Everything on top of the `(std csp)` primitives that would be +;;; fiddly but not novel: collection bridges, merge/split/pipe, +;;; mult/tap/untap, pub/sub, pipeline, and promise-chan. Nothing +;;; here is thread-hot — the implementations spawn helper threads +;;; freely, preferring clarity over minimum OS-thread count. + +(library (std csp ops) + (export + ;; Collection bridges + to-chan onto-chan + chan-reduce chan-into + ;; Composition + chan-merge chan-split chan-pipe-to + ;; Broadcast: mult / tap / untap + make-mult mult? mult-source + tap! untap! untap-all! + ;; Topic routing: pub / sub + make-pub pub? pub-source + sub! unsub! unsub-all! + ;; Pipelines and promise + chan-pipeline chan-pipeline-async + make-promise-channel promise-channel? + promise-channel-put! promise-channel-get!) + + (import (chezscheme) + (std csp)) + + ;;; ====================================================== + ;;; Collection bridges + ;;; ====================================================== + + ;; (to-chan lst) → channel that emits each item, then closes. + ;; (to-chan lst buf) → with the given buffer size. + (define to-chan + (case-lambda + [(lst) (to-chan lst 0)] + [(lst buf-size) + (let ([ch (if (zero? buf-size) + (make-channel) + (make-channel buf-size))]) + (fork-thread + (lambda () + (for-each (lambda (x) (chan-put! ch x)) lst) + (chan-close! ch))) + ch)])) + + ;; (onto-chan ch lst) → pump lst onto ch, close ch. + ;; (onto-chan ch lst close?) → optionally leave ch open. + (define onto-chan + (case-lambda + [(ch lst) (onto-chan ch lst #t)] + [(ch lst close?) + (fork-thread + (lambda () + (for-each (lambda (x) (chan-put! ch x)) lst) + (when close? (chan-close! ch)))) + ch])) + + ;; (chan-reduce f init ch) → fold ch into a single value. + ;; + ;; Runs on the caller's thread — this is the blocking variant. + ;; For a parking variant, wrap it in (go (chan-reduce ...)). + (define (chan-reduce f init ch) + (let loop ([acc init]) + (let ([v (chan-get! ch)]) + (if (eof-object? v) + acc + (loop (f acc v)))))) + + ;; (chan-into container ch) → collect into a list or vector. + ;; + ;; The only containers supported for now are `'()` (list) and + ;; `(vector)` (vector). Matches Clojure's `(into [] ch)` idiom. + (define (chan-into container ch) + (cond + [(null? container) (chan->list ch)] + [(pair? container) + ;; Prepend onto an existing list (reverse-append at the end + ;; to preserve incoming order). + (let loop ([acc container]) + (let ([v (chan-get! ch)]) + (if (eof-object? v) + (reverse (let rev ([a acc] [r '()]) + (if (null? a) r (rev (cdr a) (cons (car a) r))))) + (loop (cons v acc)))))] + [(vector? container) + (list->vector (append (vector->list container) (chan->list ch)))] + [else (error 'chan-into "unsupported container" container)])) + + ;;; ====================================================== + ;;; Composition — merge / split / pipe-to + ;;; ====================================================== + + ;; (chan-merge chans) → new channel receiving everything + ;; from every input, closes when the + ;; last input closes. + ;; (chan-merge chans buf) → with explicit buffer size. + (define chan-merge + (case-lambda + [(chans) (chan-merge chans 0)] + [(chans buf-size) + (let ([out (if (zero? buf-size) + (make-channel) + (make-channel buf-size))] + [remaining (length chans)] + [lk (make-mutex)]) + (for-each + (lambda (ch) + (fork-thread + (lambda () + (let loop () + (let ([v (chan-get! ch)]) + (cond + [(eof-object? v) + (with-mutex lk + (set! remaining (- remaining 1)) + (when (zero? remaining) + (chan-close! out)))] + [else (chan-put! out v) (loop)])))))) + chans) + out)])) + + ;; (chan-split pred ch) → (list true-ch false-ch). + ;; + ;; Each incoming value goes to true-ch if `(pred v)` is truthy, + ;; false-ch otherwise. Both downstream channels close when the + ;; input closes. + (define (chan-split pred ch) + (let ([tch (make-channel)] + [fch (make-channel)]) + (fork-thread + (lambda () + (let loop () + (let ([v (chan-get! ch)]) + (cond + [(eof-object? v) + (chan-close! tch) + (chan-close! fch)] + [(pred v) (chan-put! tch v) (loop)] + [else (chan-put! fch v) (loop)]))))) + (list tch fch))) + + ;; (chan-pipe-to from to) → close `to` on EOF + ;; (chan-pipe-to from to close?) → don't close `to` when + ;; from closes if #f. + ;; + ;; Named `chan-pipe-to` to avoid the pre-existing `chan-pipe` + ;; in `(std csp)` which takes a transform proc and has a + ;; different signature. + (define chan-pipe-to + (case-lambda + [(from to) (chan-pipe-to from to #t)] + [(from to close?) + (fork-thread + (lambda () + (let loop () + (let ([v (chan-get! from)]) + (cond + [(eof-object? v) + (when close? (chan-close! to))] + [else (chan-put! to v) (loop)]))))) + to])) + + ;;; ====================================================== + ;;; Broadcast — mult / tap / untap + ;;; ====================================================== + ;; + ;; A `mult` fans out every value from a source channel to a + ;; dynamically maintained set of subscriber channels. Subscribers + ;; are added with `tap!` and removed with `untap!`. + ;; + ;; Semantics trade-off: this implementation blocks the fan-out + ;; thread on the slowest subscriber. Clojure's core.async allows + ;; a timeout + drop policy per sub. For v1 we document the + ;; slow-subscriber hazard and expect users to hand slow subs a + ;; sliding / dropping buffer (see `make-channel/sliding`). + + (define-record-type %mult + (fields (immutable source) (mutable subs) (immutable lock))) + + (define (make-mult source) + (let ([m (make-%mult source '() (make-mutex))]) + (fork-thread + (lambda () + (let loop () + (let ([v (chan-get! source)]) + (cond + [(eof-object? v) + (with-mutex (%mult-lock m) + (for-each + (lambda (s) + ;; chan-close! errors if already closed elsewhere + ;; — guard so untap of a still-open sub doesn't + ;; nuke the fan-out thread. + (guard (_ [else (void)]) (chan-close! s))) + (%mult-subs m)))] + [else + (let ([subs (with-mutex (%mult-lock m) (%mult-subs m))]) + (for-each + (lambda (s) + (guard (_ [else (void)]) (chan-put! s v))) + subs)) + (loop)]))))) + m)) + + (define (mult? x) (%mult? x)) + (define (mult-source m) (%mult-source m)) + + (define (tap! m ch) + (with-mutex (%mult-lock m) + (%mult-subs-set! m (cons ch (%mult-subs m)))) + ch) + + (define (untap! m ch) + (with-mutex (%mult-lock m) + (%mult-subs-set! m + (remp (lambda (x) (eq? x ch)) (%mult-subs m)))) + ch) + + (define (untap-all! m) + (with-mutex (%mult-lock m) + (%mult-subs-set! m '()))) + + ;;; ====================================================== + ;;; Topic routing — pub / sub + ;;; ====================================================== + ;; + ;; A `pub` is a source channel + a `topic-fn` + a topic→mult map. + ;; Subscribers register for a topic; the dispatcher reads the + ;; source, computes `(topic-fn v)`, and forwards to the matching + ;; topic mult. Each topic gets its own fan-out mult lazily. + + (define-record-type %pub + (fields + (immutable source) + (immutable topic-fn) + (mutable topics) ;; alist of topic → (cons topic-ch topic-mult) + (immutable lock))) + + (define (make-pub source topic-fn) + (let ([p (make-%pub source topic-fn '() (make-mutex))]) + (fork-thread + (lambda () + (let loop () + (let ([v (chan-get! source)]) + (cond + [(eof-object? v) + ;; Close each topic channel so its mult drains and + ;; closes every downstream subscriber in turn. + (with-mutex (%pub-lock p) + (for-each + (lambda (entry) + (guard (_ [else (void)]) + (chan-close! (car (cdr entry))))) + (%pub-topics p)))] + [else + (let ([topic ((%pub-topic-fn p) v)]) + (let ([entry (with-mutex (%pub-lock p) + (assoc topic (%pub-topics p)))]) + (when entry + (guard (_ [else (void)]) + (chan-put! (car (cdr entry)) v))) + (loop)))]))))) + p)) + + (define (pub? x) (%pub? x)) + (define (pub-source p) (%pub-source p)) + + (define (pub-get-or-make-mult! p topic) + (with-mutex (%pub-lock p) + (let ([found (assoc topic (%pub-topics p))]) + (cond + [found (cddr found)] + [else + (let* ([tch (make-channel 16)] + [tm (make-mult tch)]) + (%pub-topics-set! p + (cons (cons topic (cons tch tm)) (%pub-topics p))) + tm)])))) + + (define (sub! p topic ch) + (tap! (pub-get-or-make-mult! p topic) ch) + ch) + + (define (unsub! p topic ch) + (with-mutex (%pub-lock p) + (let ([entry (assoc topic (%pub-topics p))]) + (when entry + (untap! (cddr entry) ch)))) + ch) + + (define (unsub-all! p) + (with-mutex (%pub-lock p) + (for-each (lambda (entry) (untap-all! (cddr entry))) + (%pub-topics p)))) + + ;;; ====================================================== + ;;; Pipelines — N workers, ordered output + ;;; ====================================================== + ;; + ;; `chan-pipeline n out f in` spawns `n` worker threads that read + ;; from `in`, apply `f`, and hand the result to `out`. To preserve + ;; input order across workers each item is tagged with an index; + ;; a dedicated reassembler holds out-of-order results until their + ;; turn comes up. When the input closes the reassembler waits for + ;; in-flight workers to drain and then closes `out`. + ;; + ;; `f` is a plain unary procedure (not a transducer). For the + ;; transducer variant call `chan-pipeline-xf` once it's added. + ;; + ;; `chan-pipeline-async af` is the async variant: `af` takes two + ;; args `(item result-ch)` and is expected to `chan-put!` its + ;; result(s) then `chan-close!` the result channel. The pipeline + ;; reads results off the per-item result channel until it closes. + + (define (chan-pipeline n out f in) + (pipeline-run n out in + (lambda (item result-ch) + (chan-put! result-ch (f item)) + (chan-close! result-ch)))) + + (define (chan-pipeline-async n out af in) + (pipeline-run n out in af)) + + (define (pipeline-run n out in af) + ;; Step 1: tag each input with a monotonically increasing index + ;; and fan it out to a queue-channel that workers read from. + (let ([job-ch (make-channel (* n 2))] + [result-ch (make-channel (* n 2))] + [job-lock (make-mutex)]) + ;; Producer: read `in`, tag, enqueue + (fork-thread + (lambda () + (let loop ([i 0]) + (let ([v (chan-get! in)]) + (cond + [(eof-object? v) + (chan-close! job-ch)] + [else + (chan-put! job-ch (cons i v)) + (loop (+ i 1))]))))) + ;; Workers + (let loop-workers ([k 0]) + (when (< k n) + (fork-thread + (lambda () + (let loop () + (let ([job (chan-get! job-ch)]) + (cond + [(eof-object? job) (void)] + [else + (let* ([idx (car job)] + [item (cdr job)] + [rc (make-channel 1)]) + (af item rc) + ;; Drain the per-item result channel into the + ;; shared result-ch, each result tagged with idx. + (let drain () + (let ([r (chan-get! rc)]) + (unless (eof-object? r) + (chan-put! result-ch (cons idx r)) + (drain)))) + (loop))]))))) + (loop-workers (+ k 1)))) + ;; Reassembler: track next expected index and a pending table. + (fork-thread + (lambda () + (let ([next 0] [pending (make-eqv-hashtable)] [seen-eof 0]) + (let loop () + (let ([tagged (chan-get! result-ch)]) + (cond + [(eof-object? tagged) + (chan-close! out)] + [else + (let ([i (car tagged)] [v (cdr tagged)]) + (hashtable-set! pending i + (cons v (hashtable-ref pending i '())))) + (let emit () + (let ([ready (hashtable-ref pending next #f)]) + (when ready + (for-each (lambda (x) (chan-put! out x)) (reverse ready)) + (hashtable-delete! pending next) + (set! next (+ next 1)) + (emit)))) + (loop)])))))) + ;; Closer: when all workers are done (job-ch drained) we close + ;; the result-ch so the reassembler can close `out`. + (fork-thread + (lambda () + ;; Wait for every worker to be idle. A simple heuristic: + ;; once job-ch is closed AND empty, spin-wait briefly and + ;; then close result-ch. Workers that are mid-task will + ;; push to result-ch before noticing — the reassembler + ;; reads eagerly so nothing is lost. + (let wait () + (unless (and (chan-closed? job-ch) + (chan-empty? job-ch)) + (sleep (make-time 'time-duration 1000000 0)) ;; 1ms + (wait))) + ;; Grace period for workers in the middle of their last job. + (sleep (make-time 'time-duration 5000000 0)) ;; 5ms + (chan-close! result-ch))) + out)) + + ;;; ====================================================== + ;;; promise-chan + ;;; ====================================================== + ;; + ;; Semantics: + ;; - First put wins; subsequent puts are silently dropped. + ;; - Every taker gets the winning value, even after it's set. + ;; - On close-without-put, every pending and future taker gets + ;; (eof-object). + ;; - `chan-get!` semantics: block until set or closed. + + (define-record-type %promise-channel + (fields + (immutable lock) + (immutable ready) ;; condition: broadcast on first put/close + (mutable value) + (mutable set?) + (mutable closed?))) + + (define (make-promise-channel) + (make-%promise-channel + (make-mutex) (make-condition) #f #f #f)) + + (define (promise-channel? x) (%promise-channel? x)) + + (define (promise-channel-put! pc v) + (with-mutex (%promise-channel-lock pc) + (cond + [(%promise-channel-closed? pc) #f] + [(%promise-channel-set? pc) #f] + [else + (%promise-channel-value-set! pc v) + (%promise-channel-set?-set! pc #t) + (condition-broadcast (%promise-channel-ready pc)) + #t]))) + + (define (promise-channel-get! pc) + (with-mutex (%promise-channel-lock pc) + (let loop () + (cond + [(%promise-channel-set? pc) + (%promise-channel-value pc)] + [(%promise-channel-closed? pc) + (eof-object)] + [else + (condition-wait (%promise-channel-ready pc) + (%promise-channel-lock pc)) + (loop)])))) + +) ;; end library new file mode 100644 --- /dev/null +++ b/lib/std/csp/select.sls @@ -0,0 +1,275 @@ +#!chezscheme +;;; (std csp select) — multi-channel select / alts! / timeout +;;; +;;; Implements the core.async-style rendezvous operators over +;;; `(std csp)` channels. Built on a spin-poll loop with exponential +;;; back-off: each iteration tries every spec in (optionally +;;; randomized) order via `chan-try-get` / `chan-try-put!`, then +;;; sleeps a short, growing interval if nothing was ready. The +;;; back-off caps at ~10ms, so latency is bounded but idle CPU is +;;; negligible. +;;; +;;; alts! specs +;;; ----------- +;;; Each spec is either: +;;; ch -- a take spec: `(chan-try-get ch)` → val on hit +;;; (list ch val) -- a put spec: `(chan-try-put! ch val)` → #t on hit +;;; +;;; alts!/alts!! return (list result ch), matching Clojure's +;;; [val ch] shape. On a take spec, `result` is the received value +;;; (or `(eof-object)` if the channel closed). On a put spec, +;;; `result` is #t on success. +;;; +;;; Options (pass after the specs list as plain symbols): +;;; 'priority — try specs in order instead of a random permutation +;;; 'default val — if NO spec is ready at the first poll, return +;;; (list val 'default) immediately +;;; +;;; Plain symbols are used instead of colon-prefixed `:priority` / +;;; `:default` because the Jerboa reader reserves a leading `:` for +;;; Gerbil-style module paths — `:priority` becomes `(priority)`, +;;; not a symbol — so any user code calling alts!! from a .ss file +;;; wouldn't be able to spell the option. +;;; +;;; timeout +;;; ------- +;;; `(timeout ms)` returns a fresh channel that closes itself after +;;; `ms` milliseconds. Taking from it with `chan-get!` or inside +;;; `alts!` yields `(eof-object)` once the deadline fires. Each +;;; call spawns one helper thread — fine for low-rate timeouts, +;;; see the Phase-4 notes in core-async.md for a timer-wheel +;;; replacement. + +(library (std csp select) + (export + ;; Event bridge (expose for advanced use) + chan-recv-evt chan-send-evt + ;; Rendezvous + alts! alts!! + ;; Macro sugar — Clojure's alt!/alt!! + alt! alt!! + ;; Auxiliary keyword for alt!/alt!! default clauses. Users + ;; who want `(default ...)` as a clause head need to import it + ;; (it's re-exported from (std csp clj) for convenience). + default + ;; Timeout channel + timeout timeout-channel) + + (import (chezscheme) + (std csp) + (std event)) + + ;; ========================================================== + ;; Event bridge — expose a channel as a (std event) event. + ;; poll uses chan-try-get / chan-try-put!, sync uses blocking + ;; chan-get! / chan-put!. sync returns the same shape as poll + ;; so (choice …) works identically whether it's spin-polling + ;; or settling on a single event. + ;; ========================================================== + + (define (chan-recv-evt ch) + (make-event + ;; poll — return the received value on hit, or #f + (lambda () + (cond + [(chan-try-get ch) + => (lambda (v) (list v ch))] + [(chan-closed? ch) + ;; Drained + closed → propagate eof so the picker can