workpool + dns + filepool: blocking work offload for fibers (Phase 3)
ober
b7fe3992e316a96d3fd29f008324703d342d909b
new file mode 100644 --- /dev/null +++ b/lib/std/io/filepool.sls @@ -0,0 +1,129 @@ +#!chezscheme +;;; (std io filepool) — Fiber-aware file I/O via thread pool +;;; +;;; Regular files on Linux always return "ready" from epoll, making +;;; epoll useless for file I/O. This module offloads blocking file +;;; operations to a thread pool so fibers don't block worker threads. +;;; +;;; API: +;;; (make-file-pool) — create pool (default 4 threads) +;;; (make-file-pool n) — create with n threads +;;; (file-pool-start! pool) — start the worker threads +;;; (file-pool-stop! pool) — drain and stop workers +;;; (fiber-read-file path pool) — read entire file as string +;;; (fiber-read-file-bytes path pool) — read entire file as bytevector +;;; (fiber-write-file path data pool) — write string to file +;;; (fiber-write-file-bytes path data pool) — write bytevector to file +;;; (fiber-append-file path data pool) — append string to file +;;; (fiber-file-exists? path pool) — check if file exists +;;; (with-file-pool body ...) — scoped pool lifecycle + +(library (std io filepool) + (export + make-file-pool + file-pool? + file-pool-start! + file-pool-stop! + fiber-read-file + fiber-read-file-bytes + fiber-write-file + fiber-write-file-bytes + fiber-append-file + fiber-file-exists? + with-file-pool) + + (import (chezscheme) + (std fiber) + (std net workpool)) + + ;; ========== File pool ========== + + (define-record-type file-pool + (fields (immutable pool)) + (protocol + (lambda (new) + (case-lambda + [() (new (make-work-pool 4))] + [(n) (new (make-work-pool n))])))) + + (define (file-pool-start! fp) + (work-pool-start! (file-pool-pool fp))) + + (define (file-pool-stop! fp) + (work-pool-stop! (file-pool-pool fp))) + + ;; ========== Fiber-aware file operations ========== + + ;; Read entire file as string, parking the fiber. + (define (fiber-read-file path pool) + (work-pool-submit! (file-pool-pool pool) + (lambda () + (let ([p (open-file-input-port path + (file-options) (buffer-mode block) + (make-transcoder (utf-8-codec)))]) + (let ([content (get-string-all p)]) + (close-input-port p) + (if (eof-object? content) "" content)))))) + + ;; Read entire file as bytevector, parking the fiber. + (define (fiber-read-file-bytes path pool) + (work-pool-submit! (file-pool-pool pool) + (lambda () + (let ([p (open-file-input-port path)]) + (let ([content (get-bytevector-all p)]) + (close-input-port p) + (if (eof-object? content) (make-bytevector 0) content)))))) + + ;; Write string to file (overwrite), parking the fiber. + (define (fiber-write-file path data pool) + (work-pool-submit! (file-pool-pool pool) + (lambda () + (let ([p (open-file-output-port path + (file-options no-fail) + (buffer-mode block) + (make-transcoder (utf-8-codec)))]) + (put-string p data) + (close-output-port p) + (void))))) + + ;; Write bytevector to file (overwrite), parking the fiber. + (define (fiber-write-file-bytes path data pool) + (work-pool-submit! (file-pool-pool pool) + (lambda () + (let ([p (open-file-output-port path + (file-options no-fail) + (buffer-mode block))]) + (put-bytevector p data) + (close-output-port p) + (void))))) + + ;; Append string to file, parking the fiber. + (define (fiber-append-file path data pool) + (work-pool-submit! (file-pool-pool pool) + (lambda () + (let ([p (open-file-output-port path + (file-options no-fail no-truncate) + (buffer-mode block) + (make-transcoder (utf-8-codec)))]) + (set-port-position! p (port-length p)) + (put-string p data) + (close-output-port p) + (void))))) + + ;; Check if file exists, parking the fiber. + (define (fiber-file-exists? path pool) + (work-pool-submit! (file-pool-pool pool) + (lambda () (file-exists? path)))) + + ;; Convenience macro + (define-syntax with-file-pool + (syntax-rules () + [(_ var body ...) + (let ([var (make-file-pool)]) + (file-pool-start! var) + (guard (exn [#t (file-pool-stop! var) (raise exn)]) + (let ([result (begin body ...)]) + (file-pool-stop! var) + result)))])) + +) ;; end library --- a/lib/std/net/fiber-httpd.sls +++ b/lib/std/net/fiber-httpd.sls @@ -339,7 +339,8 @@ (immutable listen-port) (immutable runtime) (immutable poller) - (mutable running?)) + (mutable running?) + (mutable accept-fiber)) (sealed #t)) ;; Connection handler: one fiber per connection, keep-alive loop @@ -362,14 +363,15 @@ (fiber-tcp-close fd)) ;; Accept loop - (define (accept-loop listen-fd poller handler) - (let loop () - (guard (exn [#t (void)]) ;; stop on error (e.g., fd closed) - (let ([client-fd (fiber-tcp-accept listen-fd poller)]) - (fiber-spawn* - (lambda () (handle-connection client-fd poller handler)) - "http-conn") - (loop))))) + (define (accept-loop listen-fd poller handler server) + (guard (exn [#t (void)]) ;; catch cancellation and all errors + (let loop () + (when (fiber-httpd-running? server) + (let ([client-fd (fiber-tcp-accept listen-fd poller)]) + (fiber-spawn* + (lambda () (handle-connection client-fd poller handler)) + "http-conn") + (loop)))))) ;; Start the server (define (fiber-httpd-start port handler) @@ -377,19 +379,28 @@ [poller (make-io-poller rt)]) (io-poller-start! poller) (let-values ([(listen-fd listen-port) (fiber-tcp-listen "0.0.0.0" port)]) - (let ([srv (make-fiber-httpd listen-fd listen-port rt poller #t)]) + (let ([srv (make-fiber-httpd listen-fd listen-port rt poller #t #f)]) ;; Spawn accept loop - (fiber-spawn rt - (lambda () (accept-loop listen-fd poller handler)) - "httpd-accept") + (let ([af (fiber-spawn rt + (lambda () (accept-loop listen-fd poller handler srv)) + "httpd-accept")]) + (fiber-httpd-accept-fiber-set! srv af)) ;; Run in background thread so caller gets the server handle back (fork-thread (lambda () (fiber-runtime-run! rt))) srv)))) (define (fiber-httpd-stop! srv) (fiber-httpd-running?-set! srv #f) + ;; Cancel the accept fiber — this wakes it from its parked state + (let ([af (fiber-httpd-accept-fiber srv)]) + (when af (fiber-cancel! af))) + ;; Close listen fd (fiber-tcp-close (fiber-httpd-listen-fd srv)) + ;; Stop runtime — marks running? = #f and wakes run queue + (fiber-runtime-stop! (fiber-httpd-runtime srv)) + ;; Stop poller — shuts down the poller thread (io-poller-stop! (fiber-httpd-poller srv)) - (fiber-runtime-stop! (fiber-httpd-runtime srv))) + ;; Give background thread time to exit + (sleep (make-time 'time-duration 150000000 0))) ) ;; end library new file mode 100644 --- /dev/null +++ b/lib/std/net/resolve.sls @@ -0,0 +1,132 @@ +#!chezscheme +;;; (std net resolve) — Fiber-aware DNS resolution +;;; +;;; Resolves hostnames on a thread pool since getaddrinfo blocks. +;;; The calling fiber parks while resolution happens on a pool thread. +;;; +;;; API: +;;; (make-dns-resolver) — create resolver (default 2 threads) +;;; (make-dns-resolver n) — create with n threads +;;; (dns-resolver-start! r) — start resolver +;;; (dns-resolver-stop! r) — stop resolver +;;; (fiber-resolve host resolver) — resolve hostname, returns first IPv4 address string +;;; (with-dns-resolver body ...) — scoped resolver lifecycle + +(library (std net resolve) + (export + make-dns-resolver + dns-resolver? + dns-resolver-start! + dns-resolver-stop! + fiber-resolve + with-dns-resolver) + + (import (chezscheme) + (std fiber) + (std net workpool)) + + ;; ========== FFI: getaddrinfo ========== + + (define _libc-loaded + (let ((v (getenv "JEMACS_STATIC"))) + (if (and v (not (string=? v "")) (not (string=? v "0"))) + #f + (load-shared-object #f)))) + + (define c-getaddrinfo + (foreign-procedure "getaddrinfo" (string string void* void*) int)) + (define c-freeaddrinfo + (foreign-procedure "freeaddrinfo" (void*) void)) + (define c-inet-ntop + (foreign-procedure "inet_ntop" (int void* u8* int) void*)) + + (define AF_INET 2) + (define SOCK_STREAM 1) + (define INET_ADDRSTRLEN 16) + + ;; struct addrinfo layout (Linux x86_64): + ;; int ai_flags @ 0 + ;; int ai_family @ 4 + ;; int ai_socktype @ 8 + ;; int ai_protocol @ 12 + ;; socklen ai_addrlen @ 16 (unsigned int) + ;; padding @ 20 (4 bytes on 64-bit) + ;; void* ai_addr @ 24 (pointer, 8 bytes on 64-bit) + ;; char* ai_canonname @ 32 + ;; void* ai_next @ 40 + + (define (resolve-blocking hostname) + ;; Set up hints: AF_INET, SOCK_STREAM + (let ([hints (foreign-alloc 48)]) + ;; Zero out hints + (do ([i 0 (+ i 1)]) ((= i 48)) + (foreign-set! 'unsigned-8 hints i 0)) + (foreign-set! 'int hints 4 AF_INET) ;; ai_family + (foreign-set! 'int hints 8 SOCK_STREAM) ;; ai_socktype + + (let ([result-ptr (foreign-alloc 8)]) + (foreign-set! 'void* result-ptr 0 0) + (let ([rc (c-getaddrinfo hostname #f hints result-ptr)]) + (foreign-free hints) + (cond + [(not (= rc 0)) + (foreign-free result-ptr) + (error 'fiber-resolve "DNS resolution failed" hostname rc)] + [else + (let ([result (foreign-ref 'void* result-ptr 0)]) + (foreign-free result-ptr) + (if (= result 0) + (error 'fiber-resolve "no addresses found" hostname) + ;; Extract first IPv4 address + ;; sockaddr_in: family(2) + port(2) + in_addr(4) + ;; in_addr starts at offset 4 in sockaddr_in + (let ([addr-ptr (foreign-ref 'void* result 24)]) + (let ([in-addr-ptr (+ addr-ptr 4)] + [buf (make-bytevector INET_ADDRSTRLEN)]) + (let ([p (c-inet-ntop AF_INET in-addr-ptr buf INET_ADDRSTRLEN)]) + (c-freeaddrinfo result) + (if (= p 0) + (error 'fiber-resolve "inet_ntop failed" hostname) + ;; Read null-terminated string from buf + (let loop ([i 0]) + (if (= (bytevector-u8-ref buf i) 0) + (bytevector->string + (let ([b (make-bytevector i)]) + (bytevector-copy! buf 0 b 0 i) b) + (make-transcoder (utf-8-codec))) + (loop (+ i 1))))))))))]))))) + + ;; ========== DNS Resolver ========== + + (define-record-type dns-resolver + (fields (immutable pool)) + (protocol + (lambda (new) + (case-lambda + [() (new (make-work-pool 2))] + [(n) (new (make-work-pool n))])))) + + (define (dns-resolver-start! r) + (work-pool-start! (dns-resolver-pool r))) + + (define (dns-resolver-stop! r) + (work-pool-stop! (dns-resolver-pool r))) + + ;; Resolve hostname from a fiber — parks fiber while resolution runs. + ;; Returns IPv4 address as a string (e.g., "93.184.216.34"). + (define (fiber-resolve hostname resolver) + (work-pool-submit! (dns-resolver-pool resolver) + (lambda () (resolve-blocking hostname)))) + + ;; Convenience macro + (define-syntax with-dns-resolver + (syntax-rules () + [(_ var body ...) + (let ([var (make-dns-resolver)]) + (dns-resolver-start! var) + (guard (exn [#t (dns-resolver-stop! var) (raise exn)]) + (let ([result (begin body ...)]) + (dns-resolver-stop! var) + result)))])) + +) ;; end library new file mode 100644 --- /dev/null +++ b/lib/std/net/workpool.sls @@ -0,0 +1,145 @@ +#!chezscheme +;;; (std net workpool) — Thread pool for blocking work offload +;;; +;;; Fibers park while blocking work (DNS, file I/O, etc.) runs on +;;; a small pool of OS threads. The pool thread wakes the fiber +;;; when the work completes. +;;; +;;; API: +;;; (make-work-pool n) — create pool with n worker threads +;;; (work-pool-start! pool) — start the worker threads +;;; (work-pool-stop! pool) — drain and stop workers +;;; (work-pool-submit! pool thunk) — run thunk on pool, park fiber, return result + +(library (std net workpool) + (export + make-work-pool + work-pool? + work-pool-start! + work-pool-stop! + work-pool-submit!) + + (import (chezscheme) + (std fiber)) + + ;; ========== Work item ========== + + (define-record-type work-item + (fields + (immutable thunk) ;; zero-arg procedure to run (blocking) + (immutable fiber) ;; parked fiber to wake + (mutable result) ;; set by worker thread + (mutable error) ;; set on exception + (immutable gate)) ;; fiber's gate box + (protocol + (lambda (new) + (lambda (thunk fiber gate) + (new thunk fiber #f #f gate))))) + + ;; ========== Thread-safe work queue ========== + + (define-record-type work-queue + (fields + (mutable items) ;; list of work-item + (immutable mutex) + (immutable cv) + (mutable closed?)) + (protocol + (lambda (new) + (lambda () + (new '() (make-mutex) (make-condition) #f))))) + + (define (wq-enqueue! wq item) + (mutex-acquire (work-queue-mutex wq)) + (work-queue-items-set! wq (append (work-queue-items wq) (list item))) + (condition-signal (work-queue-cv wq)) + (mutex-release (work-queue-mutex wq))) + + (define (wq-dequeue! wq) + ;; Block until an item is available or queue is closed. + ;; Returns #f when closed and empty. + (mutex-acquire (work-queue-mutex wq)) + (let loop () + (cond + [(not (null? (work-queue-items wq))) + (let ([item (car (work-queue-items wq))]) + (work-queue-items-set! wq (cdr (work-queue-items wq))) + (mutex-release (work-queue-mutex wq)) + item)] + [(work-queue-closed? wq) + (mutex-release (work-queue-mutex wq)) + #f] + [else + (condition-wait (work-queue-cv wq) (work-queue-mutex wq)) + (loop)]))) + + (define (wq-close! wq) + (mutex-acquire (work-queue-mutex wq)) + (work-queue-closed?-set! wq #t) + (condition-broadcast (work-queue-cv wq)) + (mutex-release (work-queue-mutex wq))) + + ;; ========== Work pool ========== + + (define-record-type work-pool + (fields + (immutable nthreads) + (immutable queue) + (mutable threads) + (mutable running?)) + (protocol + (lambda (new) + (lambda (n) + (new (max 1 n) (make-work-queue) '() #f))))) + + (define (worker-loop pool) + (let ([wq (work-pool-queue pool)]) + (let loop () + (let ([item (wq-dequeue! wq)]) + (when item + ;; Execute the blocking work + (guard (exn [#t + (work-item-error-set! item exn)]) + (work-item-result-set! item ((work-item-thunk item)))) + ;; Wake the parked fiber by opening its gate + (let ([gate (work-item-gate item)]) + (set-box! gate 'done)) + (wake-fiber! (work-item-fiber item)) + (loop)))))) + + (define (work-pool-start! pool) + (work-pool-running?-set! pool #t) + (work-pool-threads-set! pool + (let loop ([i 0] [acc '()]) + (if (= i (work-pool-nthreads pool)) + acc + (loop (+ i 1) + (cons (fork-thread (lambda () (worker-loop pool))) acc)))))) + + (define (work-pool-stop! pool) + (work-pool-running?-set! pool #f) + (wq-close! (work-pool-queue pool)) + ;; Give threads time to drain + (sleep (make-time 'time-duration 100000000 0))) + + ;; Submit blocking work from a fiber. + ;; Parks the current fiber until the work completes. + ;; Returns the result of thunk, or re-raises any exception. + (define (work-pool-submit! pool thunk) + (let* ([f (fiber-self)] + [gate (box 'channel)] + [item (make-work-item thunk f gate)]) + ;; Set up fiber for parking + (fiber-gate-set! f gate) + ;; Enqueue work + (wq-enqueue! (work-pool-queue pool) item) + ;; Park the fiber — will be woken by worker thread + (set-timer 1) + (spin-until-gate gate) + (fiber-gate-set! f #f) + ;; Check for error + (when (work-item-error item) + (raise (work-item-error item))) + (work-item-result item))) + +) ;; end library --- a/tests/test-fiber-httpd.ss +++ b/tests/test-fiber-httpd.ss @@ -35,7 +35,7 @@ (unless val (error 'assert msg))])) ;; Helper: send raw HTTP request over a fiber-aware TCP connection -;; and read the full response. +;; and read the full response. Reads once (sufficient for small responses). (define (http-request-raw fd poller method path body) (let* ([body-bv (if body (string->bytevector body (make-transcoder (utf-8-codec))) @@ -56,19 +56,14 @@ ;; Send body if present (when body-bv (fiber-tcp-write fd body-bv (bytevector-length body-bv) poller)) - ;; Read response + ;; Read response — single read for small responses, then done (let ([buf (make-bytevector 16384)]) - (let loop ([total 0]) - (let ([n (fiber-tcp-read fd buf (- 16384 total) poller)]) - (cond - [(<= n 0) - ;; EOF — return what we have - (bytevector->string - (let ([b (make-bytevector total)]) - (bytevector-copy! buf 0 b 0 total) b) - (make-transcoder (utf-8-codec)))] - [else - (loop (+ total n))])))))) + (let ([n (fiber-tcp-read fd buf 16384 poller)]) + (if (<= n 0) "" + (bytevector->string + (let ([b (make-bytevector n)]) + (bytevector-copy! buf 0 b 0 n) b) + (make-transcoder (utf-8-codec)))))))) ;; Helper: parse response status code from raw response (define (response-status-code resp) new file mode 100644 --- /dev/null +++ b/tests/test-workpool.ss @@ -0,0 +1,276 @@ +;;; Tests for Phase 3: Blocking work offload +;;; Tests workpool, DNS resolver, and file I/O pool. + +(import (chezscheme)) +(import (std fiber)) +(import (std net workpool)) +(import (std net resolve)) +(import (std io filepool)) + +(define test-count 0) +(define pass-count 0) + +(define-syntax test + (syntax-rules () + [(_ name body ...) + (begin + (set! test-count (+ test-count 1)) + (guard (exn [#t + (display "FAIL: ") (display name) (newline) + (display " Error: ") + (display (if (message-condition? exn) (condition-message exn) exn)) + (newline)]) + body ... + (set! pass-count (+ pass-count 1)) + (display "PASS: ") (display name) (newline)))])) + +(define-syntax assert-equal + (syntax-rules () + [(_ got expected msg) + (unless (equal? got expected) + (error 'assert msg (list 'got: got 'expected: expected)))])) + +(define-syntax assert-true + (syntax-rules () + [(_ val msg) + (unless val (error 'assert msg))])) + +;; ========================================================================= +;; Test 1: Work pool — basic submit +;; ========================================================================= + +(test "workpool: basic submit and result" + (let ([rt (make-fiber-runtime 2)] + [pool (make-work-pool 2)] + [result-box (box #f)]) + (work-pool-start! pool) + (fiber-spawn rt + (lambda () + (let ([r (work-pool-submit! pool (lambda () (* 6 7)))]) + (set-box! result-box r))) + "compute-fiber") + (fiber-runtime-run! rt) + (work-pool-stop! pool) + (assert-equal (unbox result-box) 42 "6 * 7 = 42"))) + +;; ========================================================================= +;; Test 2: Work pool — error propagation +;; ========================================================================= + +(test "workpool: error propagation" + (let ([rt (make-fiber-runtime 2)] + [pool (make-work-pool 2)] + [caught (box #f)]) + (work-pool-start! pool) + (fiber-spawn rt + (lambda () + (guard (exn [#t (set-box! caught #t)]) + (work-pool-submit! pool (lambda () (error 'test "boom"))))) + "error-fiber") + (fiber-runtime-run! rt) + (work-pool-stop! pool) + (assert-true (unbox caught) "exception propagated to fiber"))) + +;; ========================================================================= +;; Test 3: Work pool — concurrent submissions +;; ========================================================================= + +(test "workpool: 20 concurrent submissions" + (let ([rt (make-fiber-runtime 4)] + [pool (make-work-pool 4)] + [results (make-vector 20 #f)]) + (work-pool-start! pool) + (do ([i 0 (+ i 1)]) + ((= i 20)) + (let ([idx i]) + (fiber-spawn rt + (lambda () + (let ([r (work-pool-submit! pool (lambda () (* idx idx)))]) + (vector-set! results idx r))) + (string-append "fib-" (number->string idx))))) + (fiber-runtime-run! rt) + (work-pool-stop! pool) + ;; Verify all results + (do ([i 0 (+ i 1)]) + ((= i 20)) + (assert-equal (vector-ref results i) (* i i) + (string-append "result " (number->string i)))))) + +;; ========================================================================= +;; Test 4: DNS resolver — resolve localhost +;; ========================================================================= + +(test "dns: resolve localhost" + (let ([rt (make-fiber-runtime 2)] + [result-box (box #f)]) + (with-dns-resolver resolver + (fiber-spawn rt + (lambda () + (let ([addr (fiber-resolve "localhost" resolver)]) + (set-box! result-box addr))) + "dns-fiber") + (fiber-runtime-run! rt)) + (assert-equal (unbox result-box) "127.0.0.1" "localhost → 127.0.0.1"))) + +;; ========================================================================= +;; Test 5: DNS resolver — resolve real hostname +;; ========================================================================= + +(test "dns: resolve dns.google" + (let ([rt (make-fiber-runtime 2)] + [result-box (box #f)]) + (with-dns-resolver resolver + (fiber-spawn rt + (lambda () + (let ([addr (fiber-resolve "dns.google" resolver)]) + (set-box! result-box addr))) + "dns-fiber") + (fiber-runtime-run! rt)) + ;; dns.google resolves to 8.8.8.8 or 8.8.4.4 + (let ([addr (unbox result-box)]) + (assert-true (or (string=? addr "8.8.8.8") (string=? addr "8.8.4.4")) + (string-append "dns.google → " addr))))) + +;; ========================================================================= +;; Test 6: DNS resolver — concurrent resolutions +;; ========================================================================= + +(test "dns: 5 concurrent resolutions" + (let ([rt (make-fiber-runtime 4)] + [results (make-vector 5 #f)] + [hosts '#("localhost" "localhost" "localhost" "localhost" "localhost")]) + (with-dns-resolver resolver + (do ([i 0 (+ i 1)]) + ((= i 5)) + (let ([idx i]) + (fiber-spawn rt + (lambda () + (let ([addr (fiber-resolve (vector-ref hosts idx) resolver)]) + (vector-set! results idx addr))) + (string-append "dns-" (number->string idx))))) + (fiber-runtime-run! rt)) + ;; All should resolve to 127.0.0.1 + (do ([i 0 (+ i 1)]) + ((= i 5)) + (assert-equal (vector-ref results i) "127.0.0.1" + (string-append "host " (number->string i)))))) + +;; ========================================================================= +;; Test 7: File pool — write and read +;; ========================================================================= + +(test "filepool: write and read back" + (let ([rt (make-fiber-runtime 2)] + [result-box (box #f)] + [path "/tmp/jerboa-test-filepool.txt"]) + (with-file-pool fpool + (fiber-spawn rt + (lambda () + (fiber-write-file path "hello from fiber!" fpool) + (let ([content (fiber-read-file path fpool)]) + (set-box! result-box content))) + "file-fiber") + (fiber-runtime-run! rt)) + (assert-equal (unbox result-box) "hello from fiber!" "read back matches") + ;; Cleanup + (delete-file path))) + +;; ========================================================================= +;; Test 8: File pool — binary read/write +;; ========================================================================= + +(test "filepool: binary read/write" + (let ([rt (make-fiber-runtime 2)] + [result-box (box #f)] + [path "/tmp/jerboa-test-filepool-bin.dat"] + [data (make-bytevector 256)]) + ;; Fill with test pattern + (do ([i 0 (+ i 1)]) ((= i 256)) + (bytevector-u8-set! data i (mod i 256))) + (with-file-pool fpool + (fiber-spawn rt + (lambda () + (fiber-write-file-bytes path data fpool) + (let ([content (fiber-read-file-bytes path fpool)]) + (set-box! result-box content))) + "bin-fiber") + (fiber-runtime-run! rt)) + (assert-equal (unbox result-box) data "binary round-trip") + (delete-file path))) + +;; ========================================================================= +;; Test 9: File pool — append +;; ========================================================================= + +(test "filepool: append" + (let ([rt (make-fiber-runtime 2)] + [result-box (box #f)] + [path "/tmp/jerboa-test-filepool-append.txt"]) + (with-file-pool fpool + (fiber-spawn rt + (lambda () + (fiber-write-file path "line1\n" fpool) + (fiber-append-file path "line2\n" fpool) + (let ([content (fiber-read-file path fpool)]) + (set-box! result-box content))) + "append-fiber") + (fiber-runtime-run! rt)) + (assert-equal (unbox result-box) "line1\nline2\n" "append worked") + (delete-file path))) + +;; ========================================================================= +;; Test 10: File pool — file-exists? +;; ========================================================================= + +(test "filepool: file-exists?" + (let ([rt (make-fiber-runtime 2)] + [exists-box (box #f)] + [not-exists-box (box #t)]) + (with-file-pool fpool + (fiber-spawn rt + (lambda () + (set-box! exists-box (fiber-file-exists? "/tmp" fpool)) + (set-box! not-exists-box + (fiber-file-exists? "/tmp/no-such-file-jerboa-test-xxx" fpool))) + "exists-fiber") + (fiber-runtime-run! rt)) + (assert-true (unbox exists-box) "/tmp exists") + (assert-true (not (unbox not-exists-box)) "nonexistent file"))) + +;; ========================================================================= +;; Test 11: File pool — concurrent file operations +;; ========================================================================= + +(test "filepool: 10 concurrent reads/writes" + (let ([rt (make-fiber-runtime 4)] + [results (make-vector 10 #f)]) + (with-file-pool fpool + (do ([i 0 (+ i 1)]) + ((= i 10)) + (let ([idx i] + [path (string-append "/tmp/jerboa-test-concurrent-" (number->string i) ".txt")]) + (fiber-spawn rt + (lambda () + (let ([msg (string-append "fiber-" (number->string idx))]) + (fiber-write-file path msg fpool) + (let ([content (fiber-read-file path fpool)]) + (vector-set! results idx (string=? content msg))))) + (string-append "file-" (number->string idx))))) + (fiber-runtime-run! rt)) + ;; Verify and clean up + (do ([i 0 (+ i 1)]) + ((= i 10)) + (assert-true (vector-ref results i) + (string-append "file " (number->string i))) + (delete-file (string-append "/tmp/jerboa-test-concurrent-" (number->string i) ".txt"))))) + +;; ========================================================================= +;; Summary +;; ========================================================================= +(newline) +(display "=========================================") (newline) +(display "Results: ") (display pass-count) (display "/") +(display test-count) (display " passed") (newline) +(display "=========================================") (newline) +(when (< pass-count test-count) + (exit 1))