csp: non-blocking callback-style put! / take!
ober
ef4022ccc4b54bc849e13fa4106d3f95c0d5c3c8
--- a/docs/clojure-remaining.md +++ b/docs/clojure-remaining.md @@ -435,6 +435,15 @@ soak testing. ### 3.4 Callback-style `put!` / `take!` +**Status.** [landed] `(std csp ops)` exports `put!` and `take!` and the +Clojure names are re-exported from `(std csp clj)`. Both spawn one +helper thread per callback (documented thread-explosion hazard) and +guard the user callback so a raising callback prints a warning to +`current-error-port` instead of silently killing the helper thread. +Exercised by `tests/test-csp.ss` — seven tests covering successful put, +put on a closed channel, fire-and-forget, take that sees a value, take +that sees `eof-object` on close, and a full round-trip. + **What Clojure does.** In addition to the blocking / parking `>!!`/`>!` and `<!!`/`<!`, core.async offers non-blocking *callback* forms: @@ -1683,7 +1692,7 @@ in this doc. **[deferred]** items are non-goals. | `(chan n xform)` | [current] `(std csp clj)` | §3.1 landed | | `mix`/`admix`/`toggle` | [gap] | §3.2 | | Timer wheel | [gap] | §3.3 | -| `put!`/`take!` with callbacks | [gap] | §3.4 | +| `put!`/`take!` with callbacks | [current] `(std csp ops)` | §3.4 landed | | `async/reduce`, `onto-chan!` | [gap] | §3.5 | | `split` n-way | [gap] | §3.6 | | Mult slow-sub policies | [gap] | §3.7 | --- a/lib/std/csp/clj.sls +++ b/lib/std/csp/clj.sls @@ -41,6 +41,8 @@ chan sliding-buffer dropping-buffer buffer-spec? ;; Put / take / close / non-blocking >! <! >!! <!! close! poll! offer! + ;; Async callback forms + put! take! ;; Select / timeout (re-exported from std csp select) alts! alts!! alt! alt!! default timeout ;; Go / thread --- a/lib/std/csp/ops.sls +++ b/lib/std/csp/ops.sls @@ -12,6 +12,8 @@ ;; Collection bridges to-chan onto-chan chan-reduce chan-into + ;; Non-blocking callbacks + put! take! ;; Composition chan-merge chan-split chan-pipe-to ;; Broadcast: mult / tap / untap @@ -91,6 +93,74 @@ [else (error 'chan-into "unsupported container" container)])) ;;; ====================================================== + ;;; Callback-style put! / take! + ;;; ====================================================== + ;; + ;; Non-blocking versions of put / take that spawn a helper thread, + ;; perform the operation, then invoke the user's callback with the + ;; result. These are the Clojure core.async `put!` / `take!` — the + ;; foundation for bridging callback-based APIs (Netty, raw sockets, + ;; AJAX shims) into channel pipelines without writing a go-block + ;; per request. + ;; + ;; Semantics: + ;; - (put! ch v) — fire-and-forget async put. No callback. + ;; Errors from a closed channel are swallowed. + ;; - (put! ch v fn) — async put; callback fn receives #t on a + ;; successful put and #f if the channel was + ;; closed before the put could land. + ;; - (take! ch fn) — async take; callback fn receives the value + ;; that arrives, or the eof-object when the + ;; channel closes without delivering one. + ;; + ;; Both `fn` callbacks run on the helper thread — they should not + ;; do heavy work, since each outstanding callback keeps one thread + ;; alive. If the callback itself raises an exception, a warning is + ;; printed to current-error-port and the helper thread exits (the + ;; exception does NOT propagate to the caller of put!/take!). + ;; + ;; WARNING — thread-per-callback. At high request rates the naive + ;; implementation creates one short-lived OS thread per callback. + ;; For production workloads consider batching onto a single + ;; dedicated dispatch thread. See core-async.md §3.4. + + (define (%run-callback who fn . args) + ;; Guarded callback invocation — keeps a misbehaved user callback + ;; from bringing down the helper thread silently with no trace. + (guard (exn [else + (fprintf (current-error-port) + "~a callback raised: ~a~%" + who + (if (message-condition? exn) + (condition-message exn) + exn))]) + (apply fn args))) + + (define put! + (case-lambda + [(ch v) + (fork-thread + (lambda () + ;; fire-and-forget: swallow closed-channel errors. + (guard (_ [else (void)]) (chan-put! ch v)))) + (void)] + [(ch v fn) + (fork-thread + (lambda () + (let ([ok? (guard (_ [else #f]) + (chan-put! ch v) + #t)]) + (%run-callback 'put! fn ok?)))) + (void)])) + + (define (take! ch fn) + (fork-thread + (lambda () + (let ([v (chan-get! ch)]) + (%run-callback 'take! fn v)))) + (void)) + + ;;; ====================================================== ;;; Composition — merge / split / pipe-to ;;; ====================================================== --- a/tests/test-csp.ss +++ b/tests/test-csp.ss @@ -326,6 +326,75 @@ (chan->list ch)) '(1 oops 2)) +;;; ======== Callback-style put! / take! (Phase C.1) ======== + +(test "put! with callback — successful put reports #t" + (let ([ch (chan 4)] + [result (make-channel 1)]) + (put! ch 'hello (lambda (ok?) (chan-put! result ok?) (chan-close! result))) + (let ([ok? (chan-get! result)]) + (list ok? (chan-get! ch)))) + '(#t hello)) + +(test "put! with callback — closed channel reports #f" + (let ([ch (chan 1)] + [result (make-channel 1)]) + (chan-close! ch) + (put! ch 'bye (lambda (ok?) (chan-put! result ok?) (chan-close! result))) + (chan-get! result)) + #f) + +(test "put! fire-and-forget (2-arg form)" + (let ([ch (chan 4)]) + (put! ch 'fire) + (sleep (millis 20)) ;; give helper thread a chance + (chan-get! ch)) + 'fire) + +(test "take! delivers arriving value" + (let ([ch (chan 1)] + [seen (make-channel 1)]) + (take! ch (lambda (v) (chan-put! seen v) (chan-close! seen))) + (chan-put! ch 42) + (chan-get! seen)) + 42) + +(test "take! delivers eof on close" + (let ([ch (chan 1)] + [seen (make-channel 1)]) + (take! ch (lambda (v) + (chan-put! seen (if (eof-object? v) 'eof v)) + (chan-close! seen))) + (chan-close! ch) + (chan-get! seen)) + 'eof) + +(test "take! runs after put! — full async round-trip" + (let ([ch (chan 4)] + [out (make-channel 1)]) + (take! ch (lambda (v) (chan-put! out (+ v 100)) (chan-close! out))) + (put! ch 7) + (chan-get! out)) + 107) + +(test "put! callback runs for buffered channel without blocking caller" + ;; The caller should NOT block even if the channel buffer fills up + ;; — put! delegates the wait to a helper thread. Here we fill a + ;; size-1 channel with one item; the second put! must return + ;; immediately to the caller while waiting in a helper thread, and + ;; fire its callback only after a reader drains the channel. + (let ([ch (chan 1)] + [tag-ch (make-channel 4)]) + (chan-put! ch 'first) + (put! ch 'second + (lambda (ok?) (chan-put! tag-ch (if ok? 'second-delivered 'failed)))) + (chan-put! tag-ch 'after-put-returned) ;; should land BEFORE 'second-delivered + (let ([a (chan-get! tag-ch)]) ;; drain tag before reading ch + (chan-get! ch) ;; unblock helper: makes room + (let ([b (chan-get! tag-ch)]) + (list a b)))) + '(after-put-returned second-delivered)) + ;;; Summary (printf "~%CSP: ~a passed, ~a failed~%" pass fail)