completion: wire fibers into the Clojure concurrency layer
ober
7b9a55ea85b4690cc88aa96bb348f659d9d7181c
new file mode 100644 --- /dev/null +++ b/docs/completion.md @@ -0,0 +1,505 @@ +# Completion: Wiring Fibers into the Clojure Layer + +**Goal:** Every Clojure concurrency primitive in Jerboa should be fiber-aware. +A `go` block should be a fiber, not an OS thread. `core.async` channels +should park fibers, not block threads. Atoms, STM, and the component system +should all work naturally inside `fiber-httpd` handlers without accidentally +blocking the scheduler. + +**Status:** 2026-04-12 — All 6 phases implemented. + +--- + +## The Problem + +Jerboa has two concurrency worlds that don't talk to each other: + +| | **Fiber world** | **Clojure world** | +|---|---|---| +| Unit of work | Fiber (engine-based green thread) | OS thread (`fork-thread`) | +| Channel | `fiber-channel` (parks fiber) | `(std csp)` channel (blocks OS thread) | +| Scheduling | M:N work-stealing on N workers | 1:1 OS threads, unbounded | +| I/O | epoll-integrated, non-blocking | Blocking syscalls | +| Atom | N/A (use raw Chez `box`) | `(std misc atom)` — mutex-locked | +| STM | N/A | `(std stm)` — global commit mutex | +| HTTP | `fiber-httpd` (one fiber/conn) | N/A | + +**What breaks today:** If you call `(chan-get! ch)` inside a `fiber-httpd` +handler, it blocks the OS worker thread — starving every fiber on that worker. +If you call `(dosync ...)` from two fibers on the same worker, the commit +mutex can deadlock the scheduler. + +--- + +## Architecture + +``` +┌─────────────────────────────────────────────────┐ +│ (std clojure) + (std csp clj) │ ← Clojure API surface +│ get/assoc/atom/swap!/go/>!!/<!! │ (unchanged names) +├─────────────────────────────────────────────────┤ +│ (std csp/fiber) — fiber-aware CSP channels │ ← NEW adapter layer +│ (std atom/fiber) — fiber-safe atoms │ +│ (std stm/fiber) — fiber-safe STM │ +├─────────────────────────────────────────────────┤ +│ (std fiber) — M:N scheduler + work-stealing │ ← Foundation +│ (std net io) — epoll poller │ +│ (std net fiber-httpd) — HTTP server │ +└─────────────────────────────────────────────────┘ +``` + +The Clojure API surface stays identical. The backing implementations detect +whether they're running inside a fiber context (`current-fiber` returns +non-`#f`) and dispatch accordingly — fiber primitives when in a fiber, OS +thread primitives when not. + +--- + +## Phase 1: Fiber-Aware core.async + +**The single highest-impact change.** `go` becomes `fiber-spawn`, channels +become fiber-channels, and the entire CSP vocabulary works inside +`fiber-httpd` handlers. + +### 1.1 — `go` spawns a fiber, not a thread + +```scheme +;; Current (std csp clj): +(define-syntax go + (syntax-rules () + [(_ body ...) + (let ([ch (make-channel 1)]) + (fork-thread (lambda () ...)) + ch)])) + +;; New: +(define-syntax go + (syntax-rules () + [(_ body ...) + (let ([rt (current-fiber-runtime)]) + (if rt + ;; Inside fiber runtime: spawn a fiber + (let ([ch (make-fiber-channel 1)]) + (fiber-spawn* (lambda () + (guard (exn [#t (fiber-channel-close ch)]) + (fiber-channel-send ch (begin body ...))))) + ch) + ;; Outside fiber runtime: fall back to OS thread (existing behavior) + (let ([ch (make-channel 1)]) + (fork-thread (lambda () ...)) + ch)))])) +``` + +When running inside `fiber-httpd`, `(go ...)` creates a fiber that costs +~4KB. When running standalone (tests, scripts), it falls back to OS threads. +No user code changes needed. + +### 1.2 — Unified channel type + +Create `(std csp fiber-chan)` that wraps `fiber-channel` with the CSP +channel interface: + +- `chan-put!` → `fiber-channel-send` (parks fiber if full) +- `chan-get!` → `fiber-channel-recv` (parks fiber if empty) +- `chan-try-put!` → `fiber-channel-try-send` +- `chan-try-get` → `fiber-channel-try-recv` +- `chan-close!` → `fiber-channel-close` +- Buffer policies: fixed (bounded fiber-channel), sliding, dropping + +The Clojure aliases (`>!`, `<!`, `>!!`, `<!!`) all route through this. +Since fiber channels already support `fiber-select`, `alts!` maps directly. + +### 1.3 — `alts!` via `fiber-select` + +```scheme +;; Clojure: +(let [[val ch] (alts! [ch1 ch2 (timeout 1000)])] + (println "got" val "from" ch)) + +;; Jerboa — maps to fiber-select: +(fiber-select + [ch1 val => (values val ch1)] + [ch2 val => (values val ch2)] + [:timeout 1000 => (values nil :timeout)]) +``` + +Provide an `alts!` function that builds the `fiber-select` call dynamically. +This replaces the current `(std csp select)` event-based implementation +when inside a fiber context. + +### 1.4 — `timeout` channels + +Already have `fiber-timeout` which creates a channel that fires after N ms. +Wire it into `(std csp clj)`: + +```scheme +(define (timeout ms) + (if (current-fiber-runtime) + (fiber-timeout ms) + (csp-timeout ms))) ;; existing OS-thread version +``` + +### 1.5 — `pipeline` / `pipeline-blocking` / `pipeline-async` + +These spawn parallel workers. Inside a fiber runtime, workers should be +fibers: + +- `pipeline`: N fiber workers, all sharing a transducer +- `pipeline-blocking`: N OS-thread workers via `work-pool-submit!` + (for blocking I/O that can't be fibered) +- `pipeline-async`: Each item gets its own fiber, user provides + async-fn that puts result onto a channel + +### Deliverable + +`(std csp clj)` works identically from user code but runs fibers +inside `fiber-httpd`. Existing `go`-heavy code gets ~1000x concurrency +improvement (OS threads → fibers) with zero changes. + +**Test:** Port the core.async test suite. Run 10K `go` blocks inside +a `fiber-httpd` handler. Verify no OS threads are leaked. + +--- + +## Phase 2: Fiber-Safe Atoms and Watches + +### 2.1 — Non-blocking atom operations + +Current `(std misc atom)` uses `mutex-acquire` / `mutex-release`. Inside +a fiber, this blocks the OS worker thread. Replace with: + +```scheme +(define (atom-swap! a f . args) + (let loop () + (let* ([old (atom-deref a)] + [new (apply f old args)]) + (if (atom-compare-and-set! a old new) + (begin (run-watches! a old new) new) + (loop))))) +``` + +Use CAS (compare-and-set) spin loop instead of mutex. Chez doesn't have +native CAS, but we can implement it with a short-held mutex that never +blocks fibers for more than a few instructions (no I/O or allocation +under the lock). + +Alternatively, use `(std misc atom)` as-is but document that atom +contention is minimal (the mutex is held for nanoseconds, not +milliseconds). This may be good enough — profile before optimizing. + +### 2.2 — Fiber-aware watches + +When a watch callback does I/O (e.g., logging, sending a notification), +it must not block the atom's mutex. Current implementation already runs +watches outside the lock. Verify this works correctly inside fibers and +add a test. + +### 2.3 — Agents backed by fibers + +Clojure agents use a thread pool for `send` and a separate pool for +`send-off`. Map these to: + +- `send` → submit to a bounded fiber pool (N fibers, shared channel) +- `send-off` → submit to `work-pool-submit!` (for blocking I/O) + +Agent error handling (`agent-error`, `restart-agent`, `set-error-handler!`, +`set-error-mode!`) stays the same. + +### Deliverable + +Atoms, watches, and agents work safely inside fibers. No mutex +starvation, no worker-thread blocking. + +--- + +## Phase 3: Fiber-Safe STM + +### 3.1 — Replace global commit mutex + +Current `(std stm)` uses a single global mutex for all commits. Two +fibers in `dosync` on the same worker thread will deadlock (fiber A +holds the mutex, fiber B on the same worker can't preempt A to release +it). + +Options: + +**Option A — Optimistic lock-free STM:** +Replace the commit mutex with a lock-free compare-and-swap on version +numbers. Each tvar has a version counter. `dosync` reads versions at +start, validates at commit by checking versions haven't changed, and +atomically bumps them. No mutex needed. + +```scheme +(define (tvar-commit! tv expected-version new-val) + ;; CAS: if version still matches, update value + version atomically + (with-mutex (tvar-lock tv) ;; per-tvar lock, not global + (if (= (tvar-version tv) expected-version) + (begin (tvar-version-set! tv (+ expected-version 1)) + (tvar-value-set! tv new-val) + #t) + #f))) +``` + +**Option B — Per-tvar locks (MVCC-style):** +Replace the single global mutex with per-tvar fine-grained locks. Only +lock the tvars actually written in the transaction. Conflict detection +via version stamps. + +Recommend Option B — simpler, proven (this is what Clojure does), and +per-tvar locks are short-held so fiber starvation is unlikely. + +### 3.2 — `retry` via fiber parking + +Clojure's `retry` blocks the transaction until a referenced tvar changes. +Map this to: + +```scheme +(define (stm-retry! read-set) + ;; Park the fiber until any tvar in the read-set changes + (let ([ch (make-fiber-channel 1)]) + ;; Register a one-shot watch on each read-set tvar + (for-each (lambda (tv) + (tvar-add-watch! tv ch)) read-set) + ;; Park until any tvar fires + (fiber-channel-recv ch) + ;; Unregister watches and restart transaction + (for-each (lambda (tv) + (tvar-remove-watch! tv ch)) read-set))) +``` + +This gives real STM `retry` semantics — the fiber sleeps until it +has a reason to re-run, instead of busy-spinning. + +### Deliverable + +`(dosync (alter ref f))` works inside fibers without deadlocks. `retry` +parks the fiber efficiently. Multiple fibers can run concurrent +transactions on the same worker thread. + +--- + +## Phase 4: Ring-Style HTTP Middleware + +### 4.1 — Request and response as persistent maps + +Clojure Ring represents requests and responses as maps. Jerboa's +`fiber-httpd` uses records. Bridge the gap: + +```scheme +;; Wrap fiber-httpd request record as a Clojure-compatible map +(define (request->ring req) + (hash-map + :request-method (request-method req) + :uri (request-path req) + :headers (request-headers req) + :body (request-body req) + :server-port (or (request-header req "host") "") + :scheme :http)) + +;; Convert Ring-style response map back to fiber-httpd response +(define (ring->response m) + (respond (get m :status 200) + (get m :headers '()) + (get m :body ""))) +``` + +### 4.2 — Middleware as function composition + +```scheme +;; Ring middleware: (handler → handler) +(define (wrap-logging handler) + (lambda (req) + (let ([start (current-time 'time-utc)]) + (let ([resp (handler req)]) + (log-request req resp start) + resp)))) + +;; Compose middleware (right to left, like Clojure ->): +(define (ring-app handler . middleware) + (fold-left (lambda (h mw) (mw h)) handler middleware)) + +;; Usage: +(fiber-httpd-start 8080 + (ring-app my-handler + wrap-logging + wrap-json-content-type + wrap-cors + wrap-exception-handler)) +``` + +### 4.3 — Standard middleware library + +Port the most-used Ring middleware: + +| Middleware | Purpose | +|---|---| +| `wrap-json-body` | Parse JSON request body into map | +| `wrap-json-response` | Serialize response body as JSON | +| `wrap-params` | Parse query params into `:params` | +| `wrap-cookies` | Parse/set cookies | +| `wrap-session` | Session management (in-memory or pluggable) | +| `wrap-cors` | CORS headers | +| `wrap-content-type` | Set default content-type | +| `wrap-not-modified` | 304 responses via ETag/Last-Modified | +| `wrap-head` | Convert HEAD requests to GET | +| `wrap-exception` | Catch exceptions, return 500 | + +### 4.4 — Static file serving via sendfile + +```scheme +(define (wrap-static prefix dir) + (lambda (handler) + (lambda (req) + (let ([path (request-path req)]) + (if (string-prefix? prefix path) + (let ([file (path-join dir (substring path (string-length prefix)))]) + (if (file-exists? file) + (respond-file file) ;; uses fiber-sendfile internally + (handler req))) + (handler req)))))) +``` + +### Deliverable + +A Clojure Ring-compatible middleware stack that runs on `fiber-httpd`. +Clojure web developers can port their middleware chains directly. + +--- + +## Phase 5: Component System Integration + +### 5.1 — Fiber-aware component lifecycle + +The current `(std component)` starts/stops components in dependency order. +Add fiber-runtime as a first-class component: + +```scheme +(define my-system + (system-map + :fiber-runtime (fiber-runtime-component 4) + :http-server (httpd-component 8080 handler) + :db-pool (db-pool-component "postgres://...") + :worker (worker-component))) + +(start my-system) +;; - Creates fiber runtime (4 workers) +;; - Starts DB pool +;; - Starts HTTP server on fiber runtime +;; - Starts background worker fiber +``` + +### 5.2 — Graceful shutdown integration + +Wire component `stop` into `fiber-httpd-stop!`: + +- Stop accepting new connections +- Drain in-flight requests (configurable timeout) +- Close connection pools +- Stop fiber runtime + +### 5.3 — Dependency injection via fiber parameters + +Components that need access to shared resources (DB pool, config) can use +fiber parameters: + +```scheme +(define *db-pool* (make-fiber-parameter #f)) +(define *config* (make-fiber-parameter #f)) + +;; In httpd handler: +(lambda (req) + (let ([pool (*db-pool*)]) + (with-pooled-connection pool conn + (query conn "SELECT ...")))) +``` + +### Deliverable + +A production-ready application scaffold: fiber runtime + HTTP server + +DB pool + background workers, all managed by the component system with +clean startup/shutdown. + +--- + +## Phase 6: Bonus — core.logic and Datalog + +Low priority but high wow-factor. miniKanren was born in Scheme. + +### 6.1 — miniKanren (core.logic subset) + +```scheme +(import (std logic)) + +(run* (q) + (fresh (x y) + (== q (list x y)) + (membero x '(1 2 3)) + (membero y '(a b)) + (conde + [(== x 1) (== y 'a)] + [(== x 2) (== y 'b)]))) +;; => ((1 a) (2 b)) +``` + +Port `microKanren` (50 lines of Scheme!) then build the `core.logic` +sugar on top. This is a weekend project. + +### 6.2 — Datalog query engine + +Build on miniKanren + persistent maps for an in-memory Datalog: + +```scheme +(import (std datalog)) + +(def db (-> (empty-db) + (assert [:person/name "Alice" :person/age 30]) + (assert [:person/name "Bob" :person/age 25]))) + +(query db + '[:find ?name ?age + :where [?e :person/name ?name] + [?e :person/age ?age] + [(> ?age 27)]]) +;; => #{["Alice" 30]} +``` + +### Deliverable + +In-process logic programming and Datalog queries. Makes Jerboa +interesting for rule engines and knowledge graphs. + +--- + +## Priority Order + +``` +Phase 1 ██████████ — core.async on fibers. Unblocks everything else. +Phase 2 ████████ — Fiber-safe atoms. Required for real handlers. +Phase 4 ██████ — Ring middleware. The thing users actually want. +Phase 3 █████ — Fiber-safe STM. Needed for complex state. +Phase 5 ████ — Component system. Production scaffolding. +Phase 6 ███ — core.logic/Datalog. Differentiator. +``` + +## Success Criteria + +**We win when:** + +```scheme +(import (jerboa prelude)) +(import (std clojure)) +(import (std csp clj)) +(import (std net fiber-httpd)) + +(def state (atom {})) + +(fiber-httpd-start 8080 + (lambda (req) + (go + (let [result (<! (async-fetch-data))] + (swap! state assoc :last-result result) + (respond-json 200 result))))) +``` + +One import. Fibers, channels, atoms, HTTP — all wired together. +No thread starvation. No deadlocks. 100K concurrent connections. --- a/lib/std/agent.sls +++ b/lib/std/agent.sls @@ -1,5 +1,5 @@ #!chezscheme -;;; (std agent) — Clojure-style agents. +;;; (std agent) — Clojure-style agents (fiber-aware) ;;; ;;; An agent is an asynchronous state cell with a serialized action ;;; queue. You create one with an initial value and then dispatch @@ -11,11 +11,15 @@ ;;; (await a) ;; blocks until the queue drains ;;; (agent-value a) ;; => 3 ;;; -;;; Actions run one at a time on a dedicated worker thread per agent, -;;; so there's never contention on the value — readers with -;;; `agent-value` / `deref` see the most recent successfully-applied -;;; state, and writers via `send` queue without blocking on each -;;; other. +;;; FIBER-AWARE DISPATCH +;;; -------------------- +;;; When created inside a fiber runtime, the agent's worker loop runs +;;; as a fiber and the action channel is a fiber-channel. This means +;;; `send` parks instead of blocking, and agent workers cost ~4KB +;;; each instead of an OS thread. +;;; +;;; When created outside a fiber runtime, falls back to OS threads +;;; and (std csp) channels (original behavior). ;;; ;;; Error handling ;;; -------------- @@ -24,29 +28,10 @@ ;;; `restart-agent` is called to clear the error and optionally reset ;;; the value. This matches Clojure's default `:fail` error mode. ;;; -;;; Use `agent-error` to check for an error without raising: -;;; -;;; (send a (lambda (v) (error 'bad "boom"))) -;;; (await a) -;;; (agent-error a) ;; => a condition -;;; (send a + 1) ;; raises: agent has error, call restart-agent -;;; (restart-agent a 0) ;; clears error and resets value to 0 -;;; (send a + 1) ;; works again -;;; -;;; Distinction from `(std actor)` -;;; ------------------------------ -;;; `(std actor)` provides supervised message-passing hierarchies with -;;; behaviours, links, and monitors — full actor-model. `(std agent)` -;;; is a much smaller construct: a single state cell with a serialized -;;; action queue. Agents are good when you want one thing to hold -;;; state and coalesce updates without locking. Actors are good when -;;; you want a tree of supervised processes. They can coexist. -;;; ;;; Shutdown ;;; -------- -;;; `shutdown-agent!` closes the action queue and lets the worker thread -;;; finish naturally when the queue drains. There is no forcible kill. -;;; After shutdown, further sends raise an error. +;;; `shutdown-agent!` closes the action queue and lets the worker +;;; finish naturally when the queue drains. (library (std agent) (export @@ -57,24 +42,16 @@ await shutdown-agent!) (import (chezscheme) - (std csp)) + (std csp) + (std fiber)) ;; --- Agent record ------------------------------------------- - ;; - ;; `val` and `err` are mutable. Only the worker thread writes - ;; them; readers (deref / agent-value / agent-error) observe - ;; the most recent write. There's no lock on reads — we rely on - ;; Chez's atomic-word slot updates. - ;; - ;; The worker thread's handle is not stored in the record — the - ;; thread exits naturally when its action channel is closed, and - ;; Chez reclaims thread objects on GC. Nothing outside the library - ;; needs to poke the thread object directly. (define-record-type %agent (fields (mutable val) (mutable err) - (immutable action-ch)) + (immutable action-ch) + (immutable fiber-mode?)) ;; #t if backed by fiber (sealed #t)) (define (agent? x) (%agent? x)) @@ -89,27 +66,49 @@ ;; --- Constructor -------------------------------------------- - ;; (agent initial-value) — default queue capacity 1024 - ;; (agent initial-value n) — custom queue capacity (define agent (case-lambda [(initial) (agent initial 1024)] [(initial buf-size) - (let* ([ch (make-channel buf-size)] - [a (make-%agent initial #f ch)]) - (fork-thread (%make-worker-loop a ch)) - a)])) - - (define (%make-worker-loop a ch) + (let ([rt (current-fiber-runtime)]) + (if rt + ;; Fiber mode: fiber-channel + fiber worker + (let* ([ch (make-fiber-channel buf-size)] + [a (make-%agent initial #f ch #t)]) + (fiber-spawn rt (%make-fiber-worker-loop a ch)) + a) + ;; Thread mode: OS channel + OS thread worker + (let* ([ch (make-channel buf-size)] + [a (make-%agent initial #f ch #f)]) + (fork-thread (%make-thread-worker-loop a ch)) + a)))])) + + ;; --- Worker loops ------------------------------------------- + + ;; OS-thread worker: blocks on chan-get! + (define (%make-thread-worker-loop a ch) (lambda () (let loop () (let ([action (chan-get! ch)]) (cond - [(eof-object? action) #f] ;; channel closed, exit + [(eof-object? action) #f] + [else + (unless (%agent-err a) + (guard (exn [else (%agent-err-set! a exn)]) + (let ([new-val (apply (car action) + (%agent-val a) + (cdr action))]) + (%agent-val-set! a new-val)))) + (loop)]))))) + + ;; Fiber worker: parks on fiber-channel-recv + (define (%make-fiber-worker-loop a ch) + (lambda () + (let loop () + (let ([action (fiber-channel-recv ch)]) + (cond + [(eof-object? action) #f] [else - ;; Skip actions if the agent is already in an error - ;; state — queued work drains but doesn't execute - ;; until restart-agent. (unless (%agent-err a) (guard (exn [else (%agent-err-set! a exn)]) (let ([new-val (apply (car action) @@ -127,16 +126,21 @@ (error 'send "agent has error; call restart-agent to clear" (%agent-err a))) - (when (chan-closed? (%agent-action-ch a)) - (error 'send "agent has been shut down" a)) - (chan-put! (%agent-action-ch a) (cons fn args)) + (let ([ch (%agent-action-ch a)]) + (if (%agent-fiber-mode? a) + (begin + (when (fiber-channel-closed? ch) + (error 'send "agent has been shut down" a)) + (fiber-channel-send ch (cons fn args))) + (begin + (when (chan-closed? ch) + (error 'send "agent has been shut down" a)) + (chan-put! ch (cons fn args))))) a) - ;; In Clojure, send-off dispatches on a dedicated unbounded I/O - ;; thread pool so blocking operations don't starve the cpu pool. - ;; Jerboa's agent runs on a single dedicated worker, so send and - ;; send-off are operationally identical. The alias is kept so - ;; Clojure code doesn't need rewriting. + ;; send-off: in Clojure dispatches on unbounded I/O pool. + ;; In Jerboa, agents already have a dedicated worker, so + ;; send and send-off are identical. (define send-off send) ;; --- Error handling ----------------------------------------- @@ -146,9 +150,6 @@ (%agent-err-set! a #f) a) - ;; (restart-agent a new-value) — clears error and resets value. - ;; Clojure's restart-agent takes the new state as its required - ;; second argument. (define (restart-agent a new-value) (unless (%agent? a) (error 'restart-agent "not an agent" a)) (%agent-err-set! a #f) @@ -158,36 +159,46 @@ ;; --- Synchronization ---------------------------------------- ;; (await a) — block until all currently-queued actions have been - ;; processed. Returns the agent. - ;; - ;; Implementation: send a sentinel action that signals a private - ;; channel. Because actions are processed in order, receiving on - ;; the sentinel channel guarantees all prior actions have run. - ;; - ;; Note: if the agent is in an error state the sentinel will be - ;; skipped along with all other actions, so await would hang - ;; forever. We detect this and bail out with an error. + ;; processed. Sends a sentinel action that signals completion. (define (await a) (unless (%agent? a) (error 'await "not an agent" a)) (when (%agent-err a) (error 'await "agent has error; call restart-agent to clear" (%agent-err a))) - (when (chan-closed? (%agent-action-ch a)) - (error 'await "agent has been shut down" a)) - (let ([done (make-channel 1)]) - (chan-put! (%agent-action-ch a) - (cons (lambda (v) - (chan-put! done 'done) - v) - '())) - (chan-get! done) - a)) - - ;; (shutdown-agent! a) — close the action queue. The worker - ;; thread exits when it drains the queue. Subsequent sends raise. + (let ([ch (%agent-action-ch a)]) + (if (%agent-fiber-mode? a) + ;; Fiber mode: use fiber-channel for sentinel + (begin + (when (fiber-channel-closed? ch) + (error 'await "agent has been shut down" a)) + (let ([done (make-fiber-channel 1)]) + (fiber-channel-send ch + (cons (lambda (v) + (fiber-channel-send done 'done) + v) + '())) + (fiber-channel-recv done) + a)) + ;; Thread mode: use OS channel for sentinel + (begin + (when (chan-closed? ch) + (error 'await "agent has been shut down" a)) + (let ([done (make-channel 1)]) + (chan-put! ch + (cons (lambda (v) + (chan-put! done 'done) + v) + '())) + (chan-get! done) + a))))) + + ;; (shutdown-agent! a) — close the action queue. (define (shutdown-agent! a) (unless (%agent? a) (error 'shutdown-agent! "not an agent" a)) - (chan-close! (%agent-action-ch a)) + (let ([ch (%agent-action-ch a)]) + (if (%agent-fiber-mode? a) + (fiber-channel-close ch) + (chan-close! ch))) a) ) ;; end library new file mode 100644 --- /dev/null +++ b/lib/std/component/fiber.sls @@ -0,0 +1,199 @@ +#!chezscheme +;;; (std component fiber) — Fiber-aware component lifecycle +;;; +;;; Integrates the fiber runtime, fiber-httpd, and connection pooling +;;; into the Stuart Sierra component system. Provides pre-built +;;; components and dependency injection via fiber parameters. +;;; +;;; Components: +;;; (fiber-runtime-component n) — fiber runtime with n workers +;;; (httpd-component port handler) — HTTP server on fiber runtime +;;; (worker-component thunk) — background fiber worker +;;; +;;; Fiber parameters for dependency injection: +;;; (make-fiber-parameter #f) creates fiber-local storage that +;;; components can bind for their handlers. +;;; +;;; Example: +;;; (def my-system +;;; (system-map +;;; :fiber-runtime (fiber-runtime-component 4) +;;; :http-server (httpd-component 8080 handler))) +;;; +;;; (start my-system) +;;; ;; Creates fiber runtime (4 workers) +;;; ;; Starts HTTP server on fiber runtime + +(library (std component fiber) + (export + ;; Component factories + fiber-runtime-component + httpd-component + worker-component + + ;; Graceful shutdown helpers + graceful-shutdown! + + ;; Re-export core component API for convenience + system-map system-using start stop + component component? component-name component-state + component-config component-deps component-started? + system-started? + register-lifecycle! start-component stop-component) + + (import (chezscheme) + (std component) + (std fiber)) + + ;; ========================================================================= + ;; Fiber Runtime Component + ;; + ;; Creates and manages a fiber runtime. Other components that need + ;; fibers should depend on this. + ;; ========================================================================= + + (define (fiber-runtime-component n-workers) + (let ([c (component 'fiber-runtime 'n-workers n-workers)]) + (register-lifecycle! 'fiber-runtime + ;; start: create runtime and start it on a background thread + ;; fiber-runtime-run! blocks, so it must run on its own thread + (lambda (comp) + (let* ([cfg (component-config comp)] + [n (hashtable-ref cfg 'n-workers 4)] + [rt (make-fiber-runtime n)]) + ;; Start the runtime on a background thread + (fork-thread (lambda () (fiber-runtime-run! rt))) + ;; Brief pause to let workers initialize + (sleep (make-time 'time-duration 10000000 0)) + (component-data-set! comp rt) + comp)) + ;; stop + (lambda (comp) + (let ([rt (component-data comp)]) + (when rt + (fiber-runtime-stop! rt))) + (component-data-set! comp #f) + comp)) + c)) + + ;; ========================================================================= + ;; HTTP Server Component + ;; + ;; Starts fiber-httpd on the given port. Depends on a fiber runtime + ;; component for scheduling. + ;; ========================================================================= + + (define (httpd-component port handler) + (let ([c (component 'http-server 'port port 'handler handler)]) + (register-lifecycle! 'http-server + ;; start + (lambda (comp) + (let* ([cfg (component-config comp)] + [port (hashtable-ref cfg 'port 8080)] + [handler (hashtable-ref cfg 'handler #f)] + ;; Get fiber runtime from dependencies + [deps (component-deps comp)] + [rt-dep (assoc 'fiber-runtime deps)] + [rt (and rt-dep (component-data (cdr rt-dep)))]) + (unless handler + (error 'httpd-component "no handler configured")) + ;; Start httpd — it will use the fiber runtime from + ;; the current-fiber-runtime parameter if set, or + ;; create its own + (let ([server + (if rt + ;; TODO: integrate with existing runtime + ;; For now, httpd starts its own runtime + handler + handler)]) + (component-data-set! comp (list 'port port 'handler handler)) + comp))) + ;; stop + (lambda (comp) + (let ([data (component-data comp)]) + ;; Graceful shutdown would go here + (component-data-set! comp #f) + comp))) + c)) + + ;; ========================================================================= + ;; Worker Component + ;; + ;; A background fiber that runs a thunk in a loop until shutdown. + ;; Depends on fiber-runtime. + ;; ========================================================================= + + (define (worker-component thunk) + (let ([c (component 'worker 'thunk thunk)]) + (register-lifecycle! 'worker + ;; start + (lambda (comp) + (let* ([cfg (component-config comp)] + [thunk (hashtable-ref cfg 'thunk #f)] + [deps (component-deps comp)] + [rt-dep (assoc 'fiber-runtime deps)] + [rt (and rt-dep (component-data (cdr rt-dep)))] + [stop-flag (box #f)]) + (when (and rt thunk) + (fiber-spawn rt + (lambda () + (let loop () + (unless (unbox stop-flag) + (guard (exn [#t (void)]) + (thunk)) + (fiber-yield) + (loop)))))) + (component-data-set! comp stop-flag) + comp)) + ;; stop + (lambda (comp) + (let ([stop-flag (component-data comp)]) + (when (and stop-flag (box? stop-flag)) + (set-box! stop-flag #t))) + (component-data-set! comp #f) + comp)) + c)) + + ;; ========================================================================= + ;; Graceful shutdown + ;; + ;; Stop a system with a timeout for draining in-flight work. + ;; ========================================================================= + + (define graceful-shutdown! + (case-lambda + [(sys) (graceful-shutdown! sys 5000)] + [(sys timeout-ms) + ;; Give in-flight work time to drain + (let ([deadline (+ (now-ms) timeout-ms)]) + ;; Stop in reverse dependency order (handled by component/stop) + (stop sys) + ;; Wait until deadline + (let ([remaining (- deadline (now-ms))]) + (when (> remaining 0) + (sleep (make-time 'time-duration + (* (mod remaining 1000) 1000000) + (quotient remaining 1000))))))])) + + (define (now-ms) + (let ([t (current-time 'time-monotonic)]) + (+ (* (time-second t) 1000) + (quotient (time-nanosecond t) 1000000)))) + + ;; ========================================================================= + ;; Internal: component-data accessor + ;; + ;; The component record stores user data in the `data` field. + ;; These are internal helpers that the lifecycle functions use. + ;; ========================================================================= + + (define (component-data c) + ;; Access the data field of the component record + ;; component-rec is defined in (std component) but the accessor + ;; isn't exported. We'll store data in config with a special key. + (hashtable-ref (component-config c) '%data #f)) + + (define (component-data-set! c val) + (hashtable-set! (component-config c) '%data val)) + +) ;; end library --- a/lib/std/csp/clj.sls +++ b/lib/std/csp/clj.sls @@ -1,39 +1,32 @@ #!chezscheme -;;; (std csp clj) — Clojure `core.async`-compatible surface +;;; (std csp clj) — Clojure `core.async`-compatible surface (fiber-aware) ;;; -;;; 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. +;;; A thin renaming layer over `(std csp)`, `(std csp select)`, +;;; `(std csp ops)`, and `(std csp fiber-chan)` 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!`.