fiber: land all 7 gaps — cancellation, locals, join, link, select, timeouts, groups
ober
303674a864cbe0b6ea76c270fc7a24cb491e8786
new file mode 100644 --- /dev/null +++ b/docs/fix-fibers.md @@ -0,0 +1,291 @@ +# Fiber Gaps — Implementation Plan + +Status: **draft** (2026-04-11) + +The `(std fiber)` core (668 lines, 18 tests) is solid: M:N scheduling, engine-based preemption, cooperative yield, fiber-aware channels, sleep with timer queue. This document covers the gaps and how to fill them, ordered by value. + +--- + +## 1. Fix `cpu-count` (trivial) + +**Gap**: `fiber.sls:221-222` hardcodes `(if (threaded?) 4 1)`. + +**Fix**: Reuse the real detection from `actor/scheduler.sls:135-146`, which reads `/proc/cpuinfo` with a fallback to 4. + +Options: +- (a) Factor the scheduler's `cpu-count` into a shared `(std misc cpu)` module that both import. +- (b) Inline the same `/proc/cpuinfo` reader into `fiber.sls`. + +Prefer (a) — one source of truth, and other modules (engine pool, task groups) can use it too. + +**Effort**: ~30 minutes. **Tests**: Verify `(make-fiber-runtime)` picks up actual core count. + +--- + +## 2. Fiber-local storage + +**Gap**: No per-fiber variable bindings. `current-fiber` and `current-fiber-runtime` are `make-thread-parameter`, which is OS-thread-scoped. Since multiple fibers share one OS worker thread, a fiber-local needs its own mechanism. + +**Design**: A `fiber-parameter` that stores values keyed by fiber ID. + +```scheme +(define (make-fiber-parameter default) + (let ([store (make-eq-hashtable)] + [mx (make-mutex)]) + (case-lambda + [() ;; read + (let ([f (current-fiber)]) + (if f + (begin (mutex-acquire mx) + (let ([v (hashtable-ref store (fiber-id f) default)]) + (mutex-release mx) v)) + default))] + [(val) ;; write + (let ([f (current-fiber)]) + (unless f (error 'fiber-parameter "not in a fiber")) + (mutex-acquire mx) + (hashtable-set! store (fiber-id f) val) + (mutex-release mx))]))) +``` + +**Cleanup**: When a fiber completes (`mark-fiber-done!`), sweep its entries from all registered fiber-parameters. Keep a global weak list of all live fiber-parameters, or add a per-fiber cleanup hook list. + +**Convenience macro**: +```scheme +(define-syntax fiber-parameterize + (syntax-rules () + [(_ ([fp val] ...) body ...) + (let ([old-fp (fp)] ...) + (dynamic-wind + (lambda () (fp val) ...) + (lambda () body ...) + (lambda () (fp old-fp) ...)))])) +``` + +**Effort**: ~2 hours. **Tests**: fiber-local isolation across concurrent fibers on same worker thread, cleanup after fiber completion. + +--- + +## 3. Fiber cancellation + +**Gap**: No way to cancel a running fiber from outside. Once spawned, a fiber runs until its thunk returns or raises. + +**Design**: Cooperative cancellation via a cancel token (same pattern as `task.sls:36-53`). + +Add to the fiber record: +```scheme +(mutable cancelled?) ;; boolean, checked at yield/sleep/channel-wait +``` + +API: +```scheme +(fiber-cancel! f) ;; set cancelled? flag, wake if parked +(fiber-cancelled? f) ;; check flag +(fiber-check-cancelled!) ;; called inside fiber — raises &fiber-cancelled if set +``` + +**Cancellation points**: `fiber-yield`, `fiber-sleep`, `fiber-channel-recv`, and `fiber-channel-send` all check the cancelled flag before parking. Raise `&fiber-cancelled` condition. + +**Force-cancel after timeout**: Like the actor supervisor's graceful shutdown (`supervisor.sls:215-240`), allow a deadline after which the fiber's engine is simply not resumed. + +```scheme +(fiber-cancel! f) ;; cooperative — sets flag +(fiber-cancel! f timeout-ms) ;; cooperative, then force-abandon after timeout +``` + +**Condition type**: +```scheme +(define-condition-type &fiber-cancelled &serious + make-fiber-cancelled fiber-cancelled? + (fiber-id cancelled-fiber-id)) +``` + +**Effort**: ~3 hours. **Tests**: cancel parked fiber, cancel running fiber at yield point, cancel with timeout, double-cancel idempotent. + +--- + +## 4. Error propagation + +**Gap**: Exceptions in a fiber are silently captured in `fiber-result` (`fiber.sls:436-441`). No notification to parent or supervisor. + +**Design**: Two mechanisms, matching the actor system's patterns. + +### 4a. fiber-join (blocking result retrieval) + +```scheme +(fiber-join f) ;; block current fiber until f completes, return result +(fiber-join f timeout-ms) ;; with timeout, raises &fiber-timeout on expiry +``` + +If `f` completed with an exception, `fiber-join` re-raises it in the joining fiber. Implementation: add a "join-waiters" list to the fiber record; `mark-fiber-done!` wakes them. + +### 4b. fiber-link (Erlang-style crash propagation) + +```scheme +(fiber-link! f) ;; link current fiber to f — if f dies with error, current fiber gets &fiber-linked-crash +(fiber-unlink! f) +``` + +When a linked fiber crashes, all linked fibers receive a `&fiber-linked-crash` condition at their next cancellation point. Simpler than full OTP monitors but covers the "if my child dies, I die" use case. + +**Condition types**: +```scheme +(define-condition-type &fiber-timeout &serious + make-fiber-timeout fiber-timeout? + (fiber-id timeout-fiber-id)) + +(define-condition-type &fiber-linked-crash &serious + make-fiber-linked-crash fiber-linked-crash? + (source-fiber-id linked-crash-source) + (original-condition linked-crash-condition)) +``` + +**Effort**: 4a ~2 hours, 4b ~3 hours. **Tests**: join on completed fiber, join on crashed fiber re-raises, join timeout, link propagation. + +--- + +## 5. Fiber-channel select (`fiber-select`) + +**Gap**: No way to wait on multiple fiber-channels simultaneously. + +**Design**: Adapt the CSP select spin-poll pattern (`csp/select.sls:166-174`) for fiber-channels. + +```scheme +(fiber-select + [ch1 val => (handle-val val)] ;; recv from ch1 + [ch2 :send msg => (handle-sent)] ;; send msg to ch2 + [:timeout 5000 => (handle-timeout)] ;; optional timeout clause + [:default => (handle-none)]) ;; optional non-blocking clause +``` + +**Implementation**: Macro expands to a loop that: +1. Try each clause's channel with `fiber-channel-try-recv` / `fiber-channel-try-send`. +2. If any succeeds, evaluate its body and return. +3. If `:default` clause exists and nothing ready, run it. +4. Otherwise, park the fiber on all channels' waiter lists, use a shared gate so that whichever channel fires first wakes the fiber. + +The shared-gate approach avoids the spin-poll cost: register the fiber on multiple channels, first one to wake it wins, others remove the stale waiter entry. + +**Alternative**: Integrate with `(std event)` by providing `fiber-recv-evt` / `fiber-send-evt` that return event objects compatible with `choice` and `sync`. This is cleaner but couples the two modules. + +Recommend: start with the macro (self-contained in `(std fiber)`), add event integration later. + +**Effort**: ~4 hours. **Tests**: select across 2+ channels, select with timeout, select with default, select with send+recv mix. + +--- + +## 6. Structured concurrency (`with-fiber-group`) + +**Gap**: No scoped lifecycle management. Fibers spawned with `fiber-spawn*` are fire-and-forget; no automatic cleanup, no "wait for all children" scope. + +**Design**: Follows the pattern from `task.sls` and `concur/structured.sls`. + +```scheme +(with-fiber-group + (lambda (group) + (fiber-group-spawn group (lambda () (do-work-a))) + (fiber-group-spawn group (lambda () (do-work-b))) + ;; implicit: waits for all children to complete + ;; if any child raises, cancels siblings, re-raises in parent + )) +``` + +**Semantics**: +- `with-fiber-group` creates a group, evaluates the body, then blocks until all spawned fibers complete. +- If any fiber in the group raises an unhandled exception, all other fibers in the group are cancelled (cooperative), and the exception is re-raised in the calling fiber. +- If the calling fiber is itself cancelled, all children are cancelled. +- Cleanup is guaranteed via `dynamic-wind`. + +**Group record**: +```scheme +(define-record-type fiber-group + (fields + (mutable fibers) ;; list of child fibers + (mutable first-exn) ;; first exception, or #f + (mutable cancelled?) + (immutable mutex) + (immutable all-done) ;; condition variable + (mutable done-count) + (mutable total-count))) +``` + +**Effort**: ~4 hours. **Tests**: all-succeed, first-error-cancels-rest, parent-cancel-propagates, nested groups. + +--- + +## 7. Fiber-aware timeouts + +**Gap**: `fiber-sleep` parks for a duration, but there's no "do X or timeout" pattern. + +**Design**: Build on `fiber-select` (gap 5) plus a timeout channel. + +```scheme +(define (fiber-timeout ms) + ;; Returns a fiber-channel that receives (void) after ms milliseconds. + ;; Uses the existing timer-queue infrastructure. + (let ([ch (make-fiber-channel 1)] + [rt (current-fiber-runtime)]) + (let* ([now (current-time 'time-utc)] + [deadline (add-duration now + (make-time 'time-duration + (* (fxmod ms 1000) 1000000) + (fxquotient ms 1000)))]) + ;; Spawn a tiny fiber that sleeps then sends + (fiber-spawn* (lambda () + (fiber-sleep ms) + (fiber-channel-try-send ch (void))))) + ch)) +``` + +Usage with `fiber-select`: +```scheme +(fiber-select + [work-ch result => (process result)] + [(fiber-timeout 5000) _ => (error 'timeout "took too long")]) +``` + +Better approach (no extra fiber): register directly on the timer queue, wake a channel when the deadline fires. Requires a small extension to the timer queue to fire arbitrary callbacks, not just wake fibers. + +**Effort**: ~1 hour (spawn approach), ~2 hours (timer-queue callback approach). **Tests**: timeout fires, timeout not needed (work completes first). + +--- + +## Implementation Order + +| Priority | Gap | Effort | Dependencies | +|----------|-----|--------|--------------| +| 1 | cpu-count fix | 30 min | none | +| 2 | Fiber cancellation | 3 hours | none | +| 3 | Fiber-local storage | 2 hours | none | +| 4 | Error propagation (fiber-join) | 2 hours | cancellation (for timeout variant) | +| 5 | Fiber-channel select | 4 hours | none (but better with timeouts) | +| 6 | Fiber-aware timeouts | 2 hours | select | +| 7 | Structured concurrency | 4 hours | cancellation + error propagation | +| 8 | Error propagation (fiber-link) | 3 hours | cancellation | + +**Total**: ~20 hours of implementation. + +Gaps 1-4 are independent and can be done in any order. Gap 5 (select) unlocks gap 6 (timeouts). Gap 7 (structured concurrency) needs gaps 2 + 4. + +--- + +## Existing Infrastructure to Reuse + +| Need | Source | Path | +|------|--------|------| +| Real CPU detection | actor scheduler | `lib/std/actor/scheduler.sls:135-146` | +| Cancel tokens | task groups | `lib/std/task.sls:36-53` | +| Supervision patterns | actor supervisor | `lib/std/actor/supervisor.sls` | +| Spin-poll select | CSP select | `lib/std/csp/select.sls:166-174` | +| Timer wheel | CSP select | `lib/std/csp/select.sls:299-410` | +| Event abstraction | event system | `lib/std/event.sls:43-61` | +| Error aggregation | task groups | `lib/std/task.sls:121-149` | +| Hierarchical cleanup | custodians | `lib/std/misc/custodian.sls` | + +--- + +## What's NOT in scope + +- **Work stealing**: The current run-queue is a single shared FIFO. A per-worker deque with stealing (like `actor/scheduler.sls`) would improve cache locality under heavy load, but the current design works well up to ~10K fibers. Optimize later if profiling shows contention. +- **Fiber migration**: Moving a parked fiber from one worker to another. Engine continuations are not trivially portable across threads in Chez. Not needed for correctness, only for load balancing (which work-stealing would address). +- **Async I/O integration**: Tying fiber parking to epoll/kqueue so that file/socket readiness wakes a fiber. This is a large project (essentially an event loop runtime) and orthogonal to the gaps above. --- a/lib/std/actor/scheduler.sls +++ b/lib/std/actor/scheduler.sls @@ -16,7 +16,7 @@ current-scheduler default-scheduler cpu-count) - (import (chezscheme) (std actor deque)) + (import (chezscheme) (std actor deque) (std misc cpu)) ;; Per-worker state (one per OS thread in the pool) (define-record-type worker @@ -132,17 +132,4 @@ (with-mutex (scheduler-mutex sched) (condition-broadcast (scheduler-work-available sched)))) - ;; Read CPU count from /proc/cpuinfo on Linux; fallback to 4 - (define (cpu-count) - (guard (exn [#t 4]) - (let ([p (open-input-file "/proc/cpuinfo")]) - (let loop ([n 0]) - (let ([line (get-line p)]) - (cond - [(eof-object? line) (close-port p) (fxmax n 1)] - [(and (fx>= (string-length line) 9) - (string=? (substring line 0 9) "processor")) - (loop (fx+ n 1))] - [else (loop n)])))))) - ) ;; end library --- a/lib/std/fiber.sls +++ b/lib/std/fiber.sls @@ -25,12 +25,52 @@ fiber-yield fiber-sleep fiber-self + fiber-id fiber? fiber-state fiber-name fiber-done? + ;; Cancellation + fiber-cancel! + fiber-cancelled? + fiber-check-cancelled! + &fiber-cancelled + make-fiber-cancelled + fiber-cancelled-condition? + cancelled-fiber-id + + ;; Fiber-local storage + make-fiber-parameter + fiber-parameterize + + ;; Join / error propagation + fiber-join + &fiber-timeout + make-fiber-timeout + fiber-timeout-condition? + timeout-fiber-id + + ;; Link (Erlang-style crash propagation) + fiber-link! + fiber-unlink! + &fiber-linked-crash + make-fiber-linked-crash + fiber-linked-crash? + linked-crash-source + linked-crash-condition + + ;; Channel select + fiber-select + + ;; Timeouts + fiber-timeout + + ;; Structured concurrency + with-fiber-group + fiber-group-spawn + make-fiber-channel fiber-channel? fiber-channel-send @@ -44,7 +84,24 @@ with-fibers) - (import (chezscheme)) + (import (chezscheme) (std misc cpu)) + + ;; ========================================================================= + ;; Condition types + ;; ========================================================================= + + (define-condition-type &fiber-cancelled &serious + make-fiber-cancelled fiber-cancelled-condition? + (fiber-id cancelled-fiber-id)) + + (define-condition-type &fiber-timeout &serious + make-fiber-timeout fiber-timeout-condition? + (fiber-id timeout-fiber-id)) + + (define-condition-type &fiber-linked-crash &serious + make-fiber-linked-crash fiber-linked-crash? + (source-fiber-id linked-crash-source) + (original-condition linked-crash-condition)) ;; ========================================================================= ;; Fiber record @@ -59,16 +116,76 @@ (mutable result) (mutable fiber-rt) ;; back-pointer to fiber-runtime (mutable gate) ;; #f or box: 'yield|'sleep|'channel -> 'done - (immutable mx)) ;; per-fiber mutex for park/wake coordination + (immutable mx) ;; per-fiber mutex for park/wake coordination + (mutable cancelled) ;; boolean — cooperative cancellation flag + (mutable join-waiters) ;; list of fibers waiting for this fiber to complete + (mutable linked-fibers) ;; list of fibers linked for crash propagation + (mutable pending-crash)) ;; #f or &fiber-linked-crash condition to deliver (protocol (lambda (new) (lambda (id thunk name rt) - (new id 'ready thunk name (void) rt #f (make-mutex)))))) + (new id 'ready thunk name (void) rt #f (make-mutex) + #f '() '() #f))))) (define (fiber-done? f) (eq? (fiber-state f) 'done)) ;; ========================================================================= + ;; Fiber-local storage + ;; ========================================================================= + + ;; Global registry of all live fiber-parameters for cleanup on fiber death. + (define *fiber-parameters* '()) + (define *fiber-parameters-mx* (make-mutex)) + + (define (register-fiber-parameter! fp) + (mutex-acquire *fiber-parameters-mx*) + (set! *fiber-parameters* (cons fp *fiber-parameters*)) + (mutex-release *fiber-parameters-mx*)) + + (define (cleanup-fiber-parameters! fid) + (mutex-acquire *fiber-parameters-mx*) + (for-each (lambda (fp) (fp fid #t)) *fiber-parameters*) + (mutex-release *fiber-parameters-mx*)) + + (define (make-fiber-parameter default) + (let ([store (make-eq-hashtable)] + [mx (make-mutex)]) + (define fp + (case-lambda + [() ;; read + (let ([f (current-fiber)]) + (if f + (begin (mutex-acquire mx) + (let ([v (hashtable-ref store (fiber-id f) default)]) + (mutex-release mx) v)) + default))] + [(val) ;; write + (let ([f (current-fiber)]) + (unless f (error 'fiber-parameter "not in a fiber")) + (mutex-acquire mx) + (hashtable-set! store (fiber-id f) val) + (mutex-release mx))] + [(fid cleanup?) ;; internal: cleanup by fiber id + (when cleanup? + (mutex-acquire mx) + (hashtable-delete! store fid) + (mutex-release mx))])) + (register-fiber-parameter! fp) + fp)) + + (define-syntax fiber-parameterize + (syntax-rules () + [(_ () body ...) + (begin body ...)] + [(_ ([fp val] rest ...) body ...) + (let ([old (fp)]) + (dynamic-wind + (lambda () (fp val)) + (lambda () (fiber-parameterize (rest ...) body ...)) + (lambda () (fp old))))])) + + ;; ========================================================================= ;; Run queue ;; ========================================================================= @@ -218,9 +335,6 @@ (make-mutex) (make-condition) (max 100 fuel))])))) - (define (cpu-count) - (if (threaded?) 4 1)) - (define (fiber-runtime-fiber-count rt) (- (fiber-runtime-total-fibers rt) (fiber-runtime-done-fibers rt))) @@ -282,9 +396,79 @@ (define (spin-until-gate gate) (let loop () (unless (eq? (unbox gate) 'done) (loop)))) + ;; ========================================================================= + ;; Cancellation + ;; ========================================================================= + + (define (fiber-cancelled? f) + (fiber-cancelled f)) + + (define fiber-cancel! + (case-lambda + [(f) + (let ([fmx (fiber-mx f)]) + (mutex-acquire fmx) + (unless (fiber-cancelled f) + (fiber-cancelled-set! f #t)) + (cond + [(eq? (fiber-state f) 'parked) + (let ([gate (fiber-gate f)]) + (when (and gate (box? gate)) + (set-box! gate 'done)) + (fiber-state-set! f 'ready) + (mutex-release fmx) + (rq-enqueue! (fiber-runtime-run-queue (fiber-fiber-rt f)) f))] + [else + ;; Running or ready — flag is set, will be checked at next + ;; cancellation point (yield/sleep/channel) + (mutex-release fmx)]))] + [(f timeout-ms) + ;; Cooperative cancel, then force-abandon after timeout + (fiber-cancel! f) + (fork-thread + (lambda () + (sleep (make-time 'time-duration + (* (fxmod timeout-ms 1000) 1000000) + (fxquotient timeout-ms 1000))) + (unless (fiber-done? f) + ;; Force: mark done without running further + (let ([fmx (fiber-mx f)] + [rt (fiber-fiber-rt f)] + [forced? #f]) + (mutex-acquire fmx) + (unless (eq? (fiber-state f) 'done) + (fiber-state-set! f 'done) + (fiber-result-set! f + (make-fiber-cancelled (fiber-id f))) + (fiber-continuation-set! f #f) + (fiber-gate-set! f #f) + (set! forced? #t)) + (mutex-release fmx) + (when forced? + (fiber-done-hooks! f rt))))))])) + + (define (fiber-check-cancelled!) + (let ([f (current-fiber)]) + (when (and f (fiber-cancelled f)) + (raise (make-fiber-cancelled (fiber-id f)))))) + + ;; Check cancellation + linked crash at cancellation points + (define (check-cancellation-point! f) + (when (fiber-cancelled f) + (raise (make-fiber-cancelled (fiber-id f)))) + (let ([crash (fiber-pending-crash f)]) + (when crash + (fiber-pending-crash-set! f #f) + (raise crash)))) + + ;; ========================================================================= + ;; Yield / Sleep + ;; ========================================================================= + (define (fiber-yield) (let ([f (current-fiber)]) (unless f (error 'fiber-yield "not running inside a fiber")) + (check-cancellation-point! f) (let ([gate (box 'yield)]) (fiber-gate-set! f gate) ;; Force immediate preemption @@ -298,6 +482,7 @@ [rt (current-fiber-runtime)]) (unless (and f rt) (error 'fiber-sleep "not running inside a fiber")) + (check-cancellation-point! f) (let ([gate (box 'sleep)]) (fiber-gate-set! f gate) ;; Register timer @@ -311,6 +496,7 @@ (set-timer 1) (spin-until-gate gate) (fiber-gate-set! f #f) + (check-cancellation-point! f) (void)))) ;; ========================================================================= @@ -331,21 +517,12 @@ ;; Core: run-fiber! ;; ========================================================================= - (define (mark-fiber-done! rt) - (mutex-acquire (fiber-runtime-done-mutex rt)) - (fiber-runtime-done-fibers-set! rt - (fx+ (fiber-runtime-done-fibers rt) 1)) - (when (fx= (fiber-runtime-done-fibers rt) - (fiber-runtime-total-fibers rt)) - (condition-broadcast (fiber-runtime-all-done rt))) - (mutex-release (fiber-runtime-done-mutex rt))) - (define (handle-expire rt f remaining result) (fiber-state-set! f 'done) (fiber-result-set! f result) (fiber-continuation-set! f #f) (fiber-gate-set! f #f) - (mark-fiber-done! rt)) + (fiber-done-hooks! f rt)) (define (handle-complete rt f new-engine) ;; Engine fuel exhausted — fiber was preempted. @@ -396,6 +573,51 @@ ;; Fiber still in engine — gate is set, will exit spin on resume (mutex-release fmx)]))) + ;; Post-completion hooks: wake join-waiters, propagate linked crashes, + ;; clean up fiber-parameters. Called after fiber state is 'done and + ;; result is set. Must be called OUTSIDE the fiber's mx. + (define (fiber-done-hooks! f rt) + ;; Wake join-waiters + (let ([waiters (fiber-join-waiters f)]) + (fiber-join-waiters-set! f '()) + (for-each (lambda (w) + (fiber-result-set! w (fiber-result f)) + (wake-fiber! w)) + waiters)) + ;; Crash propagation to linked fibers + (let ([result (fiber-result f)]) + (when (condition? result) + (let ([crash-cond (make-fiber-linked-crash (fiber-id f) result)]) + (for-each (lambda (linked) + (unless (fiber-done? linked) + (fiber-pending-crash-set! linked crash-cond) + ;; If parked, wake it so it can check the crash + (let ([fmx (fiber-mx linked)] + [need-enqueue? #f]) + (mutex-acquire fmx) + (when (eq? (fiber-state linked) 'parked) + (let ([gate (fiber-gate linked)]) + (when (and gate (box? gate)) + (set-box! gate 'done))) + (fiber-state-set! linked 'ready) + (set! need-enqueue? #t)) + (mutex-release fmx) + (when need-enqueue? + (rq-enqueue! (fiber-runtime-run-queue + (fiber-fiber-rt linked)) + linked))))) + (fiber-linked-fibers f))))) + ;; Fiber-parameter cleanup + (cleanup-fiber-parameters! (fiber-id f)) + ;; Decrement runtime counter + (mutex-acquire (fiber-runtime-done-mutex rt)) + (fiber-runtime-done-fibers-set! rt + (fx+ (fiber-runtime-done-fibers rt) 1)) + (when (fx= (fiber-runtime-done-fibers rt) + (fiber-runtime-total-fibers rt)) + (condition-broadcast (fiber-runtime-all-done rt))) + (mutex-release (fiber-runtime-done-mutex rt))) + (define (run-fiber! rt f fuel) (let ([cont (fiber-continuation f)]) (current-fiber-runtime rt) @@ -438,7 +660,7 @@ (fiber-result-set! f exn) (fiber-continuation-set! f #f) (fiber-gate-set! f #f) - (mark-fiber-done! rt)]) + (fiber-done-hooks! f rt)]) (run-fiber! rt f fuel)))) (loop))))) @@ -519,6 +741,8 @@ (fiber-channel-tail-set! ch count))) (define (fiber-channel-send ch val) + (let ([f (current-fiber)]) + (when f (check-cancellation-point! f))) (let ([mx (fiber-channel-mutex ch)]) (mutex-acquire mx) (when (fiber-channel-closed? ch) @@ -557,6 +781,8 @@ (fiber-gate-set! f #f))]))) (define (fiber-channel-recv ch) + (let ([f (current-fiber)]) + (when f (check-cancellation-point! f))) (let ([mx (fiber-channel-mutex ch)]) (mutex-acquire mx) (cond @@ -664,4 +890,349 @@ (for-each wake-fiber! waiters) (for-each (lambda (entry) (wake-fiber! (car entry))) senders)))) + ;; ========================================================================= + ;; Fiber join — block until fiber completes, return result or re-raise + ;; ========================================================================= + + (define fiber-join + (case-lambda + [(f) (fiber-join f #f)] + [(f timeout-ms) + (let ([caller (current-fiber)]) + (unless caller (error 'fiber-join "not running inside a fiber")) + (cond + [(fiber-done? f) + (let ([r (fiber-result f)]) + (if (condition? r) (raise r) r))] + [else + (let ([fmx (fiber-mx f)] + [gate (box 'channel)]) + (when timeout-ms + (let* ([rt (current-fiber-runtime)] + [now (current-time 'time-utc)] + [deadline (add-duration now + (make-time 'time-duration + (* (fxmod timeout-ms 1000) 1000000) + (fxquotient timeout-ms 1000)))]) + (tq-add! (fiber-runtime-timer-queue rt) deadline caller))) + (mutex-acquire fmx) + (cond + [(eq? (fiber-state f) 'done) + (mutex-release fmx) + (let ([r (fiber-result f)]) + (if (condition? r) (raise r) r))] + [else + (fiber-join-waiters-set! f + (cons caller (fiber-join-waiters f))) + (fiber-gate-set! caller gate) + (fiber-result-set! caller (void)) + (mutex-release fmx) + (set-timer 1) + (spin-until-gate gate) + (fiber-gate-set! caller #f) + (let ([r (fiber-result caller)]) + (cond + [(and timeout-ms (eq? r (void)) (not (fiber-done? f))) + (mutex-acquire fmx) + (fiber-join-waiters-set! f + (remq caller (fiber-join-waiters f))) + (mutex-release fmx) + (raise (make-fiber-timeout (fiber-id f)))] + [(condition? r) (raise r)] + [else r]))]))]))])) + + ;; ========================================================================= + ;; Fiber link — Erlang-style crash propagation + ;; ========================================================================= + + (define (fiber-link! f) + (let ([caller (current-fiber)]) + (unless caller (error 'fiber-link! "not running inside a fiber")) + (let ([fmx (fiber-mx f)]) + (mutex-acquire fmx) + (cond + ;; Already done with error — deliver crash immediately + [(and (fiber-done? f) (condition? (fiber-result f))) + (mutex-release fmx) + (raise (make-fiber-linked-crash (fiber-id f) (fiber-result f)))] + [else + (fiber-linked-fibers-set! f + (cons caller (fiber-linked-fibers f))) + (mutex-release fmx)])))) + + (define (fiber-unlink! f) + (let ([caller (current-fiber)]) + (when caller + (let ([fmx (fiber-mx f)]) + (mutex-acquire fmx) + (fiber-linked-fibers-set! f + (remq caller (fiber-linked-fibers f))) + (mutex-release fmx))))) + + ;; ========================================================================= + ;; Fiber-channel select — auxiliary syntax and macros + ;; ========================================================================= + + ;; ========================================================================= + ;; Fiber-channel select — wait on multiple channels + ;; ========================================================================= + ;; + ;; (fiber-select clause ...) + ;; Clause forms: + ;; [ch val => body ...] — recv from ch, bind val + ;; [ch :send expr => body ...] — send expr to ch + ;; [:timeout ms => body ...] — timeout in milliseconds + ;; [:default => body ...] — non-blocking fallback + + (define-syntax fiber-select + (lambda (x) + (define (keyword? id sym) + (and (identifier? id) + (eq? (syntax->datum id) sym))) + (define (parse-clauses clauses) + ;; Returns (recvs sends timeout default) as syntax objects + (let loop ([cls clauses] [recvs '()] [sends '()] [tout #f] [dflt #f]) + (if (null? cls) + (values (reverse recvs) (reverse sends) tout dflt) + (syntax-case (car cls) (=>) + [(kw => body ...) + (keyword? #'kw ':default) + (loop (cdr cls) recvs sends tout #'(body ...))] + [(kw ms => body ...) + (keyword? #'kw ':timeout) + (loop (cdr cls) recvs sends #'(ms body ...) dflt)] + [(ch kw expr => body ...) + (keyword? #'kw ':send) + (loop (cdr cls) recvs (cons #'(ch expr body ...) sends) tout dflt)] + [(ch val => body ...) + (loop (cdr cls) (cons #'(ch val body ...) recvs) sends tout dflt)])))) + (syntax-case x () + [(_ clause ...) + (let-values ([(recvs sends tout dflt) (parse-clauses #'(clause ...))]) + (with-syntax + ([recv-specs + (if (null? recvs) #''() + (let loop ([rs recvs]) + (if (null? rs) #''() + (syntax-case (car rs) () + [(ch val body ...) + #`(cons (cons ch (lambda (val) body ...)) + #,(loop (cdr rs)))]))))] + [send-specs + (if (null? sends) #''() + (let loop ([ss sends]) + (if (null? ss) #''() + (syntax-case (car ss) () + [(ch expr body ...) + #`(cons (cons* ch expr (lambda () body ...)) + #,(loop (cdr ss)))]))))] + [timeout-spec + (if tout + (syntax-case tout () + [(ms body ...) + #'(cons ms (lambda () body ...))]) + #'#f)] + [default-spec + (if dflt + (syntax-case dflt () + [(body ...) + #'(lambda () body ...)]) + #'#f)]) + #'(fiber-select-impl recv-specs send-specs timeout-spec default-spec)))]))) + + + ;; Runtime implementation of fiber-select + (define (fiber-select-impl recv-specs send-specs timeout-spec default-spec) + (define (try-recv specs) + (cond + [(null? specs) #f] + [else + (let* ([spec (car specs)] + [ch (car spec)] + [handler (cdr spec)]) + (let-values ([(val ok) (fiber-channel-try-recv ch)]) + (if ok + (cons 'ok (handler val)) + (try-recv (cdr specs)))))])) + (define (try-send specs) + (cond + [(null? specs) #f] + [else + (let* ([spec (car specs)] + [ch (car spec)] + [val (cadr spec)] + [handler (cddr spec)]) + (if (fiber-channel-try-send ch val) + (cons 'ok (handler)) + (try-send (cdr specs))))])) + (define (block-on-channels f) + (let ([gate (box 'channel)]) + (fiber-gate-set! f gate) + (fiber-result-set! f (void)) + ;; Register timeout if present + (when timeout-spec + (let* ([rt (current-fiber-runtime)] + [ms (car timeout-spec)] + [now (current-time 'time-utc)] + [deadline (add-duration now + (make-time 'time-duration + (* (fxmod ms 1000) 1000000) + (fxquotient ms 1000)))]) + (tq-add! (fiber-runtime-timer-queue rt) deadline f))) + ;; Register on recv channels + (for-each (lambda (spec) + (let ([ch (car spec)] + [mx (fiber-channel-mutex (car spec))]) + (mutex-acquire mx) + (fiber-channel-recv-waiters-set! ch + (append (fiber-channel-recv-waiters ch) (list f))) + (mutex-release mx))) + recv-specs) + ;; Park the fiber + (set-timer 1) + (spin-until-gate gate) + (fiber-gate-set! f #f) + ;; Remove from all recv waiter lists + (for-each (lambda (spec) + (let ([ch (car spec)] + [mx (fiber-channel-mutex (car spec))]) + (mutex-acquire mx) + (fiber-channel-recv-waiters-set! ch + (remq f (fiber-channel-recv-waiters ch))) + (mutex-release mx))) + recv-specs) + (check-cancellation-point! f) + ;; Determine what woke us + (let ([result (fiber-result f)]) + (cond + [(not (eq? result (void))) + (if (pair? recv-specs) + ((cdr (car recv-specs)) result) + result)] + [timeout-spec + ((cdr timeout-spec))] + [else (select-loop)])))) + (define (select-loop) + (let ([hit (try-recv recv-specs)]) + (cond + [hit (cdr hit)] + [else + (let ([hit2 (try-send send-specs)]) + (cond + [hit2 (cdr hit2)] + [default-spec (default-spec)] + [else + (let ([f (current-fiber)]) + (unless f (error 'fiber-select "blocking select requires fiber context")) + (block-on-channels f))]))]))) + ;; Entry point + (let ([f (current-fiber)]) + (when f (check-cancellation-point! f))) + (select-loop)) + + ;; ========================================================================= + ;; Fiber timeout — channel that fires after N milliseconds + ;; ========================================================================= + + (define (fiber-timeout ms) + (let ([ch (make-fiber-channel 1)]) + (fiber-spawn* (lambda () + (fiber-sleep ms) + (fiber-channel-try-send ch (void)))) + ch)) + + ;; ========================================================================= + ;; Structured concurrency — with-fiber-group + ;; ========================================================================= + + (define-record-type fiber-group + (fields + (mutable fibers) ;; list of child fibers + (mutable first-exn) ;; first exception, or #f + (mutable group-cancelled?) ;; has group been cancelled? + (immutable group-mutex) + (immutable all-done-cv) ;; condition variable + (mutable done-count) + (mutable total-count)) + (protocol + (lambda (new) + (lambda () + (new '() #f #f (make-mutex) (make-condition) 0 0))))) + + (define (fiber-group-spawn group thunk) + (let ([rt (current-fiber-runtime)]) + (unless rt (error 'fiber-group-spawn "no active fiber runtime")) + (let ([f (fiber-spawn rt + (lambda () + (guard (exn [#t + ;; Record first exception and cancel siblings + (let ([gmx (fiber-group-group-mutex group)]) + (mutex-acquire gmx) + (unless (fiber-group-first-exn group) + (fiber-group-first-exn-set! group exn)) + (mutex-release gmx)) + ;; Cancel all siblings + (for-each (lambda (sib) + (unless (fiber-done? sib) + (fiber-cancel! sib))) + (fiber-group-fibers group)) + (raise exn)]) + (thunk))))]) + ;; Track child in group + (let ([gmx (fiber-group-group-mutex group)]) + (mutex-acquire gmx) + (fiber-group-fibers-set! group + (cons f (fiber-group-fibers group))) + (fiber-group-total-count-set! group + (fx+ (fiber-group-total-count group) 1)) + (mutex-release gmx)) + f))) + + (define (%fiber-group-wait group) + ;; Wait for all children in the group to complete. + ;; Uses fiber-join to block on each non-done child. + (let wait () + (let ([pending #f]) + (let ([gmx (fiber-group-group-mutex group)]) + (mutex-acquire gmx) + (let find ([fibs (fiber-group-fibers group)]) + (cond