Phase 2: Async I/O runtime (Steps 11-13)
ober
b035a1990abaf1cf8e9fe68a260f84ff845254b7
--- a/Makefile +++ b/Makefile @@ -83,6 +83,8 @@ test-features: @$(SCHEME) --libdirs $(LIBDIRS) --script tests/test-typed.ss @$(SCHEME) --libdirs $(LIBDIRS) --script tests/test-cache.ss @$(SCHEME) --libdirs $(LIBDIRS) --script tests/test-effect.ss + @$(SCHEME) --libdirs $(LIBDIRS) --script tests/test-async.ss + @$(SCHEME) --libdirs $(LIBDIRS) --script tests/test-iouring.ss test-all: test test-features test-wrappers new file mode 100644 --- /dev/null +++ b/lib/std/async.sls @@ -0,0 +1,199 @@ +#!chezscheme +;;; (std async) — Async I/O runtime built on algebraic effects +;;; +;;; Uses a thread-per-task model internally: +;;; - Each async task runs in its own thread +;;; - Async await blocks the thread until the promise resolves +;;; - Async spawn creates a new task thread +;;; - Async sleep sleeps the current thread +;;; +;;; The Async effect provides a clean API; threads handle actual suspension. +;;; +;;; API: +;;; Async::descriptor — effect descriptor +;;; (Async await promise) — block until promise resolves, return value +;;; (Async spawn thunk) — launch concurrent task (non-blocking) +;;; (Async sleep ms) — sleep for ms milliseconds +;;; (run-async thunk) — run thunk in async context, block until done +;;; (make-async-promise) — create a fulfillable promise +;;; (async-promise-resolve! p val) — fulfill a promise +;;; (async-channel-get ch) — get from channel, suspending via Async effect +;;; (async-channel-put ch val) — put to channel (async) +;;; (async-task thunk) — spawn a task, return a promise for its result + +(library (std async) + (export + ;; Async effect + Async + Async::descriptor + + ;; Event loop + run-async + run-async/workers + + ;; Promises + make-async-promise + async-promise? + async-promise-resolve! + async-promise-resolved? + async-promise-value + + ;; Task management + async-task + async-task? + + ;; Async channels + async-channel-get + async-channel-put + + ;; Async sleep + async-sleep) + + (import (chezscheme) (std effect) (std misc channel)) + + ;; ========== Async Effect Definition ========== + + (defeffect Async + (await promise) + (spawn thunk) + (sleep ms)) + + ;; ========== Promise ========== + + (define-record-type async-promise + (fields + (mutable resolved?) + (mutable value) + (immutable mutex) + (immutable cond)) + (protocol + (lambda (new) + (lambda () + (new #f #f (make-mutex) (make-condition))))) + (sealed #t)) + + (define (async-promise-resolve! p val) + (with-mutex (async-promise-mutex p) + (unless (async-promise-resolved? p) + (async-promise-resolved?-set! p #t) + (async-promise-value-set! p val) + (condition-broadcast (async-promise-cond p))))) + + ;; Block current OS thread until promise resolves, then return value. + (define (promise-wait! p) + (with-mutex (async-promise-mutex p) + (let loop () + (if (async-promise-resolved? p) + (async-promise-value p) + (begin + (condition-wait (async-promise-cond p) (async-promise-mutex p)) + (loop)))))) + + ;; ========== Async Effect Handlers ========== + ;; + ;; The handlers use thread-level blocking for true task suspension. + ;; Each handler receives (k arg ...) where k is the one-shot continuation. + ;; Instead of storing k and returning, handlers block the current thread + ;; until the effect completes, then resume by calling k. + ;; + ;; Since we run each task in a thread, this correctly suspends the task. + + (define (install-async-handlers! thunk) + (with-handler + ([Async + ;; await: block current thread until promise resolves + (await (k promise) + (let ([val (promise-wait! promise)]) + (resume k val))) + ;; spawn: fork a new thread for the task, resume immediately + (spawn (k task-thunk) + (fork-thread + (lambda () + (with-handler + ([Async + (await (k2 p) (resume k2 (promise-wait! p))) + (spawn (k2 t) + (fork-thread (lambda () (install-async-handlers! t))) + (resume k2 (void))) + (sleep (k2 ms) + (sleep (make-time 'time-duration + (fx* (fxmod ms 1000) 1000000) + (fxquotient ms 1000))) + (resume k2 (void)))]) + (task-thunk)))) + (resume k (void))) + ;; sleep: sleep the current thread + (sleep (k ms) + (sleep (make-time 'time-duration + (fx* (fxmod ms 1000) 1000000) + (fxquotient ms 1000))) + (resume k (void)))]) + (thunk))) + + ;; ========== run-async ========== + + (define (run-async thunk) + (let ([result-promise (make-async-promise)]) + ;; Run the thunk in a thread with Async handlers installed + (fork-thread + (lambda () + (guard (exn [#t + (fprintf (current-error-port) + "run-async error: ~a~%" + (if (message-condition? exn) (condition-message exn) exn)) + (async-promise-resolve! result-promise + (raise-continuable exn))]) + (install-async-handlers! + (lambda () + (let ([val (thunk)]) + (async-promise-resolve! result-promise val))))))) + ;; Block main thread until done + (promise-wait! result-promise))) + + ;; (run-async/workers thunk n) — same as run-async (threads handle workers) + (define (run-async/workers thunk n-workers) + ;; The n-workers hint is noted but not used (each spawn creates its own thread) + (run-async thunk)) + + ;; ========== async-task ========== + + (define (async-task thunk) + (let ([p (make-async-promise)]) + (Async spawn + (lambda () + (let ([v (thunk)]) + (async-promise-resolve! p v)))) + p)) + + (define async-task? async-promise?) + + ;; ========== async-sleep ========== + + (define (async-sleep ms) + (Async sleep ms)) + + ;; ========== Async Channels (Step 12) ========== + + ;; Get from channel. If empty, wait via Async await (suspends the task thread). + (define (async-channel-get ch) + (let-values ([(val ok) (channel-try-get ch)]) + (if ok + val + ;; Channel empty — create a promise and fulfill it when data arrives + (let ([p (make-async-promise)]) + (fork-thread + (lambda () + (let ([v (channel-get ch)]) ;; blocks this helper thread + (async-promise-resolve! p v)))) + (Async await p))))) + + ;; Put to channel. If bounded and full, wait via Async await. + (define (async-channel-put ch val) + (let ([p (make-async-promise)]) + (fork-thread + (lambda () + (channel-put ch val) ;; may block if bounded and full + (async-promise-resolve! p (void)))) + (Async await p))) + + ) ;; end library --- a/lib/std/effect.sls +++ b/lib/std/effect.sls @@ -17,7 +17,9 @@ perform resume effect-not-handled? - effect-perform) + effect-perform + *effect-handlers* + run-with-handler) (import (chezscheme)) new file mode 100644 --- /dev/null +++ b/lib/std/os/iouring.sls @@ -0,0 +1,238 @@ +#!chezscheme +;;; (std os iouring) — Linux io_uring via liburing +;;; +;;; Provides zero-copy async I/O using io_uring (Linux 5.1+). +;;; Requires: liburing.so.2 +;;; +;;; API: +;;; (iouring-available?) — #t if liburing.so is present +;;; (make-iouring [depth]) — initialize ring +;;; (iouring-close! ring) — shut down ring +;;; (iouring-read! ring fd buf n) — async read, returns promise<bytes-read> +;;; (iouring-write! ring fd buf n) — async write, returns promise<bytes-written> +;;; (iouring-accept! ring fd) — async accept, returns promise<client-fd> +;;; (iouring-submit! ring) — submit pending SQEs +;;; (iouring-wait! ring) — wait for 1 completion +;;; (run-iouring-loop ring) — completion loop in background thread + +(library (std os iouring) + (export + iouring-available? + make-iouring + iouring? + iouring-ring-addr + iouring-pending + iouring-close! + iouring-nop! + iouring-read! + iouring-write! + iouring-accept! + iouring-submit! + iouring-wait! + run-iouring-loop) + + (import (chezscheme) (std async)) + + ;; ========== Constants ========== + + ;; io_uring struct size (liburing 2.x is 216 bytes; we use 256 for safety) + (define *ring-struct-size* 256) + + ;; ========== liburing availability ========== + + ;; liburing-ffi.so.2 exposes all inline functions as real symbols + (define *liburing-available* + (guard (exn [#t #f]) + (load-shared-object "liburing-ffi.so.2") + #t)) + + (define (iouring-available?) *liburing-available*) + + ;; ========== FFI stubs (replaced when library is present) ========== + + (define (not-available . args) + (error 'iouring "liburing not available; install liburing2")) + + (define io-uring-queue-init not-available) + (define io-uring-queue-exit not-available) + (define io-uring-get-sqe not-available) + (define io-uring-submit not-available) + (define io-uring-wait-cqe not-available) + (define io-uring-cqe-seen not-available) + (define io-uring-sqe-set-data64 not-available) + (define io-uring-cqe-get-data64 not-available) + (define io-uring-prep-read not-available) + (define io-uring-prep-write not-available) + (define io-uring-prep-accept not-available) + + ;; ========== iouring record ========== + + (define-record-type (iouring %make-iouring iouring?) + (fields + (immutable ring-addr) ;; uptr: address of foreign-alloc'd io_uring struct + (immutable depth) ;; queue depth + (mutable pending) ;; eq-hashtable: op-id -> async-promise + (immutable mutex)) ;; protects pending table + (sealed #t)) + + ;; ========== Operation ID counter ========== + + (define *next-op-id* 0) + (define *op-id-mutex* (make-mutex)) + + (define (next-op-id!) + (with-mutex *op-id-mutex* + (let ([id *next-op-id*]) + (set! *next-op-id* (+ id 1)) + id))) + + ;; ========== make-iouring ========== + + (define make-iouring + (case-lambda + [() (make-iouring-impl 256)] + [(depth) (make-iouring-impl depth)])) + + (define (make-iouring-impl depth) + (unless *liburing-available* + (error 'make-iouring "liburing not available; install liburing2")) + (let ([ring-addr (foreign-alloc *ring-struct-size*)]) + (do ([i 0 (+ i 1)]) + ((= i *ring-struct-size*)) + (foreign-set! 'unsigned-8 ring-addr i 0)) + (let ([ret (io-uring-queue-init depth ring-addr 0)]) + (when (< ret 0) + (foreign-free ring-addr) + (error 'make-iouring "io_uring_queue_init failed, errno" (- ret))) + (%make-iouring ring-addr depth (make-eq-hashtable) (make-mutex))))) + + ;; ========== iouring-close! ========== + + (define (iouring-close! ring) + (io-uring-queue-exit (iouring-ring-addr ring)) + (foreign-free (iouring-ring-addr ring))) + + ;; ========== Internal: submit one op, return promise ========== + + (define (iouring-op! ring prep-thunk) + (let ([sqe-addr (io-uring-get-sqe (iouring-ring-addr ring))]) + (when (zero? sqe-addr) + (error 'iouring-op! "submission queue full")) + (let ([op-id (next-op-id!)] + [p (make-async-promise)]) + (prep-thunk sqe-addr) + (io-uring-sqe-set-data64 sqe-addr op-id) + (with-mutex (iouring-mutex ring) + (hashtable-set! (iouring-pending ring) op-id p)) + p))) + + ;; ========== iouring-nop! ========== + ;; Submit a no-op for testing ring functionality. + + (define (iouring-nop! ring) + (iouring-op! ring + (lambda (sqe) + ((foreign-procedure "io_uring_prep_nop" (uptr) void) sqe)))) + + ;; ========== iouring-read! ========== + ;; buf must be a uptr (foreign-alloc'd buffer); returns promise<bytes-read>. + + (define (iouring-read! ring fd buf-addr n) + (iouring-op! ring + (lambda (sqe) + (io-uring-prep-read sqe fd buf-addr n 0)))) + + ;; ========== iouring-write! ========== + + (define (iouring-write! ring fd buf-addr n) + (iouring-op! ring + (lambda (sqe) + (io-uring-prep-write sqe fd buf-addr n 0)))) + + ;; ========== iouring-accept! ========== + + (define (iouring-accept! ring fd) + (iouring-op! ring + (lambda (sqe) + (io-uring-prep-accept sqe fd 0 0 0)))) + + ;; ========== iouring-submit! ========== + + (define (iouring-submit! ring) + (let ([ret (io-uring-submit (iouring-ring-addr ring))]) + (when (< ret 0) + (error 'iouring-submit! "io_uring_submit failed" ret)) + ret)) + + ;; ========== iouring-wait! ========== + ;; Wait for one completion, resolve its promise. + + (define (iouring-wait! ring) + (let ([cqe-ptr-addr (foreign-alloc 8)]) + (foreign-set! 'unsigned-64 cqe-ptr-addr 0 0) + (let ([ret (io-uring-wait-cqe (iouring-ring-addr ring) cqe-ptr-addr)]) + (let ([cqe-addr (foreign-ref 'uptr cqe-ptr-addr 0)]) + (foreign-free cqe-ptr-addr) + (when (< ret 0) + (error 'iouring-wait! "io_uring_wait_cqe failed" ret)) + ;; io_uring_cqe: user_data(u64 @0), res(s32 @8), flags(u32 @12) + (let ([op-id (io-uring-cqe-get-data64 cqe-addr)] + [res (foreign-ref 'integer-32 cqe-addr 8)]) + (io-uring-cqe-seen (iouring-ring-addr ring) cqe-addr) + (let ([p (with-mutex (iouring-mutex ring) + (let ([entry (hashtable-ref (iouring-pending ring) op-id #f)]) + (when entry + (hashtable-delete! (iouring-pending ring) op-id)) + entry))]) + (when p + (async-promise-resolve! p res)))))))) + + ;; ========== run-iouring-loop ========== + + (define (run-iouring-loop ring) + (fork-thread + (lambda () + (let loop () + (guard (exn [#t (void)]) + (iouring-submit! ring) + (iouring-wait! ring)) + (loop))))) + + ;; ========== Initialize FFI when library is available ========== + + (when *liburing-available* + (set! io-uring-queue-init + (foreign-procedure "io_uring_queue_init" + (unsigned-32 uptr unsigned-32) int)) + (set! io-uring-queue-exit + (foreign-procedure "io_uring_queue_exit" + (uptr) void)) + (set! io-uring-get-sqe + (foreign-procedure "io_uring_get_sqe" + (uptr) uptr)) + (set! io-uring-submit + (foreign-procedure "io_uring_submit" + (uptr) int)) + (set! io-uring-wait-cqe + (foreign-procedure "io_uring_wait_cqe" + (uptr uptr) int)) + (set! io-uring-cqe-seen + (foreign-procedure "io_uring_cqe_seen" + (uptr uptr) void)) + (set! io-uring-sqe-set-data64 + (foreign-procedure "io_uring_sqe_set_data64" + (uptr unsigned-64) void)) + (set! io-uring-cqe-get-data64 + (foreign-procedure "io_uring_cqe_get_data64" + (uptr) unsigned-64)) + (set! io-uring-prep-read + (foreign-procedure "io_uring_prep_read" + (uptr int uptr unsigned-32 unsigned-64) void)) + (set! io-uring-prep-write + (foreign-procedure "io_uring_prep_write" + (uptr int uptr unsigned-32 unsigned-64) void)) + (set! io-uring-prep-accept + (foreign-procedure "io_uring_prep_accept" + (uptr int uptr uptr int) void))) + + ) ;; end library new file mode 100644 --- /dev/null +++ b/tests/test-async.ss @@ -0,0 +1,134 @@ +#!chezscheme +;;; Tests for (std async) — Async I/O runtime + +(import (chezscheme) (std async) (std misc channel)) + +(define pass 0) +(define fail 0) + +(define-syntax test + (syntax-rules () + [(_ name expr expected) + (guard (exn [#t (set! fail (+ fail 1)) + (printf "FAIL ~a: ~a~%" name + (if (message-condition? exn) (condition-message exn) exn))]) + (let ([got expr]) + (if (equal? got expected) + (begin (set! pass (+ pass 1)) (printf " ok ~a~%" name)) + (begin (set! fail (+ fail 1)) + (printf "FAIL ~a: got ~s expected ~s~%" name got expected)))))])) + +(printf "--- (std async) tests ---~%") + +;;; Test 1: run-async with a simple value +(test "run-async simple" + (run-async (lambda () 42)) + 42) + +;;; Test 2: run-async with spawn +(test "run-async spawn" + (run-async + (lambda () + (let ([result #f]) + ;; Spawn a task that sets result + (Async spawn (lambda () (set! result 'done))) + ;; Give spawned task time to run (sleep briefly) + (async-sleep 10) + result))) + 'done) + +;;; Test 3: async-task returns a promise +(test "async-task/promise" + (run-async + (lambda () + (let ([p (async-task (lambda () (+ 1 2)))]) + ;; Wait for the task to complete + (async-sleep 20) + (if (async-promise-resolved? p) + (async-promise-value p) + 'not-resolved)))) + 3) + +;;; Test 4: promises resolve correctly +(test "promise/resolve" + (let ([p (make-async-promise)]) + (async-promise-resolve! p 99) + (and (async-promise-resolved? p) + (async-promise-value p))) + 99) + +;;; Test 5: promise can only be resolved once +(test "promise/resolve-once" + (let ([p (make-async-promise)]) + (async-promise-resolve! p 'first) + (async-promise-resolve! p 'second) ;; should be ignored + (async-promise-value p)) + 'first) + +;;; Test 6: Async await on pre-resolved promise +(test "await/pre-resolved" + (run-async + (lambda () + (let ([p (make-async-promise)]) + (async-promise-resolve! p 'ready) + (Async await p)))) + 'ready) + +;;; Test 7: async sleep +(test "async-sleep" + (run-async + (lambda () + (let ([t0 (current-time 'time-monotonic)]) + (async-sleep 50) + (let ([t1 (current-time 'time-monotonic)]) + ;; Should have slept at least ~50ms + (let ([elapsed-ms + (+ (* 1000 (- (time-second t1) (time-second t0))) + (quotient (- (time-nanosecond t1) (time-nanosecond t0)) 1000000))]) + (>= elapsed-ms 40)))))) ;; allow some slack + #t) + +;;; Test 8: async channels — basic get/put +(test "async-channel/basic" + (run-async + (lambda () + (let ([ch (make-channel)]) + (Async spawn (lambda () (async-channel-put ch 'hello))) + (async-sleep 20) + (let-values ([(v ok) (channel-try-get ch)]) + (if ok v 'empty))))) + 'hello) + +;;; Test 9: multiple spawned tasks +(test "async/multiple-tasks" + (run-async + (lambda () + (let ([results (make-channel)]) + (Async spawn (lambda () (async-channel-put results 1))) + (Async spawn (lambda () (async-channel-put results 2))) + (Async spawn (lambda () (async-channel-put results 3))) + (async-sleep 50) + (let ([got '()]) + (let loop () + (let-values ([(v ok) (channel-try-get results)]) + (when ok + (set! got (cons v got)) + (loop)))) + (length got))))) + 3) + +;;; Test 10: run-async/workers +(test "run-async/workers" + (run-async/workers + (lambda () + (let ([p (async-task (lambda () (* 6 7)))]) + (async-sleep 30) + (if (async-promise-resolved? p) + (async-promise-value p) + 'timeout))) + 2) + 42) + +(printf "~%~a tests: ~a passed, ~a failed~%" + (+ pass fail) pass fail) +(when (> fail 0) (exit 1)) new file mode 100644 --- /dev/null +++ b/tests/test-iouring.ss @@ -0,0 +1,81 @@ +#!chezscheme +;;; Tests for (std os iouring) — io_uring async I/O + +(import (chezscheme) (std os iouring) (std async)) + +(define pass 0) +(define fail 0) + +(define-syntax test + (syntax-rules () + [(_ name expr expected) + (guard (exn [#t (set! fail (+ fail 1)) + (printf "FAIL ~a: ~a~%" name + (if (message-condition? exn) (condition-message exn) exn))]) + (let ([got expr]) + (if (equal? got expected) + (begin (set! pass (+ pass 1)) (printf " ok ~a~%" name)) + (begin (set! fail (+ fail 1)) + (printf "FAIL ~a: got ~s expected ~s~%" name got expected)))))])) + +(printf "--- (std os iouring) tests ---~%") + +(if (not (iouring-available?)) + (begin + (printf " SKIP: liburing-ffi.so.2 not available~%")) + (begin + + ;; Test 1: availability check + (test "iouring-available?" (iouring-available?) #t) + + ;; Test 2: create a ring with default depth + (test "make-iouring/default" + (let ([ring (make-iouring)]) + (let ([ok (iouring? ring)]) + (iouring-close! ring) + ok)) + #t) + + ;; Test 3: create a ring with explicit depth + (test "make-iouring/depth" + (let ([ring (make-iouring 8)]) + (let ([ok (iouring? ring)]) + (iouring-close! ring) + ok)) + #t) + + ;; Test 4: submit with no ops returns 0 + (test "iouring-submit!/empty" + (let ([ring (make-iouring 16)]) + (let ([n (iouring-submit! ring)]) + (iouring-close! ring) + n)) + 0) + + ;; Test 5: multiple rings can coexist + (test "iouring/multiple-rings" + (let ([ring1 (make-iouring 8)] + [ring2 (make-iouring 8)]) + (let ([ok (and (iouring? ring1) (iouring? ring2))]) + (iouring-close! ring1) + (iouring-close! ring2) + ok)) + #t) + + ;; Test 6: async NOP op via io_uring + (test "iouring/nop-completion" + (run-async + (lambda () + (let ([ring (make-iouring 16)]) + (let ([p (iouring-nop! ring)]) + (iouring-submit! ring) + (run-iouring-loop ring) + (let ([result (Async await p)]) + (iouring-close! ring) + ;; NOP returns 0 on success + result))))) + 0))) + +(printf "~%~a tests: ~a passed, ~a failed~%" + (+ pass fail) pass fail) +(when (> fail 0) (exit 1))