The (aux cml) module
Events, threads, synchronous variables, the library and I/O
Events#
Concurrent ML: events, sync and select
A port of SML/NJ CML (John Reppy) to CHICKEN on first-class continuations, see the header of aux.cml.scm for the design.
test/version: pass
(define (test/version _)
(⊦= "Concurrent ML (aux cml)" (cdr (assq 'system cml/version)))
(⊦= '(1 0 10) (cdr (assq 'version-id cml/version)))
(⊦= "Concurrent ML (aux cml), Version 1.0.10, September 15, 1997"
cml/banner))
((eta 0.0) (memory #(12582912 5524048 1048576)) (stdout "") (stderr ""))
test/run-status: pass
(define (test/run-status _)
(⊦= 'failure (run-cml (τ (sync never-evt))))
(⊦= 'failure (run-cml (τ (void))))
(⊦= 'done (run-cml (τ (cml/shutdown 'done))))
(⊦= 'success (run-cml (τ (cml/shutdown))))
(⊦= 'late
(run-cml
(τ (spawn (τ (cml/sleep 0.01) (cml/shutdown 'late)))
(sync never-evt))))
(⊭ (cml-running?))
(⊦= '(#t) (cdr (run/log (τ (note! (cml-running?)) (cml/shutdown))))))
((eta 0.001) (memory #(12582912 5529232 1048576)) (stdout "") (stderr ""))
test/not-running: pass
(define (test/not-running _)
(⊦raises (exn cml not-running) (sync (always-evt 1)))
(⊦raises (exn cml not-running) (spawn void))
(⊦raises (exn cml not-running) (recv (make-channel)))
(⊦raises (exn cml not-running) (cml/shutdown))
(⊦= '(raised)
(cdr (run/log
(τ (note! (raised-or (exn cml running) (run-cml void)))
(cml/shutdown)))))
(let1 (ch (make-channel))
(⊦raises (exn cml not-running) (recv ch))
(⊦raises (exn cml not-running) (send ch 42))
(⊦raises (exn cml not-running) (send-poll ch 1))
(⊦raises (exn cml not-running) (recv-poll ch)))
(let ((iv (make-ivar)) (mv (make-mvar)) (mb (make-mailbox)))
(⊦raises (exn cml not-running) (ivar-get iv))
(ivar-put! iv 1)
(⊦= '(1) (ivar-get-poll iv))
(⊦raises (exn cml not-running) (mvar-take! mv))
(⊦raises (exn cml not-running) (mvar-get mv))
(⊦raises (exn cml not-running) (mvar-swap! mv 2))
(mvar-put! mv 3)
(⊦= '(3) (mvar-get-poll mv))
(⊦= '(3) (mvar-take-poll mv))
(⊦= '() (mvar-take-poll mv))
(⊦raises (exn cml not-running) (mailbox-recv mb))
(⊦raises (exn cml not-running) (mailbox-send! mb 4))
(⊦= '() (mailbox-recv-poll mb)))
(⊦= 'ok (run/value (τ 'ok))))
((eta 0.0) (memory #(12582912 5533520 1048576)) (stdout "") (stderr ""))
test/run-cml/bad-arguments: pass
(define (test/run-cml/bad-arguments _)
(⊦raises (exn) (run-cml void #:quantum 0))
(⊦raises (exn) (run-cml void #:quantum -3))
(⊦raises (exn) (run-cml void #:quantum 1.5))
(⊦raises (exn) (run-cml 'not-a-thunk))
(⊭ (cml-running?))
(⊦= 'ok (run/value (τ 'ok))))
((eta 0.0) (memory #(12582912 5541760 1048576)) (stdout "") (stderr ""))
test/run-cml/escape: pass
(define (test/run-cml/escape _)
(⊦= 'escaped
(call-with-current-continuation
(λ (k) (run-cml (τ (spawn (τ (k 'escaped))) (sync never-evt))))))
(⊭ (cml-running?))
(⊦= 'ok (run/value (τ 'ok))))
((eta 0.0) (memory #(12582912 5534472 1048576)) (stdout "") (stderr ""))
test/internal-names: pass
(define (test/internal-names _)
(define (bound? name)
(condition-case (begin (eval name) #t) (ignored () #f)))
(⊦= '(#f #f #f #f #t #t #t #t)
(map bound?
'(atomic-begin
atomic-end
atomic-dispatch
dispatch
%atomic-begin
%atomic-end
%atomic-dispatch
%dispatch))))
((eta 0.001) (memory #(12582912 5534160 1048576)) (stdout "") (stderr ""))
test/rendezvous/receiver-first: pass
(define (test/rendezvous/receiver-first _)
(⊦= '((got x) (state non-atomic) sent)
(cdr (run/log
(τ (let1 (ch (make-channel))
(spawn (τ (note! (list 'got (recv ch)))
(note! (list 'state %atomic-state))))
(send ch 'x)
(note! 'sent)
(cml/yield)
(cml/shutdown)))))))
((eta 0.0) (memory #(12582912 5535328 1048576)) (stdout "") (stderr ""))
test/rendezvous/sender-first: pass
(define (test/rendezvous/sender-first _)
(⊦= '((got y) sent)
(cdr (run/log
(τ (let* ((ch (make-channel))
(t (spawn (τ (send ch 'y) (note! 'sent)))))
(note! (list 'got (recv ch)))
(sync (join-evt t))
(cml/shutdown)))))))
((eta 0.0) (memory #(12582912 5536680 1048576)) (stdout "") (stderr ""))
test/rendezvous/events: pass
(define (test/rendezvous/events _)
(⊦= '(1 2 3 (sent-evt))
(cdr (run/log
(τ (let1 (ch (make-channel))
(spawn (τ (send ch 1)
(sync (send-evt ch 2))
(sync (send-evt ch 3))
(note! '(sent-evt))))
(note! (sync (recv-evt ch)))
(note! (recv ch))
(cml/yield)
(note! (sync (recv-evt ch)))
(cml/yield)
(cml/shutdown)))))))
((eta 0.0) (memory #(12582912 5537632 1048576)) (stdout "") (stderr ""))
test/polls: pass
(define (test/polls _)
(⊦= '(#f () got-a #t (b) #f)
(cdr (run/log
(τ (let1 (ch (make-channel))
(note! (send-poll ch 'a))
(note! (recv-poll ch))
(spawn (τ (recv ch) (note! 'got-a)))
(note! (send-poll ch 'a))
(spawn (τ (send ch 'b)))
(note! (recv-poll ch))
(note! (send-poll ch 'c))
(cml/shutdown)))))))
((eta 0.0) (memory #(12582912 5539176 1048576)) (stdout "") (stderr ""))
test/polls/tick: pass
(define (test/polls/tick _)
(define (spins poll make put!)
(run/value
(τ (let1 (v (make))
(spawn (τ (cml/yield) (put! v 'done)))
(let loop ((n 0))
(cond ((pair? (poll v)) 'filled)
((< n 100000) (loop (add1 n)))
(else 'starved)))))))
(⊦= 'filled (spins ivar-get-poll make-ivar ivar-put!))
(⊦= 'filled (spins mvar-get-poll make-mvar mvar-put!))
(⊦= 'filled (spins mvar-take-poll make-mvar mvar-put!))
(⊦= 'filled (spins mailbox-recv-poll make-mailbox mailbox-send!))
(let1 (iv (make-ivar))
(⊦= '() (ivar-get-poll iv))
(ivar-put! iv 1)
(⊦= '(1) (ivar-get-poll iv))
(⊦= '(2) (mvar-get-poll (make-mvar 2)))))
((eta 0.0) (memory #(12582912 5538680 1048576)) (stdout "") (stderr ""))
test/choose/many-channels: pass
(define (test/choose/many-channels _)
(let* ((n 12)
(got (run/value
(τ (let1 (chs (map (λ (i) (make-channel)) (iota n)))
(for-each
(λ (ch i) (spawn (τ (send ch i))))
chs
(iota n))
(let loop ((k n) (acc '()))
(if (zero? k)
acc
(loop (sub1 k)
(cons (select* (map recv-evt chs)) acc)))))))))
(⊦= (iota n) (sort got <))))
((eta 0.0) (memory #(12582912 5543608 1048576)) (stdout "") (stderr ""))
test/choose/structure: pass
(define (test/choose/structure _)
(let* ((c (make-channel))
(e (choose
(always-evt 1)
(choose (recv-evt c) never-evt)
(guard (τ never-evt)))))
(⊨ (event? e))
(⊨ (event? never-evt))
(⊭ (event? 42))
(⊦= 'choose (%event-tag e))
(⊦= 'base (%event-tag (choose (always-evt 1) (recv-evt c) never-evt)))
(⊦= 2
(length
(%event-payload (choose (always-evt 1) (recv-evt c) never-evt))))
(⊦= 7 (run/value (τ (sync (wrap (choose never-evt (always-evt 6)) add1)))))
(⊦= 'failure (run-cml (τ (sync (choose)))))
(⊦= 'base (%event-tag (choose* '())))
(⊦= 2 (run/value (τ (sync (choose* (list never-evt (always-evt 2)))))))
(⊦= 'failure (run-cml (τ (sync (choose* '())))))))
((eta 0.0) (memory #(12582912 5545424 1048576)) (stdout "") (stderr ""))
test/select/case: pass
(define (test/select/case _)
(⊦= '(one 3 unit (4 5) none)
(cdr (run/log
(τ (let ((c (make-channel)) (d (make-channel)))
(spawn (τ (send c 1)))
(note! (select/case
((recv-evt c) (x) (if (= x 1) 'one 'other))
((recv-evt d) (x) 'from-d)))
(note! (select/case ((always-evt 2) => add1)))
(note! (select/case ((timeout-evt 0) () 'unit)))
(note! (select/case ((always-evt 4) xs (cons 4 (list 5)))))
(note! (sync/timeout (recv-evt d) 0.01 'none))
(cml/shutdown)))))))
((eta 0.0) (memory #(12582912 5544632 1048576)) (stdout "") (stderr ""))
test/guard/per-sync: pass
(define (test/guard/per-sync _)
(⊦= '(1 2 3 3)
(cdr (run/log
(τ (let* ((n 0) (e (guard (τ (set! n (add1 n)) (always-evt n)))))
(note! (sync e))
(note! (sync e))
(note! (select e never-evt))
(note! n)
(cml/shutdown)))))))
((eta 0.0) (memory #(12582912 5543304 1048576)) (stdout "") (stderr ""))
test/guard/exceptions-not-handled: pass
(define (test/guard/exceptions-not-handled _)
(⊦= '(raised handled)
(cdr (run/log
(τ (note! (raised-or
(exn)
(sync (wrap-handler
(guard (τ (error "in guard")))
(λ (e) 'handled)))))
(note! (sync (wrap-handler
(guard (τ (wrap (always-evt 0) (λ (x) (error "in wrap")))))
(λ (e) 'handled))))
(cml/shutdown))))))
((eta 0.0) (memory #(12582912 5550128 1048576)) (stdout "") (stderr ""))
test/guard/not-an-event: pass
(define (test/guard/not-an-event _)
(define (message thunk)
(condition-case
(begin (thunk) 'not-raised)
(e (exn) (get-condition-property e 'exn 'message))))
(⊦= '("cml: guard did not return an event"
"cml: guard did not return an event"
"cml: guard did not return an event"
"cml: with-nack did not return an event"
"cml: with-nack did not return an event"
"not an event"
"not an event"
"not an event"
"not an event")
(run/value
(τ (map message
(list (τ (sync (guard (τ 42))))
(τ (sync (wrap (guard (τ 42)) add1)))
(τ (sync (wrap-handler (guard (τ 42)) (λ (e) 'handled))))
(τ (sync (with-nack (λ (n) 42))))
(τ (select (wrap (with-nack (λ (n) 42)) add1) never-evt))
(τ (choose 42 (always-evt 1)))
(τ (select (always-evt 1) 42))
(τ (wrap 42 add1))
(τ (wrap-handler 42 void))))))))
((eta 0.0) (memory #(12582912 5566328 1048576)) (stdout "") (stderr ""))
test/wrap-handler/nesting: pass
(define (test/wrap-handler/nesting _)
(define (boom x) (error "boom"))
(⊦= '(inner outer 1 raised (msg "boom"))
(cdr (run/log
(τ (note! (sync (wrap-handler
(wrap-handler
(wrap (always-evt 1) boom)
(λ (e) 'inner))
(λ (e) 'outer))))
(note! (sync (wrap-handler
(wrap-handler
(wrap (always-evt 1) boom)
(λ (e) (error "again")))
(λ (e) 'outer))))
(note! (sync (wrap-handler (always-evt 1) (λ (e) 'unused))))
(note! (raised-or
(exn)
(sync (wrap (wrap-handler (always-evt 1) (λ (e) 'no))
boom))))
(note! (sync (wrap-handler
(wrap (always-evt 1) boom)
(λ (e)
(list 'msg
((condition-property-accessor 'exn 'message) e))))))
(cml/shutdown))))))
((eta 0.0) (memory #(12582912 5554576 1048576)) (stdout "") (stderr ""))
test/wrap-handler/own-branch-only: pass
(define (test/wrap-handler/own-branch-only _)
(define (boom v) (error "boom"))
(define (handled e) 'c1-handler)
(⊦= '(raised raised raised raised)
(cdr (run/log
(τ (let ((c1 (make-channel)) (c2 (make-channel)))
(define (try thunk)
(spawn (τ (cml/yield) (send c2 'x)))
(note! (raised-or (exn) (thunk))))
(try (τ (select
(wrap-handler (recv-evt c1) handled)
(wrap (recv-evt c2) boom))))
(try (τ (sync (choose
(wrap-handler (recv-evt c1) handled)
(wrap (recv-evt c2) boom)))))
(try (τ (sync (choose
(wrap (recv-evt c2) boom)
(wrap-handler (recv-evt c1) handled)))))
(try (τ (select
(wrap-handler
(with-nack (λ (nack) (recv-evt c1)))
handled)
(wrap (recv-evt c2) boom))))
(cml/shutdown)))))))
((eta 0.0) (memory #(12582912 5563280 1048576)) (stdout "") (stderr ""))
test/with-nack/losers-only: pass
(define (test/with-nack/losers-only _)
(let1 (log (cdr (run/log
(τ (define (branch i)
(with-nack
(λ (nack)
(spawn (τ (sync nack) (note! (list 'nack i))))
never-evt)))
(note! (select
(branch 1)
(branch 2)
(always-evt 'win)
(branch 3)))
(cml/sleep 0.01)
(cml/shutdown)))))
(⊦= 'win (car log))
(⊦= '(1 2 3) (sort (map cadr (cdr log)) <))))
((eta 0.0) (memory #(12582912 5551144 1048576)) (stdout "") (stderr ""))
test/with-nack/chosen-no-nack: pass
(define (test/with-nack/chosen-no-nack _)
(⊦= '(a)
(cdr (run/log
(τ (note! (select
(with-nack
(λ (nack)
(spawn (τ (sync nack) (note! 'nacked)))
(always-evt 'a)))
never-evt))
(cml/sleep 0.01)
(cml/shutdown))))))
((eta 0.0) (memory #(12582912 5555640 1048576)) (stdout "") (stderr ""))
test/with-nack/blocked: pass
(define (test/with-nack/blocked _)
(⊦= '((got 2) nack #f)
(cdr (run/log
(τ (let ((c1 (make-channel)) (c2 (make-channel)))
(spawn (τ (note! (list 'got
(select
(with-nack
(λ (nack)
(spawn (τ (sync nack) (note! 'nack)))
(recv-evt c1)))
(recv-evt c2))))))
(send c2 2)
(cml/sleep 0.01)
(note! (send-poll c1 1))
(cml/shutdown)))))))
((eta 0.0) (memory #(12582912 5553488 1048576)) (stdout "") (stderr ""))
test/with-nack/nested: pass
(define (test/with-nack/nested _)
(⊦= '((got b) inner-a)
(cdr (run/log
(τ (let1 (c (make-channel))
(spawn (τ (send c 'b)))
(note! (list 'got
(select
(with-nack
(λ (outer)
(spawn (τ (sync outer) (note! 'outer)))
(choose
(with-nack
(λ (a)
(spawn (τ (sync a) (note! 'inner-a)))
never-evt))
(recv-evt c))))
never-evt)))
(cml/sleep 0.01)
(cml/shutdown)))))))
((eta 0.0) (memory #(12582912 5558248 1048576)) (stdout "") (stderr ""))
test/with-nack/forcing-raises: pass
(define (test/with-nack/forcing-raises _)
(define (branch tag)
(with-nack
(λ (nack) (spawn (τ (sync nack) (note! (list 'nacked tag)))) never-evt)))
(⊦= '(raised (nacked 1))
(cdr (run/log
(τ (note! (raised-or
(exn)
(select
(branch 1)
(guard (τ (error "guard failed"))))))
(cml/sleep 0.01)
(cml/shutdown)))))
(⊦= '(1 (nacked 2))
(cdr (run/log
(τ (note! (with-exception-handler
(λ (e) 'ignored)
(τ (select
(branch 2)
(guard (τ (signal 'warning) (always-evt 1)))))))
(cml/sleep 0.01)
(cml/shutdown))))))
((eta 0.001) (memory #(12582912 5555336 1048576)) (stdout "") (stderr ""))
test/with-nack/partner-waits-on-the-nack: pass
(define (test/with-nack/partner-waits-on-the-nack _)
(define (client other-evt server-evt)
(sync (choose
(with-nack
(λ (nack)
(spawn (τ (note! (list 'server
(select
(wrap server-evt (λ ignored 'done))
(wrap nack (λ ignored 'nacked)))
(tid->string (current-tid))))))
never-evt))
other-evt)))
(⊦= '((client hello "[000002]") (server done "[000003]"))
(cdr (run/log
(τ (let1 (ch (make-channel))
(note! (list 'client
(client (recv-evt ch) (send-evt ch 'hello))
(tid->string (current-tid))))
(cml/sleep 0.01)
(cml/shutdown))))))
(⊦= '((server done "[000003]") (client sent "[000002]"))
(cdr (run/log
(τ (let1 (ch (make-channel))
(note! (list 'client
(client
(wrap (send-evt ch 'hello) (λ ignored 'sent))
(recv-evt ch))
(tid->string (current-tid))))
(cml/sleep 0.01)
(cml/shutdown))))))
(⊦= '((client 1 "[000002]") (server done "[000003]"))
(cdr (run/log
(τ (let* ((b (make-barrier 0 add1))
(e1 (barrier-enroll b))
(e2 (barrier-enroll b)))
(note! (list 'client
(client
(barrier-wait-evt e1)
(barrier-wait-evt e2))
(tid->string (current-tid))))
(cml/sleep 0.01)
(cml/shutdown)))))))
((eta 0.0) (memory #(12582912 5559144 1048576)) (stdout "") (stderr ""))
test/with-nack/forcing-escapes: pass
(define (test/with-nack/forcing-escapes _)
(define (branch tag)
(with-nack
(λ (nack) (spawn (τ (sync nack) (note! (list 'nacked tag)))) never-evt)))
(⊦= '(escaped (nacked 1))
(cdr (run/log
(τ (note! (call-with-current-continuation
(λ (k) (select (branch 1) (guard (τ (k 'escaped)))))))
(cml/sleep 0.01)
(cml/shutdown)))))
(⊦= '((nacked 2))
(cdr (run/log
(τ (spawn (τ (select (branch 2) (guard (τ (cml/exit))))))
(cml/sleep 0.01)
(cml/shutdown)))))
(⊦= '(x (nacked 3))
(cdr (run/log
(τ (let1 (ch (make-channel))
(spawn (τ (cml/sleep 0.01) (send ch 'x)))
(note! (select
(branch 3)
(guard (τ (always-evt (recv ch))))))
(cml/sleep 0.01)
(cml/shutdown)))))))
((eta 0.001) (memory #(12582912 5560920 1048576)) (stdout "") (stderr ""))
test/with-nack/forcing-escapes-then-reentered: pass
(define (test/with-nack/forcing-escapes-then-reentered _)
(let* ((saved #f)
(first? #t)
(r (run/log
(τ (spawn (τ (let loop ((i 0))
(cml/sleep 0.02)
(note! 'tick)
(when (< i 3) (loop (add1 i))))))
(let1 (r (call-with-current-continuation
(λ (out)
(select
(with-nack
(λ (n)
(spawn (τ (sync n) (note! 'nacked)))
never-evt))
(guard (τ (call-with-current-continuation
(λ (k) (set! saved k)))
(if first?
(begin (set! first? #f) (out 'escaped))
(wrap (timeout-evt 0.01) (λ ignored 'timeout)))))))))
(note! r)
(if (eq? r 'escaped)
(saved #f)
(begin (cml/sleep 0.2) (cml/shutdown 'ok))))))))
(⊦= 'ok (car r))
(⊦= '(escaped nacked timeout) (filter (λ (x) (not (eq? x 'tick))) (cdr r)))
(⊦= 4 (count (λ (x) (eq? x 'tick)) (cdr r)))))
((eta 0.001) (memory #(12582912 5560840 1048576)) (stdout "") (stderr ""))
test/with-nack/forcing-thread-killed-while-switched-out: pass
(define (test/with-nack/forcing-thread-killed-while-switched-out _)
(define (killed-in-guard)
(select
(with-nack (λ (n) (spawn (τ (sync n) (note! '(nacked 1)))) never-evt))
(guard (τ (dynamic-wind
void
(τ (cml/sleep 0.01) never-evt)
(τ (error "after thunk raised")))))))
(⊦= '(died (nacked 1) joined)
(cdr (run/log
(τ (sync (join-evt
(parameterize
((default-exn-handler (λ (e) (note! 'died))))
(spawn killed-in-guard))))
(note! 'joined)
(cml/sleep 0.05)
(cml/shutdown))))))
((eta 0.0) (memory #(12582912 5564224 1048576)) (stdout "") (stderr ""))
test/with-nack/set-when-sync-returns: pass
(define (test/with-nack/set-when-sync-returns _)
(⊦= '(other nacked)
(cdr (run/log
(τ (let1 (nack #f)
(note! (select
(with-nack (λ (n) (set! nack n) never-evt))
(always-evt 'other)))
(note! (sync/timeout
(wrap nack (λ ignored 'nacked))
0
'not-yet))
(cml/shutdown))))))
(⊦= '((nacked 1) (nacked 2))
(filter
pair?
(cdr (run/log
(τ (define (branch i)
(with-nack
(λ (n)
(spawn (τ (sync n) (note! (list 'nacked i))))
never-evt)))
(note! (select (branch 1) (always-evt 'a)))
(note! (select (branch 2) (always-evt 'b)))
(cml/sleep 0.01)
(cml/shutdown))))))
(⊦= '(abandoned (served 2))
(cdr (run/log
(τ (let ((req (make-channel)) (go (make-mvar)) (nack #f))
(spawn (τ (mvar-take! go)
(note! (select
(wrap (recv-evt req) (λ (x) (list 'served x)))
(wrap nack (λ ignored 'abandoned))))
(note! (list 'served (recv req)))))
(cml/yield)
(select
(with-nack
(λ (n) (set! nack n) (mvar-put! go #t) never-evt))
(always-evt 'other))
(send req 2)
(cml/sleep 0.01)
(cml/shutdown)))))))
((eta 0.001) (memory #(12582912 5562400 1048576)) (stdout "") (stderr ""))
test/with-nack/preempted-handler: pass
(define (test/with-nack/preempted-handler _)
(define warning? (condition-predicate 'warning))
(define (handler mb)
(λ (c) (mailbox-send! mb 'log) (mailbox-recv mb) 'continue))
(⊦= '(x not-nacked)
(cdr (run/log
(τ (let ((mb (make-mailbox)) (nacked #f))
(note! (with-exception-handler
(handler mb)
(τ (sync (choose
(with-nack
(λ (n)
(spawn (τ (sync n) (set! nacked #t)))
(always-evt 'x)))
(guard (τ (signal (make-property-condition 'warning))
never-evt)))))))
(cml/yield)
(cml/yield)
(note! (if nacked 'nacked 'not-nacked))
(cml/shutdown)))
#:quantum
1)))
(⊦= '(msg)
(cdr (run/log
(τ (let ((mb (make-mailbox)) (ch (make-channel)))
(spawn (τ (cml/sleep 0.02) (send ch 'msg)))
(note! (condition-case
(with-exception-handler
(handler mb)
(τ (sync (choose
(with-nack (λ (n) never-evt))
(guard (τ (signal (make-property-condition 'warning))
(recv-evt ch)))))))
(e ()
(list 'raised
(get-condition-property e 'exn 'message)))))
(cml/shutdown)))
#:quantum
1))))
((eta 0.0) (memory #(12582912 5564232 1048576)) (stdout "") (stderr ""))
test/with-nack/poll-raises: pass
(define (test/with-nack/poll-raises _)
(⊦= '(raised (nacked 1))
(cdr (run/log
(τ (note! (raised-or
(exn)
(select
(with-nack
(λ (nack)
(spawn (τ (sync nack) (note! '(nacked 1))))
never-evt))
(recv-evt 'not-a-channel))))
(cml/sleep 0.01)
(cml/shutdown))))))
((eta 0.0) (memory #(12582912 5569904 1048576)) (stdout "") (stderr ""))
test/sync/continuable-conditions: pass
(define (test/sync/continuable-conditions _)
(define (try blocked? two?)
(run/value
(τ (let ((c (make-channel)) (c2 (make-channel)))
(spawn (τ (unless blocked? (send c 1))))
(if blocked? (spawn (τ (cml/sleep 0.01) (send c 1))) (cml/yield))
(call-with-current-continuation
(λ (out)
(with-exception-handler
(λ (e) (if (eq? e 'oops) 42 (out 'raised)))
(τ (let1 (e (wrap (recv-evt c) (λ (v) (+ v (signal 'oops)))))
(sync (if two? (choose e (recv-evt c2)) e)))))))))))
(⊦= '(43 43 43 43) (list (try #f #f) (try #f #t) (try #t #f) (try #t #t))))
((eta 0.0) (memory #(12582912 5566432 1048576)) (stdout "") (stderr ""))
test/wrap/loops-do-not-nest: pass
(define (test/wrap/loops-do-not-nest _)
(define (growth n make-evt)
(run/value
(τ (let1 (ch (make-channel))
(spawn (τ (let loop () (cml/yield) (send ch 1) (loop))))
(let loop ((k 0) (w0 #f))
(let1 (w (length ##sys#dynamic-winds))
(if (= k n)
(- w w0)
(sync (wrap (make-evt ch)
(λ ignored (loop (add1 k) (or w0 w))))))))))))
(⊦= 0 (growth 300 (λ (ch) (choose (recv-evt ch) (timeout-evt 10)))))
(⊦= 0 (growth 300 (λ (ch) (with-nack (λ (nack) (recv-evt ch))))))
(⊦= 0
(growth
300
(λ (ch)
(choose (with-nack (λ (nack) (recv-evt ch))) (timeout-evt 10))))))
((eta 0.008) (memory #(12582912 5569240 1048576)) (stdout "") (stderr ""))
test/wrap-handler/blocked-branch: pass
(define (test/wrap-handler/blocked-branch _)
(⊦= '((a raised) (b handled))
(cdr (run/log
(τ (let* ((b (make-barrier 0 (λ (x) (error "update failed"))))
(ea (barrier-enroll b))
(eb (barrier-enroll b)))
(spawn (τ (cml/sleep 0.01)
(note! (list 'a (raised-or (exn) (barrier-wait ea))))))
(note! (list 'b
(select
(wrap-handler
(wrap (barrier-wait-evt eb) (λ (x) 'wrapped))
(λ (e) 'handled))
(recv-evt (make-channel)))))
(cml/yield)
(cml/shutdown))))))
(⊦= '(3 5)
(run/value
(τ (let ((ch (make-channel)) (ch2 (make-channel)))
(spawn (τ (cml/sleep 0.01) (send ch 1)))
(list (select
(wrap (wrap (recv-evt ch) add1) add1)
(recv-evt ch2))
(begin
(spawn (τ (cml/sleep 0.01) (send ch 4)))
(sync (wrap (choose (recv-evt ch2) (recv-evt ch)) add1)))))))))
((eta 0.001) (memory #(12582912 5577528 1048576)) (stdout "") (stderr ""))
test/priorities/prio-0: pass
(define (test/priorities/prio-0 _)
(⊦= '(a a a 5 a 5 a a 5 a 5 a a 5 a 5 a a 5 a)
(run/value
(τ (let1 (mv (make-mvar 5))
(map (λ (i) (select (mvar-get-evt mv) (always-evt 'a)))
(iota 20)))))))
((eta 0.0) (memory #(12582912 5569648 1048576)) (stdout "") (stderr ""))
test/priorities/reset: pass
(define (test/priorities/reset _)
(⊦= '(1 1 1 1)
(run/value
(τ (let ((iv (make-ivar)) (mv (make-mvar 'x)))
(ivar-put! iv 'v)
(sync (ivar-get-evt iv))
(sync (ivar-get-evt iv))
(sync (mvar-take-evt mv))
(mvar-put! mv 'y)
(sync (mvar-take-evt mv))
(mvar-put! mv 'z)
(let1 (take (%cell-priority mv))
(sync (mvar-get-evt mv))
(sync (mvar-get-evt mv))
(let1 (get (%cell-priority mv))
(sync (mvar-swap-evt mv 'w))
(sync (mvar-swap-evt mv 'w))
(list (%cell-priority iv)
take
get
(%cell-priority mv)))))))))
((eta 0.0) (memory #(12582912 5571056 1048576)) (stdout "") (stderr ""))
test/priorities/full-mvar-does-not-starve: pass
(define (test/priorities/full-mvar-does-not-starve _)
(define (wins mvar-evt)
(run/value
(τ (let ((mv (make-mvar 1)) (iv (make-ivar)) (m 0))
(ivar-put! iv 2)
(do ((i 0 (add1 i)))
((= i 100) m)
(select
(wrap (mvar-evt mv)
(λ ignored
(set! m (add1 m))
(when (null? (mvar-get-poll mv)) (mvar-put! mv 1))))
(ivar-get-evt iv)))))))
(⊦= '(50 50 50)
(map wins
(list mvar-get-evt (λ (mv) (mvar-swap-evt mv 1)) mvar-take-evt))))
((eta 0.001) (memory #(12582912 5571728 1048576)) (stdout "") (stderr ""))
test/priorities/tie-break: pass
(define (test/priorities/tie-break _)
(let1 (got (run/value
(τ (map (λ (i)
(select
(always-evt 'a)
(always-evt 'b)
(always-evt 'c)))
(iota 30)))))
(⊦= '(10 10 10) (map (λ (x) (count (λ (y) (eq? x y)) got)) '(a b c)))))
((eta 0.0) (memory #(12582912 5574256 1048576)) (stdout "") (stderr ""))
Threads#
test/tid: pass
(define (test/tid _)
(⊦= "[-000001]" (tid->string (current-tid)))
(let1 (r (run/value
(τ (let* ((me (current-tid)) (t (spawn void)))
(list (tid? me)
(tid? 1)
(tid=? me me)
(tid=? me t)
(tid<? me t)
(tid<? t me)
(tid<? me me)
(tid-compare t me)
(tid-compare me me)
(tid-compare me t)
(= (add1 (tid-hash me)) (tid-hash t))
(string=? (tid->string me) (tid->string me))
(string-length (tid->string me)))))))
(⊦= '(#t #f #t #f #t #f #f 1 0 -1 #t #t 8) r)))
((eta 0.0) (memory #(12582912 5536056 1048576)) (stdout "") (stderr ""))
test/runs-are-independent: pass
(define (test/runs-are-independent _)
(let ((a (run/value
(τ (list (tid->string (current-tid))
(tid->string (spawn void))))))
(b (run/value
(τ (list (tid->string (current-tid))
(tid->string (spawn void)))))))
(⊦= a b)
(⊦= "[000002]" (car a))))
((eta 0.0) (memory #(12582912 5537040 1048576)) (stdout "") (stderr ""))
test/spawn/child-first: pass
(define (test/spawn/child-first _)
(⊦= '(child parent (arg 42))
(cdr (run/log
(τ (spawn (τ (note! 'child)))
(note! 'parent)
(spawn/call (λ (x) (note! (list 'arg x))) 42)
(cml/shutdown))))))
((eta 0.0) (memory #(12582912 5530288 1048576)) (stdout "") (stderr ""))
test/spawn/not-a-procedure: pass
(define (test/spawn/not-a-procedure _)
(⊦= '(raised raised)
(run/value
(τ (map (λ (thunk)
(condition-case
(begin (thunk) (cml/yield) 'no-raise)
(e (exn) 'raised)))
(list (τ (spawn 42)) (τ (spawn/call 42 1))))))))
((eta 0.0) (memory #(12582912 5538960 1048576)) (stdout "") (stderr ""))
test/join/normal-exit-exception: pass
(define (test/join/normal-exit-exception _)
(parameterize
((default-exn-handler (λ (e) (note! 'handler))))
(⊦= '(handler joined-normal joined-exit joined-exn)
(cdr (run/log
(τ (let ((t1 (spawn (τ (cml/sleep 0.01))))
(t2 (spawn (τ (cml/sleep 0.01) (cml/exit) (note! 'not-here))))
(t3 (spawn (τ (error "die")))))
(sync (join-evt t1))
(note! 'joined-normal)
(sync (join-evt t2))
(note! 'joined-exit)
(sync (join-evt t3))
(note! 'joined-exn)
(sync (join-evt t3))
(cml/shutdown))))))))
((eta 0.0) (memory #(12582912 5533736 1048576)) (stdout "") (stderr ""))
test/join/fifo: pass
(define (test/join/fifo _)
(⊦= '(w1 w2 w3 w4)
(cdr (run/log
(τ (let* ((c (make-channel)) (t (spawn (τ (recv c)))))
(for-each
(λ (w) (spawn (τ (sync (join-evt t)) (note! w))))
'(w1 w2 w3 w4))
(send c 'go)
(cml/sleep 0.01)
(cml/shutdown)))))))
((eta 0.0) (memory #(12582912 5534792 1048576)) (stdout "") (stderr ""))
test/default-exn-handler: pass
(define (test/default-exn-handler _)
(let1 (err (with-error-output-to-string
(τ (⊦= '(after)
(cdr (run/log
(τ (spawn (τ (error "kaboom" 1 2)))
(note! 'after)
(cml/shutdown))))))))
(⊨ (and (substring-index "kaboom" err) #t))
(⊨ (and (substring-index "[000003]" err) #t))
(⊦= err
(with-error-output-to-string
(τ (run-cml
(τ (uncaught-default-handler
(spawn void)
(make-property-condition
'exn
'message
"kaboom"
'arguments
'(1 2)))
(cml/shutdown)))))))
(parameterize
((default-exn-handler (λ (e) (error "handler fails too"))))
(⊦= '(still-running)
(cdr (run/log
(τ (spawn (τ (car '())))
(note! 'still-running)
(cml/shutdown)))))))
((eta 0.0) (memory #(12582912 5540312 1048576)) (stdout "") (stderr ""))
test/dynamic-wind/after-raises: pass
(define (test/dynamic-wind/after-raises _)
(parameterize
((default-exn-handler
(λ (e) (note! (list 'died (tid->string (current-tid)))))))
(⊦= '(done (died "[000003]") joined)
(run/log
(τ (let1 (a (spawn (τ (dynamic-wind
void
(τ (cml/sleep 0.02) (note! 'resumed-after-death))
(τ (error "after"))))))
(sync (join-evt a))
(note! 'joined)
(cml/sleep 0.05)
(cml/shutdown 'done)))))
(⊦= '(bye (died "[000003]"))
(run/log
(τ (spawn (τ (dynamic-wind
void
(τ (cml/shutdown 'bye))
(τ (error "after")))))
(sync never-evt))))))
((eta 0.0) (memory #(12582912 5542152 1048576)) (stdout "") (stderr ""))
test/dynamic-wind/winders-at-every-switch: pass
(define (test/dynamic-wind/winders-at-every-switch _)
(define (who) (tid->string (current-tid)))
(⊦= '((in "[000003]")
(out "[-000001]")
(in "[000003]")
(body x)
(out "[000003]"))
(cdr (run/log
(τ (let1 (c (make-channel))
(let1 (a (spawn (τ (dynamic-wind
(τ (note! (list 'in (who))))
(τ (note! (list 'body (recv c))))
(τ (note! (list 'out (who))))))))
(send c 'x)
(sync (join-evt a))
(cml/shutdown))))))))
((eta 0.0) (memory #(12582912 5538352 1048576)) (stdout "") (stderr ""))
test/dynamic-wind/dead-receiver-not-resumed: pass
(define (test/dynamic-wind/dead-receiver-not-resumed _)
(define (try do-recv)
(parameterize
((default-exn-handler (λ (e) (note! 'died))))
(cdr (run/log
(τ (let ((ch (make-channel)) (armed #f))
(spawn (τ (send ch 'hello) (note! 'sender-resumed)))
(let1 (r (spawn (τ (dynamic-wind
void
(τ (set! armed #t)
(note! (list 'zombie (do-recv ch))))
(τ (when armed
(set! armed #f)
(error "after thunk raised")))))))
(sync (join-evt r))
(note! 'joined)
(cml/sleep 0.01)
(cml/shutdown))))))))
(⊦= '(died sender-resumed joined) (try recv))
(⊦= '(died sender-resumed joined) (try (λ (ch) (sync (recv-evt ch)))))
(⊦= '(died sender-resumed joined) (try (λ (ch) (car (recv-poll ch))))))
((eta 0.001) (memory #(12582912 5542016 1048576)) (stdout "") (stderr ""))
test/dynamic-wind/dead-sender-does-not-strand-receiver: pass
(define (test/dynamic-wind/dead-sender-does-not-strand-receiver _)
(define (try do-recv)
(parameterize
((default-exn-handler (λ (e) (note! 'died))))
(run/log
(τ (let ((ch (make-channel)) (first #t))
(spawn (τ (dynamic-wind
(τ (if first
(set! first #f)
(error "before thunk raised")))
(τ (send ch 42) (note! 'zombie))
void)))
(spawn (τ (cml/sleep 0.02) (send ch 43) (note! 'second-sent)))
(cml/yield)
(note! (list 'got (do-recv ch)))
(cml/sleep 0.03)
(cml/shutdown 'ok))))))
(⊦= '(ok died (got 43) second-sent) (try recv))
(⊦= '(ok died (got 43) second-sent) (try (λ (ch) (sync (recv-evt ch)))))
(⊦= '(ok died (got 43) second-sent)
(try (λ (ch) (sync (choose (recv-evt ch) (recv-evt (make-channel)))))))
(⊦= '(ok died (got ())) (try recv-poll)))
((eta 0.001) (memory #(12582912 5542848 1048576)) (stdout "") (stderr ""))
test/dynamic-wind/dead-reader-does-not-strand-syncvar: pass
(define (test/dynamic-wind/dead-reader-does-not-strand-syncvar _)
(define (try make read)
(parameterize
((default-exn-handler (λ (e) (note! 'died))))
(cdr (run/log
(τ (let ((v (make)) (n 0))
(spawn (τ (dynamic-wind
(τ (set! n (add1 n))
(when (= n 2) (error "before thunk raised")))
(τ (note! (list 'zombie (read v))))
void)))
(spawn (τ (note! (list 'second (read v)))))
(cml/sleep 0.01)
(if (ivar? v) (ivar-put! v 42) (mvar-put! v 42))
(note! (sync/timeout
(wrap (timeout-evt 10) (λ ignored 'never))
0.2
'stranded))
(note! (if (ivar? v) (ivar-get-poll v) (mvar-get-poll v)))
(cml/shutdown)))))))
(⊦= '(died (second 42) stranded (42)) (try make-ivar ivar-get))
(⊦= '(died (second 42) stranded (42))
(try make-ivar (λ (v) (sync (ivar-get-evt v)))))
(⊦= '(died (second 42) stranded (42)) (try make-mvar mvar-get))
(⊦= '(died (second 42) stranded (42))
(try make-mvar (λ (v) (sync (mvar-get-evt v)))))
(⊦= '(died (second 42) stranded ()) (try make-mvar mvar-take!))
(⊦= '(died (second 42) stranded ())
(try make-mvar (λ (v) (sync (choose (mvar-take-evt v) never-evt))))))
((eta 0.002) (memory #(12582912 5542000 1048576)) (stdout "") (stderr ""))
test/dynamic-wind/dead-reader-relays-the-current-value: pass
(define (test/dynamic-wind/dead-reader-relays-the-current-value _)
(define (try take)
(parameterize
((default-exn-handler (λ (e) (note! 'died))))
(cdr (run/log
(τ (let ((v (make-mvar)) (n 0))
(spawn (τ (dynamic-wind
(τ (set! n (add1 n))
(when (= n 2) (error "before thunk raised")))
(τ (note! (list 'zombie (take v))))
void)))
(spawn (τ (note! (list 'took (list (take v))))))
(cml/sleep 0.01)
(mvar-put! v 42)
(let1 (p (mvar-take-poll v))
(note! (list 'took p))
(mvar-put! v 43))
(cml/sleep 0.02)
(note! (list 'took (mvar-take-poll v)))
(cml/shutdown)))))))
(for-each
(λ (take)
(let1 (log (try take))
(⊦= 'died (car log))
(⊦= '(42 43) (sort (append-map cadr (cdr log)) <))))
(list mvar-take! (λ (v) (sync (mvar-take-evt v))))))
((eta 0.0) (memory #(12582912 5544304 1048576)) (stdout "") (stderr ""))
test/dynamic-wind/dead-receiver-does-not-lose-mailbox-message: pass
(define (test/dynamic-wind/dead-receiver-does-not-lose-mailbox-message _)
(define (try recv2)
(parameterize
((default-exn-handler (λ (e) (note! 'died))))
(cdr (run/log
(τ (let ((mb (make-mailbox)) (n 0))
(spawn (τ (dynamic-wind
(τ (set! n (add1 n))
(when (= n 2) (error "before thunk raised")))
(τ (note! (list 'zombie (mailbox-recv mb))))
void)))
(when recv2 (spawn (τ (note! (list 'second (recv2 mb))))))
(cml/sleep 0.01)
(mailbox-send! mb 42)
(mailbox-send! mb 43)
(cml/sleep 0.02)
(note! (list 'poll (mailbox-recv-poll mb)))
(cml/shutdown)))))))
(⊦= '(died (second 42) (poll (43))) (try mailbox-recv))
(⊦= '(died (second 42) (poll (43)))
(try (λ (mb) (sync (choose (mailbox-recv-evt mb) never-evt)))))
(⊦= '(died (poll (42))) (try #f)))
((eta 0.001) (memory #(12582912 5546320 1048576)) (stdout "") (stderr ""))
test/mailbox/receiver-dying-later-keeps-its-message: pass
(define (test/mailbox/receiver-dying-later-keeps-its-message _)
(define (try recv)
(parameterize
((default-exn-handler (λ (e) (note! 'died))))
(cdr (run/log
(τ (let1 (mb (make-mailbox))
(spawn (τ (let1 (j (recv mb))
(note! (list 'w1 j))
(error "worker 1 crashes on" j))))
(spawn (τ (let loop () (note! (list 'w2 (recv mb))) (loop))))
(cml/yield)
(mailbox-send! mb 'job-1)
(cml/sleep 0.02)
(note! (list 'poll (mailbox-recv-poll mb)))
(mailbox-send! mb 'job-2)
(cml/sleep 0.02)
(cml/shutdown)))))))
(⊦= '((w1 job-1) died (poll ()) (w2 job-2)) (try mailbox-recv))
(⊦= '((w1 job-1) died (poll ()) (w2 job-2))
(try (λ (mb) (sync (mailbox-recv-evt mb)))))
(⊦= '((w1 job-1) died (poll ()) (w2 job-2))
(try (λ (mb)
(sync (wrap (choose (mailbox-recv-evt mb) (timeout-evt 10))
(λ (j) j))))))
(⊦= '((w bad) died (w good))
(parameterize
((default-exn-handler (λ (e) (note! 'died))))
(cdr (run/log
(τ (let1 (mb (make-mailbox))
(for-each
(λ (i)
(spawn (τ (let loop ()
(select
(wrap (mailbox-recv-evt mb)
(λ (j)
(note! (list 'w j))
(when (eq? j 'bad) (error "bad job"))))
(timeout-evt 1))
(loop)))))
(iota 3))
(cml/yield)
(mailbox-send! mb 'bad)
(cml/sleep 0.02)
(mailbox-send! mb 'good)
(cml/sleep 0.02)
(cml/shutdown))))))))
((eta 0.001) (memory #(12582912 5550728 1048576)) (stdout "") (stderr ""))
test/dynamic-wind/dead-waiter-sets-its-nacks: pass
(define (test/dynamic-wind/dead-waiter-sets-its-nacks _)
(define (try make-evt mode)
(parameterize
((default-exn-handler (λ (e) (note! 'died))))
(cdr (run/log
(τ (let ((n 0) (nk #f))
(receive
(evt fire!)
(make-evt)
(spawn (τ (dynamic-wind
(τ (set! n (add1 n))
(when (and (eq? mode 'in) (= n 2))
(error "before")))
(τ (note! (list 'zombie
(sync (choose
evt
(with-nack (λ (nack) (set! nk nack) never-evt)))))))
(τ (when (and (eq? mode 'out) (= n 1))
(error "after"))))))
(cml/sleep 0.01)
(fire!)
(cml/sleep 0.01)
(note! (list 'nack
(sync/timeout
(wrap nk (λ ignored 'set))
0.2
'unset)))
(cml/shutdown))))))))
(define (ivar)
(let1 (iv (make-ivar)) (values (ivar-get-evt iv) (τ (ivar-put! iv 1)))))
(define (chan)
(let1 (ch (make-channel))
(values (recv-evt ch) (τ (spawn (τ (send ch 1)))))))
(define (mbox)
(let1 (mb (make-mailbox))
(values (mailbox-recv-evt mb) (τ (mailbox-send! mb 1)))))
(define (join)
(let1 (iv (make-ivar))
(values (join-evt (spawn (τ (ivar-get iv)))) (τ (ivar-put! iv 1)))))
(for-each
(λ (make-evt)
(⊦= '(died (nack set)) (try make-evt 'in))
(⊦= '(died (nack set)) (try make-evt 'out)))
(list ivar chan mbox join)))
((eta 0.002) (memory #(12582912 5549680 1048576)) (stdout "") (stderr ""))
test/dynamic-wind/handler-may-exit-or-block: pass
(define (test/dynamic-wind/handler-may-exit-or-block _)
(define (try handler)
(cdr (run/log
(τ (let* ((b (spawn (τ (cml/yield) (note! 'b-ran))))
(a (parameterize
((default-exn-handler handler))
(spawn (τ (dynamic-wind
void
(τ (cml/yield) (note! 'a-zombie))
(τ (error "after raised"))))))))
(note! (list 'joined
(sync/timeout
(wrap (join-evt b) (λ ignored 'b))
0.3
'never)))
(note! (list 'a-joined
(sync/timeout
(wrap (join-evt a) (λ ignored 'a))
0.3
'never)))
(cml/shutdown))))))
(⊦= '(a-died b-ran (joined b) (a-joined a)) (try (λ (e) (note! 'a-died))))
(⊦= '(a-died b-ran (joined b) (a-joined a))
(try (λ (e) (note! 'a-died) (cml/exit))))
(⊦= '(a-died b-ran (joined b) handler-woke (a-joined a))
(try (λ (e) (note! 'a-died) (cml/sleep 0.05) (note! 'handler-woke)))))
((eta 0.0) (memory #(12582912 5549744 1048576)) (stdout "") (stderr ""))
test/dynamic-wind/dead-thread-in-ready-queue: pass
(define (test/dynamic-wind/dead-thread-in-ready-queue _)
(define (try body)
(let1 (dead #f)
(parameterize
((default-exn-handler (λ (e) (set! dead #t) (note! 'died))))
(run/log
(τ (let1 (a (spawn (τ (let1 (armed #t)
(dynamic-wind
void
(τ (body (τ dead)))
(τ (when armed (set! armed #f) (error "after"))))))))
(sync (join-evt a))
(note! 'joined)
(cml/sleep 0.02)
(cml/shutdown 'done)))))))
(⊦= '(done died joined)
(try (λ (dead?) (cml/yield) (note! 'resumed-after-death))))
(⊦= '(done died joined)
(try (λ (dead?)
(let loop ()
(sync (always-evt 1))
(if (dead?) (note! 'resumed-after-death) (loop)))))))
((eta 0.001) (memory #(12582912 5551400 1048576)) (stdout "") (stderr ""))
test/yield: pass
(define (test/yield _)
(⊦= '(a1 b1 a2 b2 a3 b3)
(cdr (run/log
(τ (let1 (t (spawn (τ (for-each
(λ (x) (note! x) (cml/yield))
'(a1 a2 a3)))))
(for-each (λ (x) (note! x) (cml/yield)) '(b1 b2 b3))
(cml/shutdown)))))))
((eta 0.0) (memory #(12582912 5550576 1048576)) (stdout "") (stderr ""))
test/thread-properties: pass
(define (test/thread-properties _)
(⊦= '((main) () init (init) (child) () (main))
(cdr (run/log
(τ (receive
(clear! get peek put!)
(make-thread-property (τ 'init))
(put! 'main)
(note! (peek))
(sync (join-evt
(spawn (τ (note! (peek))
(note! (get))
(note! (peek))
(put! 'child)
(note! (peek))
(clear!)
(note! (peek))))))
(note! (peek))
(cml/shutdown)))))))
((eta 0.0) (memory #(12582912 5551904 1048576)) (stdout "") (stderr ""))
test/thread-property/lazy-init: pass
(define (test/thread-property/lazy-init _)
(⊦= '(() 1 1 (1) 2)
(cdr (run/log
(τ (let1 (n 0)
(receive
(clear! get peek put!)
(make-thread-property (τ (set! n (add1 n)) n))
(note! (peek))
(note! (get))
(note! (get))
(note! (peek))
(sync (join-evt (spawn (τ (note! (get))))))
(cml/shutdown))))))))
((eta 0.0) (memory #(12582912 5551192 1048576)) (stdout "") (stderr ""))
test/thread-flag: pass
(define (test/thread-flag _)
(⊦= '(#f #t (child #f #t) #t #f #t)
(cdr (run/log
(τ (receive
(get set)
(make-thread-flag)
(note! (get))
(set #t)
(set #t)
(note! (get))
(sync (join-evt
(spawn (τ (let1 (before (get))
(set #t)
(note! (list 'child before (get))))))))
(note! (get))
(set #f)
(note! (get))
(set #t)
(note! (get))
(cml/shutdown)))))))
((eta 0.0) (memory #(12582912 5552648 1048576)) (stdout "") (stderr ""))
test/fairness/two-producers: pass
(define (test/fairness/two-producers _)
(let1 (got (run/value
(τ (let1 (c (make-channel))
(for-each
(λ (tag)
(spawn (τ (let loop () (send c tag) (loop)))))
'(p q))
(map (λ (i) (recv c)) (iota 20))))))
(⊦= 10 (count (λ (x) (eq? x 'p)) got))
(⊦= (map (λ (i) (if (even? i) 'p 'q)) (iota 20)) got)))
((eta 0.0) (memory #(12582912 5554336 1048576)) (stdout "") (stderr ""))
test/fairness/preemption: pass
(define (test/fairness/preemption _)
(let1 (log (cdr (run/log
(τ (let1 (c (make-channel))
(define (worker tag)
(τ (let loop ((i 0))
(when (< i 400)
(recv-poll c)
(when (zero? (remainder i 100)) (note! tag))
(loop (add1 i))))))
(let ((a (spawn (worker 'a))) (b (spawn (worker 'b))))
(sync (join-evt a))
(sync (join-evt b))
(cml/shutdown))))
#:quantum
16)))
(⊦= 8 (length log))
(⊭ (equal? '(a a a a b b b b) log))))
((eta 0.001) (memory #(12582912 5555920 1048576)) (stdout "") (stderr ""))
test/fairness/yield-promotes: pass
(define (test/fairness/yield-promotes _)
(⊦= 'worker-done
(run-cml
(τ (let1 (done #f)
(spawn (τ (let loop ((i 0))
(if (= i 300)
(set! done #t)
(begin (sync (always-evt i)) (loop (add1 i)))))))
(let loop ((n 0))
(cond (done (cml/shutdown 'worker-done))
((< n 200000) (cml/yield) (loop (add1 n)))
(else (cml/shutdown 'starved))))))))
(let1 (sent (run/value
(τ (let ((mb (make-mailbox)) (sent 0))
(spawn (τ (let loop ()
(when (< sent 200000)
(mailbox-send! mb sent)
(set! sent (add1 sent))
(loop)))))
(let loop ((got 0))
(when (< got 2000) (mailbox-recv mb) (loop (add1 got))))
sent))))
(⊨ (< sent 3000))))
((eta 0.003) (memory #(12582912 5557720 1048576)) (stdout "") (stderr ""))
test/fairness/no-phase-lock: pass
(define (test/fairness/no-phase-lock _)
(let1 (r (run/value
(τ (let ((ch (make-channel)) (done 0))
(spawn (τ (let loop () (send ch 1) (loop))))
(let loop ((i 0))
(when (< i 5000)
(spawn (τ (set! done (add1 done))))
(recv ch)
(loop (add1 i))))
(list done
(length (%queue-front %rdy-q2))
(length (%queue-rear %rdy-q2)))))))
(⊨ (>= (car r) 4998))
(⊨ (<= (+ (cadr r) (caddr r)) 2))))
((eta 0.022) (memory #(12582912 5559168 1048576)) (stdout "") (stderr ""))
test/fairness/channel-ping-pong-ticks: pass
(define (test/fairness/channel-ping-pong-ticks _)
(define (ping-pong send! recv! server)
(run-cml
(τ (let ((a (make-channel)) (b (make-channel)) (n 0))
(spawn (τ (cml/sleep 0.01)
(cml/shutdown (list 'woke (< n 100000)))))
(spawn (τ (server a b)))
(let loop ()
(set! n (add1 n))
(when (> n 100000) (cml/shutdown 'starved))
(send! a 1)
(recv! b)
(loop))))))
(⊦= '(woke #t)
(ping-pong send recv (λ (a b) (let loop () (send b (recv a)) (loop)))))
(⊦= '(woke #t)
(ping-pong
(λ (ch v) (sync (send-evt ch v)))
(λ (ch) (sync (recv-evt ch)))
(λ (a b)
(let loop ()
(sync (send-evt b (sync (choose (recv-evt a) (recv-evt a)))))
(loop)))))
(⊦= '(timed-out #t)
(run-cml
(τ (let ((a (make-channel)) (b (make-channel)) (n 0))
(spawn (τ (cml/shutdown
(list (sync/timeout
(recv-evt (make-channel))
0.01
'timed-out)
(< n 100000)))))
(spawn (τ (let loop () (send b (recv a)) (loop))))
(let loop ()
(set! n (add1 n))
(when (> n 100000) (cml/shutdown 'starved))
(send a 1)
(recv b)
(loop))))))
(⊦= '(ran #t)
(run-cml
(τ (let ((a (make-channel)) (b (make-channel)) (n 0))
(spawn (τ (let loop () (send b (recv a)) (loop))))
(spawn (τ (let loop ((i 0))
(if (< i 2000)
(begin (sync (always-evt i)) (loop (add1 i)))
(cml/shutdown (list 'ran (< n 100000)))))))
(let loop ()
(set! n (add1 n))
(when (> n 100000) (cml/shutdown 'starved))
(send a 1)
(recv b)
(loop)))))))
((eta 0.035) (memory #(12582912 5560320 1048576)) (stdout "") (stderr ""))
Time#
test/timeouts/order: pass
(define (test/timeouts/order _)
(⊦= '(b c a)
(cdr (run/log
(τ (for-each
(λ (tag secs) (spawn (τ (cml/sleep secs) (note! tag))))
'(a b c)
'(0.06 0.02 0.04))
(cml/sleep 0.1)
(cml/shutdown))))))
((eta 0.001) (memory #(12582912 5524088 1048576)) (stdout "") (stderr ""))
test/timeouts/duration: pass
(define (test/timeouts/duration _)
(let1 (dt (run/value
(τ (let1 (t0 (cml/now)) (cml/sleep 0.05) (- (cml/now) t0)))))
(⊨ (>= dt 0.049))
(⊨ (< dt 1.0))))
((eta 0.0) (memory #(12582912 5525280 1048576)) (stdout "") (stderr ""))
test/timeouts/zero-and-past: pass
(define (test/timeouts/zero-and-past _)
(⊦= '(zero past later)
(cdr (run/log
(τ (sync (timeout-evt 0))
(note! 'zero)
(sync (at-time-evt (- (cml/now) 10)))
(note! 'past)
(sync (at-time-evt (+ (cml/now) 0.02)))
(note! 'later)
(cml/shutdown))))))
((eta 0.0) (memory #(12582912 5526208 1048576)) (stdout "") (stderr ""))
test/timeouts/lifo-ties: pass
(define (test/timeouts/lifo-ties _)
(⊦= '(second first)
(cdr (run/log
(τ (let1 (t (+ (cml/now) 0.03))
(spawn (τ (sync (at-time-evt t)) (note! 'first)))
(spawn (τ (sync (at-time-evt t)) (note! 'second)))
(cml/sleep 0.06)
(cml/shutdown)))))))
((eta 0.0) (memory #(12582912 5530176 1048576)) (stdout "") (stderr ""))
test/sync/timeout: pass
(define (test/sync/timeout _)
(⊦= '(none (got 7) timeout #f)
(cdr (run/log
(τ (let1 (c (make-channel))
(note! (sync/timeout (recv-evt c) 0.01 'none))
(spawn (τ (send c 7)))
(note! (list 'got (sync/timeout (recv-evt c) 1)))
(note! (select
(wrap (timeout-evt 0.01) (λ (x) 'timeout))
(recv-evt c)))
(note! (sync/timeout never-evt 0.01))
(cml/shutdown)))))))
((eta 0.0) (memory #(12582912 5536944 1048576)) (stdout "") (stderr ""))
test/sync/timeout/zero-is-a-poll: pass
(define (test/sync/timeout/zero-is-a-poll _)
(⊦= '(42 (1 1 1 1) (7 7 7) (x x) none none d)
(cdr (run/log
(τ (let ((ch (make-channel)) (iv (make-ivar)))
(spawn (τ (send ch 42)))
(cml/yield)
(note! (sync/timeout (recv-evt ch) 0 'none))
(note! (map (λ (i) (sync/timeout (always-evt 1) 0 'd))
(iota 4)))
(ivar-put! iv 7)
(note! (map (λ (i) (sync/timeout (ivar-get-evt iv) 0 'd))
(iota 3)))
(note! (map (λ (i)
(sync/timeout
(choose (recv-evt ch) (always-evt 'x))
-1
'd))
(iota 2)))
(note! (sync/timeout (recv-evt ch) 0 'none))
(note! (sync/timeout
(choose (recv-evt ch) never-evt)
0
'none))
(note! (sync/timeout (timeout-evt 0.5) 0 'd))
(cml/shutdown)))))))
((eta 0.0) (memory #(12582912 5533376 1048576)) (stdout "") (stderr ""))
test/timeouts/yield-loop: pass
(define (test/timeouts/yield-loop _)
(⊦= 'timer-fired
(run-cml
(τ (spawn (τ (let loop ((n 0))
(if (< n 1000000)
(begin (cml/yield) (loop (add1 n)))
(cml/shutdown 'starved)))))
(cml/sleep 0.02)
(cml/shutdown 'timer-fired))))
(⊦= 'filled
(run/value
(τ (let1 (iv (make-ivar))
(spawn (τ (cml/sleep 0.02) (ivar-put! iv 'filled)))
(let loop ((n 0))
(cond ((pair? (ivar-get-poll iv)) (car (ivar-get-poll iv)))
((< n 1000000) (cml/yield) (loop (add1 n)))
(else 'starved))))))))
((eta 0.039) (memory #(12582912 5534264 1048576)) (stdout "") (stderr ""))
test/timeouts/many-sleepers: pass
(define (test/timeouts/many-sleepers _)
(define (cost n)
(let1 (t0 (current-process-milliseconds))
(⊦= 'ok
(run/value
(τ (for-each
(λ (i) (spawn (τ (cml/sleep (+ 3600 i)))))
(iota n))
(let1 (ch (make-channel))
(spawn (τ (let loop () (send ch 1) (loop))))
(do ((i 0 (add1 i)))
((= i (* 4 n)) 'ok)
(recv ch))))))
(- (current-process-milliseconds) t0)))
(let* ((small (max 20 (cost 300))) (large (cost 3000)))
(⊨ (< large (* 30 small)))))
((eta 0.037) (memory #(12582912 5536144 1048576)) (stdout "") (stderr ""))
test/timeouts/losing-timeouts-are-dropped: pass
(define (test/timeouts/losing-timeouts-are-dropped _)
(let1 (n (run/value
(τ (let1 (ch (make-channel))
(spawn (τ (let loop () (cml/yield) (send ch 1) (loop))))
(do ((i 0 (add1 i)))
((= i 500))
(select (recv-evt ch) (timeout-evt 1000)))
%time-n))))
(⊨ (<= n (+ (* 2 %clean-budget-min) 2)))))
((eta 0.002) (memory #(12582912 5536920 1048576)) (stdout "") (stderr ""))
test/timeouts/cancelled-wait-not-woken: pass
(define (test/timeouts/cancelled-wait-not-woken _)
(⊦= '(got after)
(cdr (run/log
(τ (let1 (c (make-channel))
(spawn (τ (sync (choose
(wrap (timeout-evt 0.02)
(λ (x) (note! 'timed-out)))
(wrap (recv-evt c) (λ (x) (note! 'got)))))))
(send c 'x)
(cml/sleep 0.05)
(note! 'after)
(cml/shutdown)))))))
((eta 0.0) (memory #(12582912 5536208 1048576)) (stdout "") (stderr ""))
Synchronous variables#
test/ivar: pass
(define (test/ivar _)
(⊦= '(() (r3 1) (1) 1 1 (r1 1) (r2 1) raised)
(cdr (run/log
(τ (let1 (iv (make-ivar))
(for-each
(λ (tag) (spawn (τ (note! (list tag (ivar-get iv))))))
'(r1 r2 r3))
(note! (ivar-get-poll iv))
(ivar-put! iv 1)
(note! (ivar-get-poll iv))
(note! (ivar-get iv))
(note! (sync (ivar-get-evt iv)))
(cml/yield)
(note! (raised-or (exn cml put) (ivar-put! iv 2)))
(cml/shutdown)))))))
((eta 0.0) (memory #(12582912 5533128 1048576)) (stdout "") (stderr ""))
test/ivar/predicates: pass
(define (test/ivar/predicates _)
(let ((iv (make-ivar)) (mv (make-mvar)))
(⊨ (ivar? iv))
(⊭ (ivar? mv))
(⊨ (mvar? mv))
(⊭ (mvar? iv))
(⊨ (ivar=? iv iv))
(⊭ (ivar=? iv (make-ivar)))
(⊨ (mvar=? mv mv))
(⊭ (mvar=? mv (make-mvar)))
(⊨ (result? (make-result)))
(⊭ (result? iv))
(⊦raises (exn cml put) (ivar-put! iv 1) (ivar-put! iv 2))))
((eta 0.0) (memory #(12582912 5532832 1048576)) (stdout "") (stderr ""))
test/ivar-mvar-kinds: pass
(define (test/ivar-mvar-kinds _)
(define (kind-errors ops)
(map (λ (op)
(condition-case
(begin (op) 'no-raise)
(e (exn cml) 'cml-raised)
(e (exn) 'raised)))
ops))
(⊦= '((raised raised raised raised raised raised raised raised raised raised)
(raised raised raised raised raised)
(1 () (1)))
(run/value
(τ (let ((iv (make-ivar)) (mv (make-mvar)))
(ivar-put! iv 1)
(list (kind-errors
(list (τ (mvar-take! iv))
(τ (mvar-take-evt iv))
(τ (mvar-take-poll iv))
(τ (mvar-swap! iv 2))
(τ (mvar-swap-evt iv 2))
(τ (mvar-put! iv 2))
(τ (mvar-get iv))
(τ (mvar-get-evt iv))
(τ (mvar-get-poll iv))
(τ (sync (mvar-take-evt iv)))))
(kind-errors
(list (τ (ivar-put! mv 1))
(τ (ivar-get mv))
(τ (ivar-get-evt mv))
(τ (ivar-get-poll mv))
(τ (sync (ivar-get-evt mv)))))
(list (ivar-get iv)
(mvar-take-poll mv)
(ivar-get-poll iv)))))))
(⊦raises (exn) (mvar-put! (make-ivar) 1))
(⊦raises (exn) (ivar-put! (make-mvar) 1)))
((eta 0.0) (memory #(12582912 5566464 1048576)) (stdout "") (stderr ""))
test/condition-predicates: pass
(define (test/condition-predicates _)
(let ((put (condition-case
(let1 (iv (make-ivar)) (ivar-put! iv 1) (ivar-put! iv 2))
(e () e)))
(not-running (condition-case (sync (always-evt 1)) (e () e)))
(other (condition-case (car '()) (e () e))))
(⊨ (cml-condition? put))
(⊨ (cml-put-condition? put))
(⊭ (cml-not-running-condition? put))
(⊨ (cml-condition? not-running))
(⊨ (cml-not-running-condition? not-running))
(⊭ (cml-put-condition? not-running))
(⊭ (cml-condition? other))
(⊭ (cml-put-condition? other))))
((eta 0.0) (memory #(12582912 5537024 1048576)) (stdout "") (stderr ""))
test/mvar: pass
(define (test/mvar _)
(⊦= '(1 () 2 (2) 2 3 4 raised (taker 5) (swapped 6 7) (7))
(cdr (run/log
(τ (let1 (mv (make-mvar 1))
(note! (mvar-take! mv))
(note! (mvar-take-poll mv))
(mvar-put! mv 2)
(note! (mvar-get mv))
(note! (mvar-get-poll mv))
(note! (mvar-swap! mv 3))
(note! (sync (mvar-swap-evt mv 4)))
(note! (sync (mvar-take-evt mv)))
(mvar-put! mv 5)
(note! (raised-or (exn cml put) (mvar-put! mv 6)))
(spawn (τ (note! (list 'taker (mvar-take! mv)))))
(spawn (τ (note! (list 'swapped
(mvar-swap! mv 7)
(sync (mvar-get-evt mv))))))
(mvar-put! mv 6)
(cml/yield)
(note! (mvar-get-poll mv))
(cml/shutdown)))))))
((eta 0.0) (memory #(12582912 5533280 1048576)) (stdout "") (stderr ""))
test/mvar/relay-chain: pass
(define (test/mvar/relay-chain _)
(⊦= '((c x) (a x) (b x) blocked (d y))
(cdr (run/log
(τ (let1 (mv (make-mvar))
(spawn (τ (note! (list 'a (mvar-get mv)))))
(spawn (τ (note! (list 'b (sync (mvar-get-evt mv))))))
(spawn (τ (note! (list 'c (mvar-take! mv)))))
(spawn (τ (note! (list 'd (mvar-get mv)))))
(mvar-put! mv 'x)
(cml/sleep 0.01)
(note! 'blocked)
(mvar-put! mv 'y)
(cml/sleep 0.01)
(cml/shutdown)))))))
((eta 0.0) (memory #(12582912 5534456 1048576)) (stdout "") (stderr ""))
test/losing-waiters-are-dropped: pass
(define (test/losing-waiters-are-dropped _)
(define (size q) (+ (length (%queue-front q)) (length (%queue-rear q))))
(define bound (+ %clean-budget-min 2))
(let1 (r (run/value
(τ (let ((iv (make-ivar))
(mv (make-mvar))
(t (spawn (τ (sync never-evt))))
(ch (make-channel))
(blocked 0))
(spawn (τ (let loop () (cml/yield) (send ch 1) (loop))))
(for-each
(λ (i)
(select
(ivar-get-evt iv)
(mvar-take-evt mv)
(mvar-get-evt mv)
(join-evt t)
(recv-evt ch)))
(iota 200))
(list (size (%cell-read-q iv))
(size (%cell-read-q mv))
(length (%cvar-state (%tid-dead t))))))))
(⊨ (<= (car r) bound))
(⊨ (<= (cadr r) bound))
(⊨ (<= (caddr r) bound))))
((eta 0.001) (memory #(12582912 5543816 1048576)) (stdout "") (stderr ""))
test/many-waiters-block-in-linear-time: pass
(define (test/many-waiters-block-in-linear-time _)
(define (blocking-time n)
(let1 (t0 (current-process-milliseconds))
(⊦= (* 4 n)
(run/value
(τ (let* ((iv (make-ivar))
(mv (make-mvar))
(mb (make-mailbox))
(out (make-channel))
(stop (make-channel))
(t (spawn (τ (recv stop)))))
(for-each
(λ (i)
(spawn (τ (send out (ivar-get iv))))
(spawn (τ (send out (sync (mvar-get-evt mv)))))
(spawn (τ (sync (join-evt t)) (send out 1)))
(spawn (τ (send out (sync (mailbox-recv-evt mb))))))
(iota n))
(ivar-put! iv 1)
(mvar-put! mv 1)
(send stop #t)
(for-each (λ (i) (mailbox-send! mb 1)) (iota n))
(let loop ((i 0) (sum 0))
(if (= i (* 4 n))
sum
(loop (add1 i) (+ sum (recv out)))))))))
(- (current-process-milliseconds) t0)))
(let* ((small (max 20 (blocking-time 200))) (large (blocking-time 2000)))
(⊨ (< large (* 30 small)))))
((eta 0.079) (memory #(25165824 5535928 1048576)) (stdout "") (stderr ""))
test/many-ready-threads-die-in-linear-time: pass
(define (test/many-ready-threads-die-in-linear-time _)
(define (dying-time n)
(let1 (t0 (current-process-milliseconds))
(⊦= n
(parameterize
((default-exn-handler void))
(run/value
(τ (let* ((gate (make-ivar))
(ts (map (λ (i) (spawn (τ (ivar-get gate) (error "die"))))
(iota n))))
(ivar-put! gate #t)
(for-each (λ (t) (sync (join-evt t))) ts)
(length ts))))))
(- (current-process-milliseconds) t0)))
(let* ((small (max 20 (dying-time 300))) (large (dying-time 3000)))
(⊨ (< large (* 30 small)))))
((eta 0.02) (memory #(25165824 6089664 1048576)) (stdout "") (stderr ""))
test/result: pass
(define (test/result _)
(⊦= '(1 1 raised raised "boom")
(cdr (run/log
(τ (let ((r (make-result)) (s (make-result)))
(spawn (τ (note! (result-get r))))
(result-put! r 1)
(note! (sync (result-get-evt r)))
(result-put-exn!
s
(make-property-condition 'exn 'message "boom"))
(note! (raised-or (exn) (result-get s)))
(note! (raised-or (exn cml put) (result-put! s 2)))
(note! (condition-case
(sync (result-get-evt s))
(e (exn)
((condition-property-accessor 'exn 'message) e))))
(cml/shutdown)))))))
((eta 0.0) (memory #(25165824 6094864 1048576)) (stdout "") (stderr ""))
Mailboxes#
test/mailbox/ordering: pass
(define (test/mailbox/ordering _)
(⊦= (append (iota 50) '(()))
(cdr (run/log
(τ (let1 (mb (make-mailbox))
(for-each (λ (i) (mailbox-send! mb i)) (iota 50))
(for-each (λ (i) (note! (mailbox-recv mb))) (iota 25))
(for-each
(λ (i) (note! (sync (mailbox-recv-evt mb))))
(iota 25))
(note! (mailbox-recv-poll mb))
(cml/shutdown)))))))
((eta 0.0) (memory #(25165824 5529640 1048576)) (stdout "") (stderr ""))
test/mailbox/blocking-recv: pass
(define (test/mailbox/blocking-recv _)
(⊦= '((r1 a) (r2 b) sent (1) timeout)
(cdr (run/log
(τ (let1 (mb (make-mailbox))
(spawn (τ (note! (list 'r1 (mailbox-recv mb)))))
(spawn (τ (note! (list 'r2 (sync (mailbox-recv-evt mb))))))
(mailbox-send! mb 'a)
(mailbox-send! mb 'b)
(note! 'sent)
(cml/sleep 0.01)
(mailbox-send! mb 1)
(note! (mailbox-recv-poll mb))
(note! (sync/timeout
(mailbox-recv-evt mb)
0.01
'timeout))
(cml/shutdown)))))))
((eta 0.0) (memory #(25165824 5534672 1048576)) (stdout "") (stderr ""))
test/mailbox/predicates: pass
(define (test/mailbox/predicates _)
(let1 (mb (make-mailbox))
(⊨ (mailbox? mb))
(⊭ (mailbox? (make-channel)))
(⊨ (mailbox=? mb mb))
(⊭ (mailbox=? mb (make-mailbox)))
(⊨ (channel? (make-channel)))
(⊭ (channel=? (make-channel) (make-channel)))
(let1 (c (make-channel)) (⊨ (channel=? c c)))))
((eta 0.0) (memory #(25165824 5531152 1048576)) (stdout "") (stderr ""))
Barriers#
test/barrier/rounds-in-linear-time: pass
(define (test/barrier/rounds-in-linear-time _)
(define (rounds-time n)
(let1 (t0 (current-process-milliseconds))
(⊦= 3
(run/value
(τ (let* ((b (make-barrier 0 add1))
(es (map (λ (i) (barrier-enroll b)) (iota n)))
(ts (map (λ (e)
(spawn (τ (barrier-wait e)
(barrier-wait e)
(barrier-wait e))))
es)))
(for-each (λ (t) (sync (join-evt t))) ts)
(barrier-value (car es))))))
(- (current-process-milliseconds) t0)))
(let* ((small (max 20 (rounds-time 100))) (large (rounds-time 600)))
(⊨ (< large (* 15 small)))))
((eta 0.006) (memory #(25165824 5681640 1048576)) (stdout "") (stderr ""))
test/barrier/rounds: pass
(define (test/barrier/rounds _)
(let1 (log (cdr (run/log
(τ (let* ((b (make-barrier 0 add1))
(es (map (λ (i) (barrier-enroll b)) (iota 3)))
(ts (map (λ (e tag)
(spawn (τ (for-each
(λ (i) (note! (list tag (barrier-wait e))))
(iota 4)))))
es
'(x y z))))
(for-each (λ (t) (sync (join-evt t))) ts)
(note! (barrier-value (car es)))
(cml/shutdown))))))
(⊦= 4 (last log))
(⊦= '(1 2 3 4)
(map cadr (filter (λ (x) (and (pair? x) (eq? 'x (car x)))) log)))
(⊦= '(1 2 3 4)
(map cadr (filter (λ (x) (and (pair? x) (eq? 'z (car x)))) log)))
(⊦= 13 (length log))))
((eta 0.001) (memory #(25165824 5528376 1048576)) (stdout "") (stderr ""))
test/barrier/wake-order: pass
(define (test/barrier/wake-order _)
(⊦= '(1 2 3)
(run/value
(τ (let* ((b (make-barrier 0 add1))
(es (map (λ (i) (barrier-enroll b)) (iota 4)))
(mb (make-mailbox)))
(for-each
(λ (e i) (spawn (τ (barrier-wait e) (mailbox-send! mb i))))
(cdr es)
'(1 2 3))
(cml/yield)
(barrier-wait (car es))
(map (λ (i) (mailbox-recv mb)) '(1 2 3)))))))
((eta 0.0) (memory #(25165824 5530400 1048576)) (stdout "") (stderr ""))
test/barrier/resign-completes-round: pass
(define (test/barrier/resign-completes-round _)
(⊦= '((a 1) (b 1) resigned-again raised raised)
(cdr (run/log
(τ (let* ((b (make-barrier 0 add1))
(ea (barrier-enroll b))
(eb (barrier-enroll b))
(ec (barrier-enroll b)))
(spawn (τ (note! (list 'a (barrier-wait ea)))))
(spawn (τ (note! (list 'b (barrier-wait eb)))))
(barrier-resign ec)
(cml/sleep 0.01)
(barrier-resign ec)
(note! 'resigned-again)
(note! (raised-or (exn cml barrier) (barrier-wait ec)))
(spawn (τ (barrier-wait ea)))
(note! (raised-or
(exn cml barrier)
(sync (barrier-wait-evt ea))))
(cml/shutdown)))))))
((eta 0.0) (memory #(25165824 5536080 1048576)) (stdout "") (stderr ""))
test/barrier/misuse-raises-whatever-the-other-branches: pass
(define (test/barrier/misuse-raises-whatever-the-other-branches _)
(⊦= '(raised raised raised raised raised nacked 1)
(cdr (run/log
(τ (let* ((b (make-barrier 0 add1))
(e (barrier-enroll b))
(b2 (make-barrier 0 add1))
(r (barrier-enroll b2)))
(note! (raised-or
(exn cml barrier)
(select (barrier-wait-evt e) (barrier-wait-evt e))))
(barrier-resign r)
(note! (raised-or
(exn cml barrier)
(select (barrier-wait-evt r) (always-evt 'ok))))
(note! (raised-or
(exn cml barrier)
(select (always-evt 'ok) (barrier-wait-evt r))))
(note! (raised-or
(exn cml barrier)
(sync (wrap-handler
(barrier-wait-evt r)
(λ (x) 'handled)))))
(note! (raised-or
(exn cml barrier)
(select
(barrier-wait-evt r)
(with-nack
(λ (nack)
(spawn (τ (sync nack) (note! 'nacked)))
(always-evt 'ok))))))
(cml/sleep 0.01)
(note! (barrier-wait e))
(cml/shutdown)))))))
((eta 0.0) (memory #(12582912 5533360 1048576)) (stdout "") (stderr ""))
test/barrier/update-raises: pass
(define (test/barrier/update-raises _)
(⊦= '((b raised) (a raised) 5 (b 6) (a 6))
(cdr (run/log
(τ (let* ((b (make-barrier
5
(let1 (n 0)
(λ (x)
(set! n (add1 n))
(if (= n 1) (error "no") (add1 x))))))
(ea (barrier-enroll b))
(eb (barrier-enroll b)))
(spawn (τ (note! (list 'a (raised-or (exn) (barrier-wait ea))))))
(note! (list 'b (raised-or (exn) (barrier-wait eb))))
(cml/yield)
(note! (barrier-value ea))
(spawn (τ (note! (list 'a (barrier-wait ea)))))
(note! (list 'b (barrier-wait eb)))
(cml/yield)
(cml/shutdown)))))))
((eta 0.0) (memory #(12582912 5533472 1048576)) (stdout "") (stderr ""))
test/barrier/doc-example: pass
(define (test/barrier/doc-example _)
(⊦= '(5 5)
(cdr (run/log
(τ (let* ((clock (make-barrier 0 add1))
(parent (barrier-enroll clock))
(spawn-child
(τ (let1 (e (barrier-enroll clock))
(spawn (τ (let loop ()
(if (= 5 (barrier-wait e))
(note! (barrier-value e))
(loop))))))))
(ts (list (spawn-child) (spawn-child))))
(barrier-resign parent)
(for-each (λ (t) (sync (join-evt t))) ts)
(cml/shutdown)))))))
((eta 0.0) (memory #(12582912 5535280 1048576)) (stdout "") (stderr ""))
test/barrier/predicates: pass
(define (test/barrier/predicates _)
(let* ((b (make-barrier 0 add1)) (e (barrier-enroll b)))
(⊨ (barrier? b))
(⊭ (barrier? e))
(⊨ (enrollment? e))
(⊭ (enrollment? b))))
((eta 0.0) (memory #(12582912 5536184 1048576)) (stdout "") (stderr ""))
test/barrier/one-sync-counts-once: pass
(define (test/barrier/one-sync-counts-once _)
(define (three)
(let1 (b (make-barrier 0 add1))
(list b (barrier-enroll b) (barrier-enroll b) (barrier-enroll b))))
(⊦= '((a raised) timeout 0)
(cdr (run/log
(τ (let* ((l (three)) (e1 (cadr l)) (e2 (caddr l)))
(spawn (τ (note! (list 'a
(raised-or
(exn cml barrier)
(select
(barrier-wait-evt e1)
(barrier-wait-evt e1)))))))
(note! (sync/timeout (barrier-wait-evt e2) 0.02 'timeout))
(note! (barrier-value e1))
(cml/shutdown))))))
(⊦= '(u-blocked 0)
(cdr (run/log
(τ (let* ((l (three))
(e1 (cadr l))
(e2 (caddr l))
(e3 (cadddr l)))
(spawn (τ (note! (list 't
(select
(barrier-wait-evt e1)
(barrier-wait-evt e2))))))
(let1 (u (spawn (τ (note! (list 'u (barrier-wait e3))))))
(note! (sync/timeout
(wrap (join-evt u) (λ ignored 'u-done))
0.02
'u-blocked))
(note! (barrier-value e1))
(cml/shutdown))))))))
((eta 0.0) (memory #(12582912 5542968 1048576)) (stdout "") (stderr ""))
test/barrier/wait-evt-in-choice: pass
(define (test/barrier/wait-evt-in-choice _)
(⊦= '(timeout (a 1) (b 1))
(cdr (run/log
(τ (let* ((bar (make-barrier 0 add1))
(ea (barrier-enroll bar))
(eb (barrier-enroll bar)))
(note! (sync/timeout (barrier-wait-evt ea) 0.01 'timeout))
(spawn (τ (note! (list 'b (barrier-wait eb)))))
(note! (list 'a (barrier-wait ea)))
(cml/yield)
(cml/shutdown)))))))
((eta 0.0) (memory #(12582912 5538400 1048576)) (stdout "") (stderr ""))
Cleaners and logged channels#
test/cleaners: pass
(define (test/cleaners _)
(set! cml-test/log '())
(dynamic-wind
void
(τ (⊦= '()
(cml/add-cleaner!
"t1"
'(at-init at-shutdown)
(λ (w) (note! (list 't1 w)))))
(⊦= '() (cml/add-cleaner! "t2" 'at-init (λ (w) (note! (list 't2 w)))))
(⊦= '()
(cml/add-cleaner! "t3" 'at-shutdown (λ (w) (note! (list 't3 w)))))
(⊦= 'x (run-cml (τ (note! 'main) (cml/shutdown 'x))))
(⊦= '((t1 at-init) (t2 at-init) main (t3 at-shutdown) (t1 at-shutdown))
(reverse cml-test/log))
(let1 (old (cml/add-cleaner! "t2" 'at-shutdown (λ (w) (void))))
(⊦= 1 (length old))
(⊦= '(at-init) (car (car old))))
(⊦= 1 (length (cml/remove-cleaner! "t1")))
(⊦= 1 (length (cml/remove-cleaner! "t2")))
(⊦= 1 (length (cml/remove-cleaner! "t3")))
(⊦= '() (cml/remove-cleaner! "t3"))
(⊦= '(at-exit at-shutdown at-init at-init-fn) cml/at-all)
(⊦= '() (cml/add-cleaner! "t4" 'all (λ (w) (note! (list 't4 w)))))
(set! cml-test/log '())
(⊦= 'y (run-cml (τ (note! 'main) (cml/shutdown 'y))))
(⊦= '((t4 at-init) main (t4 at-shutdown)) (reverse cml-test/log))
(⊦= (list cml/at-all) (map car (cml/remove-cleaner! "t4"))))
(τ (for-each cml/remove-cleaner! '("t1" "t2" "t3" "t4"))))
(⊦= "[000002]" (run/value (τ (tid->string (current-tid))))))
((eta 0.0) (memory #(12582912 5527520 1048576)) (stdout "") (stderr ""))
test/logged-channel: pass
(define (test/logged-channel _)
(let ((ch (make-channel)) (control (make-channel)) (mb (make-mailbox)))
(define (state c)
(list (%channel-priority c) (%q-empty? (%channel-out-q c))))
(dynamic-wind
(τ (cml/log-channel! "test-ch" ch) (cml/log-mailbox! "test-mb" mb))
(τ (⊦= '(first x x)
(run/log
(τ (for-each
(λ (c) (spawn (τ (send c 'stale))))
(list ch control))
(note! (select (recv-evt ch) (always-evt 'x)))
(note! (select (recv-evt control) (always-evt 'x)))
(mailbox-send! mb 'old)
(cml/shutdown 'first))))
(⊦= '((1 #t) (2 #f)) (list (state ch) (state control)))
(⊦= '(() ())
(run/value (τ (list (recv-poll ch) (mailbox-recv-poll mb))))))
(τ (cml/unlog-channel! "test-ch") (cml/unlog-mailbox! "test-mb")))
(⊦raises (exn cml unlog) (cml/unlog-channel! "test-ch"))))
((eta 0.0) (memory #(12582912 5533592 1048576)) (stdout "") (stderr ""))
test/logged-server-shutdown-gets-2-seconds: pass
(define (test/logged-server-shutdown-gets-2-seconds _)
(let1 (done #f)
(dynamic-wind
(τ (cml/log-server!
"test-slow-server"
void
(τ (cml/sleep 1.3) (set! done #t))))
(τ (⊦= 'first (run-cml (τ (cml/shutdown 'first)))) (⊨ done))
(τ (cml/unlog-server! "test-slow-server")))))
((eta 0.0) (memory #(12582912 5533480 1048576)) (stdout "") (stderr ""))
test/unlog-all: pass
(define (test/unlog-all _)
(let1 (mb (make-mailbox))
(cml/log-mailbox! "test-mb-all" mb)
(cml/unlog-all!)
(⊦raises (exn cml unlog) (cml/unlog-mailbox! "test-mb-all"))
(run-cml (τ (mailbox-send! mb 'kept) (cml/shutdown)))
(⊦= '(kept) (run/value (τ (mailbox-recv-poll mb))))))
((eta 0.0) (memory #(12582912 5532456 1048576)) (stdout "") (stderr ""))
test/stale-across-runs: pass
(define (test/stale-across-runs _)
(let ((ch (make-channel)) (iv (make-ivar)))
(⊦= 'first
(run-cml
(τ (spawn (τ (send ch 'stale)))
(spawn (τ (ivar-get iv)))
(cml/shutdown 'first))))
(⊦= '(() #f ok)
(run/value
(τ (ivar-put! iv 'ok)
(list (recv-poll ch) (send-poll ch 1) (ivar-get iv)))))))
((eta 0.0) (memory #(12582912 5531144 1048576)) (stdout "") (stderr ""))
test/stale-waiters-of-blocking-operations-are-dropped: pass
(define (test/stale-waiters-of-blocking-operations-are-dropped _)
(define (size q) (+ (length (%queue-front q)) (length (%queue-rear q))))
(define (mb-size mb)
(let1 (st (%mailbox-state mb))
(if (eq? 'empty (car st))
(+ (length (cadr st)) (length (cddr st)))
0)))
(define bound (* 2 (+ %clean-budget-min 2)))
(let ((in (make-channel)) (out (make-channel)) (mb (make-mailbox)))
(for-each
(λ (i)
(run-cml
(τ (spawn (τ (recv in)))
(spawn (τ (send out i)))
(spawn (τ (mailbox-recv mb)))
(cml/yield)
(cml/shutdown))))
(iota 200))
(⊨ (<= (size (%channel-in-q in)) bound))
(⊨ (<= (size (%channel-out-q out)) bound))
(⊨ (<= (mb-size mb) bound))))
((eta 0.006) (memory #(12582912 5540000 1048576)) (stdout "") (stderr ""))
test/bad-arguments-raise-at-registration: pass
(define (test/bad-arguments-raise-at-registration _)
(define (raises? thunk)
(condition-case (begin (thunk) 'no-raise) (e (exn) 'raised)))
(let1 (mb (make-mailbox))
(dynamic-wind
void
(τ (⊦= '(raised raised raised raised raised raised raised raised)
(map raises?
(list (τ (cml/add-cleaner! "test-bad" 'at-init 42))
(τ (cml/add-cleaner! "test-bad" 'bogus void))
(τ (cml/add-cleaner!
"test-bad"
'(at-init bogus)
void))
(τ (cml/add-cleaner! "test-bad" 42 void))
(τ (cml/log-channel! "test-bad" (make-mailbox)))
(τ (cml/log-mailbox! "test-bad" (make-channel)))
(τ (cml/log-server! "test-bad" 42 void))
(τ (cml/log-server! "test-bad" void 42)))))
(⊦= '() (cml/remove-cleaner! "test-bad"))
(map (λ (unlog) (⊦raises (exn cml unlog) (unlog "test-bad")))
(list cml/unlog-channel!
cml/unlog-mailbox!
cml/unlog-server!))
(cml/log-mailbox! "test-mb-after" mb)
(⊦= 'ok
(run-cml (τ (mailbox-send! mb 'stale) (cml/shutdown 'ok))))
(⊦= '(ran ()) (run/value (τ (list 'ran (mailbox-recv-poll mb))))))
(τ (condition-case
(cml/unlog-mailbox! "test-mb-after")
(e () (void)))
(cml/remove-cleaner! "test-bad")
(for-each
(λ (unlog) (condition-case (unlog "test-bad") (e () (void))))
(list cml/unlog-channel!
cml/unlog-mailbox!
cml/unlog-server!))))))
((eta 0.0) (memory #(12582912 5534608 1048576)) (stdout "") (stderr ""))
test/registration-lock-given-back-by-a-dying-thread: pass
(define (test/registration-lock-given-back-by-a-dying-thread _)
(set! cml-test/log '())
(dynamic-wind
(τ (cml/add-cleaner!
"test-probe"
'at-shutdown
(λ (w) (note! 'shutdown-cleaner))))
(τ (parameterize
((default-exn-handler (λ (e) (note! 'died))))
(⊦= 'ok
(run-cml
(τ (let1 (t (spawn (τ (let1 (armed #t)
(dynamic-wind
void
(τ (cml/add-cleaner! "test-dying" 'at-shutdown void))
(τ (when armed
(set! armed #f)
(error "after thunk"))))))))
(sync (join-evt t))
(note! (sync/timeout
(wrap (join-evt
(spawn (τ (cml/log-channel! "test-dying-ch" (make-channel)))))
(λ ignored 'logged))
0.5
'blocked))
(note! 'shutdown)
(cml/shutdown 'ok)))
#:quantum
1)))
(⊦= '(died logged shutdown shutdown-cleaner) (reverse cml-test/log)))
(τ (for-each cml/remove-cleaner! '("test-probe" "test-dying"))
(condition-case (cml/unlog-channel! "test-dying-ch") (e () (void))))))
((eta 0.0) (memory #(12582912 5537488 1048576)) (stdout "") (stderr ""))
test/logged-server: pass
(define (test/logged-server _)
(set! cml-test/log '())
(dynamic-wind
(τ (cml/log-server! "srv" (τ (note! 'start)) (τ (note! 'stop))))
(τ (run-cml (τ (note! 'main) (cml/shutdown))))
(τ (cml/unlog-server! "srv")))
(run-cml (τ (cml/shutdown)))
(⊦= '(start main stop) (reverse cml-test/log)))
((eta 0.0) (memory #(12582912 5538360 1048576)) (stdout "") (stderr ""))
test/debug: pass
(define (test/debug _)
(⊦= "" (with-error-output-to-string (τ (cml/debug "quiet"))))
(⊦= "[-000001] hello 1\n"
(with-error-output-to-string
(τ (parameterize ((cml/debug? #t)) (cml/debug "hello" 1))))))
((eta 0.0) (memory #(12582912 5538160 1048576)) (stdout "") (stderr ""))
Operating system events#
test/io-evt/pipe: pass
(define (test/io-evt/pipe _)
(receive
(in out)
(create-pipe)
(let1 (log (cdr (run/log
(τ (spawn (τ (cml/sleep 0.02)
(note! 'writing)
(file-write out (string->utf8 "hello\n"))))
(note! (list 'ready (= in (sync (io-evt in 'input)))))
(note! (read-line (open-input-file* in)))
(note! (list 'out (= out (sync (io-evt out 'output)))))
(cml/shutdown)))))
(⊦= '(writing (ready #t) "hello" (out #t)) log))
(file-close out)))
((eta 0.0) (memory #(12582912 5524784 1048576)) (stdout "") (stderr ""))
test/poll-evt: pass
(define (test/poll-evt _)
(receive
(in1 out1)
(create-pipe)
(receive
(in2 out2)
(create-pipe)
(let1 (log (cdr (run/log
(τ (note! (sync/timeout
(poll-evt (list in1 'input) (list in2 'input))
0.02
'nothing))
(spawn (τ (cml/sleep 0.01)
(file-write out2 (string->utf8 "x"))))
(note! (sync (poll-evt (list in1 'input) (list in2 'input))))
(file-write out1 (string->utf8 "y"))
(note! (length
(sync (poll-evt (list in1 'input) (list in2 'input)))))
(cml/shutdown)))))
(⊦= `(nothing ((,in2 input)) 2) log))
(⊦= `((,in1 input) (,in2 input))
(run/value
(τ (sync (poll-evt* (list (list in1 'input) (list in2 'input)))))))
(⊨ (eq? never-evt (poll-evt* '())))
(for-each file-close (list in1 out1 in2 out2)))))
((eta 0.001) (memory #(12582912 5525592 1048576)) (stdout "") (stderr ""))
test/io-evt/yield-loop: pass
(define (test/io-evt/yield-loop _)
(receive
(in out)
(create-pipe)
(⊦= 'io-ready
(run/value
(τ (let1 (iv (make-ivar))
(spawn (τ (sync (io-evt in 'input))
(ivar-put! iv 'io-ready)))
(file-write out (string->utf8 "x"))
(let loop ((n 0))
(cond ((pair? (ivar-get-poll iv))
(car (ivar-get-poll iv)))
((< n 1000000) (cml/yield) (loop (add1 n)))
(else 'starved)))))))
(for-each file-close (list in out))))
((eta 0.0) (memory #(12582912 5529848 1048576)) (stdout "") (stderr ""))
test/io-evt/idle-waiters-do-not-slow-others: pass
(define (test/io-evt/idle-waiters-do-not-slow-others _)
(define (hand-off-time n)
(receive
(r w)
(create-pipe)
(let1 (t (run-cml
(τ (for-each
(λ (i) (spawn (τ (sync (io-evt r 'input)))))
(iota n))
(cml/sleep 0.02)
(let ((ch (make-channel))
(t0 (current-process-milliseconds)))
(spawn (τ (let loop ((i 0))
(when (< i 20000) (send ch i) (loop (add1 i))))))
(let loop ((i 0))
(when (< i 20000) (recv ch) (loop (add1 i))))
(cml/shutdown (- (current-process-milliseconds) t0))))))
(file-close r)
(file-close w)
t)))
(let* ((none (max 50 (hand-off-time 0))) (idle (hand-off-time 2000)))
(⊨ (< idle (* 4 none)))))
((eta 0.097) (memory #(12582912 5531448 1048576)) (stdout "") (stderr ""))
test/process-evt: pass
(define (test/process-evt _)
(let1 (log (cdr (run/log
(τ (let* ((p1 (process-run "true"))
(p2 (process-run "sh" '("-c" "exit 3")))
(e1 (process-evt p1)))
(note! (eq? e1 (process-evt p1)))
(note! (sync (process-evt p2)))
(note! (sync e1))
(note! (sync e1))
(cml/shutdown))))))
(⊦= '(#t (#t 3) (#t 0) (#t 0)) log))
(⊦= '(0 (#t 0) (#t 0) (#t 0) raised)
(cdr (run/log
(τ (for-each
(λ (i) (sync (process-evt (process-run "true"))))
(iota 20))
(note! (length %proc-memo))
(let ((p (process-run "true")) (q (process-run "true")))
(note! (sync (process-evt p)))
(note! (sync (process-evt p)))
(note! (sync (process-evt (process-id q))))
(note! (raised-or
(exn)
(sync (process-evt (process-id q))))))
(cml/shutdown))))))
((eta 0.017) (memory #(12582912 5534824 1048576)) (stdout "") (stderr ""))
test/process-evt/bad-pid: pass
(define (test/process-evt/bad-pid _)
(⊦= '(raised raised raised raised)
(run/value
(τ (map (λ (pid)
(condition-case
(begin (process-evt pid) 'no-raise)
(e (exn) 'raised)))
(list 'y "x" 0 -1))))))
((eta 0.0) (memory #(12582912 5545096 1048576)) (stdout "") (stderr ""))
test/process-evt/keeps-run-alive: pass
(define (test/process-evt/keeps-run-alive _)
(⊦= '(done (#t 0))
(cdr (run/log
(τ (let1 (p (process-run "sh" '("-c" "sleep 0.05")))
(spawn (τ (note! (sync (process-evt p)))))
(note! 'done)))))))
((eta 0.001) (memory #(12582912 5547136 1048576)) (stdout "") (stderr ""))
A prime sieve and a server with with-nack#
Classic CML examples
test/sieve: pass
(define (test/sieve _)
(define (counter n)
(let1 (ch (make-channel))
(spawn (τ (let loop ((i n)) (send ch i) (loop (add1 i)))))
ch))
(define (sift p in)
(let1 (out (make-channel))
(spawn (τ (let loop ()
(let1 (i (recv in))
(unless (zero? (remainder i p)) (send out i)))
(loop))))
out))
(define (primes)
(let1 (ch (make-channel))
(spawn (τ (let loop ((in (counter 2)))
(let1 (p (recv in)) (send ch p) (loop (sift p in))))))
ch))
(⊦= '(2 3 5 7 11 13 17 19 23 29 31 37 41 43 47)
(run/value (τ (let1 (ps (primes)) (map (λ (i) (recv ps)) (iota 15)))))))
((eta 0.0) (memory #(12582912 5545608 1048576)) (stdout "") (stderr ""))
test/server-with-nack: pass
(define (test/server-with-nack _)
(define (make-server)
(let1 (req (make-channel))
(spawn (τ (let loop ((n 0))
(let* ((r (recv req)) (reply (car r)) (nack (cdr r)))
(select
(wrap (send-evt reply n)
(λ (x) (note! (list 'replied n))))
(wrap nack (λ (x) (note! (list 'abandoned n)))))
(loop (add1 n))))))
(λ ()
(with-nack
(λ (nack)
(let1 (reply (make-channel))
(spawn (τ (send req (cons reply nack))))
(recv-evt reply)))))))
(⊦= '((got 0) (replied 0) gave-up (abandoned 1) (replied 2) (got 2))
(cdr (run/log
(τ (let1 (request-evt (make-server))
(note! (list 'got (sync (request-evt))))
(note! (select
(request-evt)
(wrap (timeout-evt 0) (λ (x) 'gave-up))))
(cml/sleep 0.01)
(let1 (v (sync (request-evt)))
(cml/yield)
(note! (list 'got v)))
(cml/shutdown)))))))
((eta 0.001) (memory #(12582912 5535728 1048576)) (stdout "") (stderr ""))
Multicast channels#
Concurrent ML library: Multicast
Asynchronous one-to-many channels, a port of Multicast from cml-lib.
test/predicates: pass
(define (test/predicates _)
(let1 (mc (make-multicast-channel))
(⊨ (multicast-channel? mc))
(⊭ (multicast-channel? (make-channel)))
(⊦= "#<multicast-channel>" (->string mc))
(⊦= '(#t #f)
(run/value
(τ (let1 (p (multicast-port mc))
(list (multicast-port? p) (multicast-port? mc))))))
(⊦raises (exn cml not-running) (multicast-port mc))
(multicast! mc 'nobody-listens)))
((eta 0.0) (memory #(12582912 3158992 1048576)) (stdout "") (stderr ""))
test/independent-speeds: pass
(define (test/independent-speeds _)
(let1 (got (run/value
(τ (let* ((mc (make-multicast-channel))
(ps (map (λ (i) (multicast-port mc)) (iota 3)))
(logs (make-vector 3 '()))
(reader
(λ (i p pause)
(spawn (τ (let loop ((n 6))
(unless
(zero? n)
(for-each (λ (j) (cml/yield)) (iota pause))
(vector-set!
logs
i
(cons (multicast-recv p) (vector-ref logs i)))
(loop (sub1 n)))))))))
(let1 (ts (list (reader 0 (car ps) 0)
(reader 1 (cadr ps) 3)))
(for-each (λ (v) (multicast! mc v)) '(1 2 3 4 5 6))
(let1 (late (reader 2 (caddr ps) 1))
(join-all (cons late ts))
(map reverse (vector->list logs))))))))
(⊦= '((1 2 3 4 5 6) (1 2 3 4 5 6) (1 2 3 4 5 6)) got)))
((eta 0.0) (memory #(12582912 3158264 1048576)) (stdout "") (stderr ""))
test/new-port-sees-later-messages: pass
(define (test/new-port-sees-later-messages _)
(⊦= '((a b c) (c) (d))
(run/value
(τ (let* ((mc (make-multicast-channel)) (p1 (multicast-port mc)))
(multicast! mc 'a)
(multicast! mc 'b)
(let1 (p2 (multicast-port mc))
(multicast! mc 'c)
(let1 (p3 (multicast-port mc))
(multicast! mc 'd)
(list (map (λ (i) (multicast-recv p1)) '(1 2 3))
(list (multicast-recv p2))
(list (multicast-recv p3))))))))))
((eta 0.001) (memory #(12582912 3157264 1048576)) (stdout "") (stderr ""))
test/copy-port: pass
(define (test/copy-port _)
(⊦= '(1 (2 3 4) (2 3 4) 5)
(run/value
(τ (let* ((mc (make-multicast-channel)) (p (multicast-port mc)))
(for-each (λ (v) (multicast! mc v)) '(1 2 3))
(let1 (first (multicast-recv p))
(let1 (c (multicast-copy-port p))
(multicast! mc 4)
(let* ((from-c
(map (λ (i) (multicast-recv c)) '(1 2 3)))
(from-p
(map (λ (i) (multicast-recv p)) '(1 2 3)))
(cc (multicast-copy-port (multicast-copy-port c))))
(multicast! mc 5)
(list first from-c from-p (multicast-recv cc)))))))))
(⊦= '((x y) (x y))
(run/value
(τ (let* ((mc (make-multicast-channel))
(p (multicast-port mc))
(c (multicast-copy-port p)))
(multicast! mc 'x)
(multicast! mc 'y)
(list (list (multicast-recv p) (multicast-recv p))
(list (multicast-recv c) (multicast-recv c))))))))
((eta 0.0) (memory #(12582912 3157456 1048576)) (stdout "") (stderr ""))
test/recv-evt: pass
(define (test/recv-evt _)
(⊦= '(timeout (got 7) (got 8) timeout)
(cdr (run/log
(τ (let* ((mc (make-multicast-channel))
(p (multicast-port mc))
(poll (τ (note! (sync/timeout
(wrap (multicast-recv-evt p)
(λ (v) (list 'got v)))
0.01
'timeout)))))
(poll)
(multicast! mc 7)
(multicast! mc 8)
(poll)
(note! (list 'got (sync (multicast-recv-evt p))))
(poll)
(cml/shutdown)))))))
((eta 0.0) (memory #(12582912 3165472 1048576)) (stdout "") (stderr ""))
test/shared-port: pass
(define (test/shared-port _)
(let1 (got (run/value
(τ (let* ((mc (make-multicast-channel))
(p (multicast-port mc))
(acc '())
(ts (map (λ (i)
(spawn (τ (let loop ((n 4))
(unless
(zero? n)
(set! acc (cons (multicast-recv p) acc))
(loop (sub1 n)))))))
'(1 2))))
(for-each (λ (v) (multicast! mc v)) (iota 8))
(join-all ts)
acc))))
(⊦= (iota 8) (sort got <))))
((eta 0.0) (memory #(12582912 3159984 1048576)) (stdout "") (stderr ""))
test/many-messages: pass
(define (test/many-messages _)
(⊦= (iota 500)
(run/value
(τ (let* ((mc (make-multicast-channel)) (p (multicast-port mc)))
(spawn (τ (for-each (λ (v) (multicast! mc v)) (iota 500))))
(map (λ (i) (multicast-recv p)) (iota 500)))))))
((eta 0.002) (memory #(12582912 3185192 1048576)) (stdout "") (stderr ""))
test/producer-keeps-to-its-readers-pace: pass
(define (test/producer-keeps-to-its-readers-pace _)
(let1 (lags (run/value
(τ (let* ((mc (make-multicast-channel))
(ps (map (λ (i) (multicast-port mc)) (iota 2)))
(got (make-vector 2 0)))
(for-each
(λ (i p)
(spawn (τ (let loop ()
(multicast-recv p)
(vector-set! got i (add1 (vector-ref got i)))
(loop)))))
'(0 1)
ps)
(let loop ((i 0) (lag 0))
(if (= i 20000)
lag
(begin
(multicast! mc i)
(loop (add1 i)
(max lag
(- (add1 i)
(min (vector-ref got 0) (vector-ref got 1))))))))))))
(⊨ (< lags 50))))
((eta 0.122) (memory #(12582912 3162768 1048576)) (stdout "") (stderr ""))
Remote procedure calls#
Concurrent ML library: SimpleRPC
Server-side RPC entry points, a port of SimpleRPC: every maker returns (values call entry-evt).
test/rpc: pass
(define (test/rpc _)
(⊦= '((c1 1) (c1 9) (c2 4) served-3)
(cdr (run/log
(τ (receive
(call entry-evt)
(make-rpc (λ (x) (* x x)))
(let* ((server
(spawn (τ (for-each (λ (i) (sync entry-evt)) '(1 2 3)))))
(c1-log '())
(c2-log '())
(c1 (spawn (τ (set! c1-log (list (call 1) (call 3))))))
(c2 (spawn (τ (set! c2-log (list (call 2)))))))
(join-all (list c1 c2 server))
(for-each (λ (v) (note! (list 'c1 v))) c1-log)
(for-each (λ (v) (note! (list 'c2 v))) c2-log)
(note! 'served-3)
(cml/shutdown))))))))
((eta 0.0) (memory #(12582912 3153056 1048576)) (stdout "") (stderr ""))
test/rpc/fifo: pass
(define (test/rpc/fifo _)
(⊦= '(a b c d)
(run/value
(τ (receive
(call entry-evt)
(make-rpc (λ (x) (note! x) x))
(set! cml-test/log '())
(let1 (ts (map (λ (x) (spawn (τ (call x)))) '(a b c d)))
(for-each (λ (i) (sync entry-evt)) (iota 4))
(join-all ts)
(reverse cml-test/log)))))))
((eta 0.0) (memory #(12582912 3154312 1048576)) (stdout "") (stderr ""))
test/rpc/in: pass
(define (test/rpc/in _)
(⊦= '((1 . 100) (2 . 100) (3 . 200))
(run/value
(τ (receive
(call entry-evt)
(make-rpc/in (λ (x st) (cons x st)))
(spawn (τ (sync (entry-evt 100))
(sync (entry-evt 100))
(sync (entry-evt 200))))
(map call '(1 2 3)))))))
((eta 0.0) (memory #(12582912 3154496 1048576)) (stdout "") (stderr ""))
test/rpc/out: pass
(define (test/rpc/out _)
(⊦= '(((ack a) (ack b)) (b a))
(run/value
(τ (receive
(call entry-evt)
(make-rpc/out (λ (x) (values (list 'ack x) x)))
(let* ((states '())
(server
(spawn (τ (for-each
(λ (i)
(set! states (cons (sync entry-evt) states)))
'(1 2))))))
(let1 (replies (map call '(a b)))
(sync (join-evt server))
(list replies states))))))))
((eta 0.0) (memory #(12582912 3156352 1048576)) (stdout "") (stderr ""))
test/rpc/in-out: pass
(define (test/rpc/in-out _)
(⊦= '((1 3 6 10) 10)
(run/value
(τ (receive
(call entry-evt)
(make-rpc/in-out (λ (x acc) (let1 (s (+ x acc)) (values s s))))
(let* ((final #f)
(server
(spawn (τ (let loop ((acc 0) (n 4))
(if (zero? n)
(set! final acc)
(loop (sync (entry-evt acc)) (sub1 n))))))))
(let1 (replies (map call '(1 2 3 4)))
(sync (join-evt server))
(list replies final))))))))
((eta 0.0) (memory #(12582912 3158560 1048576)) (stdout "") (stderr ""))
test/rpc/state: pass
(define (test/rpc/state _)
(⊦= '((1 2 3) (count-1 count-2 count-3))
(run/value
(τ (receive
(call entry-evt)
(make-rpc/state 0 (λ (x n) (values (add1 n) (add1 n))))
(let* ((states '())
(server
(spawn (τ (for-each
(λ (i)
(set! states (cons (sync entry-evt) states)))
'(1 2 3))))))
(let1 (replies (map call '(x y z)))
(sync (join-evt server))
(list replies
(map (λ (n)
(string->symbol
(string-append "count-" (number->string n))))
(reverse states))))))))))
((eta 0.0) (memory #(12582912 3159696 1048576)) (stdout "") (stderr ""))
test/rpc/state/shared: pass
(define (test/rpc/state/shared _)
(⊦= '(1 2 3 3)
(run/value
(τ (receive
(call entry-evt)
(make-rpc/state 0 (λ (x st) (values (+ st x) (+ st x))))
(for-each
(λ (i) (spawn (τ (let loop () (sync entry-evt) (loop)))))
'(1 2))
(let1 (replies (map call '(1 1 1)))
(append replies (list (call 0))))))))
(for-each
(λ (quantum)
(let1 (replies
(run-cml
(τ (receive
(call entry-evt)
(make-rpc/state
0
(λ (x st) (values (add1 st) (add1 st))))
(for-each
(λ (i)
(spawn (τ (let loop () (sync entry-evt) (loop)))))
(iota 5))
(let ((done (make-mailbox)) (replies '()))
(for-each
(λ (i)
(spawn (τ (for-each
(λ (j) (push! (call j) replies))
(iota 40))
(mailbox-send! done #t))))
(iota 5))
(for-each (λ (i) (mailbox-recv done)) (iota 5))
(cml/shutdown replies))))
#:quantum
quantum))
(⊦= (iota 200 1) (sort replies <))))
'(1 3 64)))
((eta 0.003) (memory #(12582912 3169160 1048576)) (stdout "") (stderr ""))
test/rpc/values: pass
(define (test/rpc/values _)
(⊦= (list (void) 'raised 5 'raised 7)
(run/value
(τ (receive
(call entry-evt)
(make-rpc (λ (x) (values)))
(receive
(call2 entry2)
(make-rpc/in-out
(λ (x st) (if (zero? x) st (values (+ x st) (+ x st)))))
(receive
(call3 entry3)
(make-rpc/state
0
(λ (x st)
(if (zero? x) (values) (values (+ x st) (+ x st)))))
(spawn (τ (let loop () (sync entry-evt) (loop))))
(spawn (τ (let loop ((st 0)) (loop (sync (entry2 st))))))
(spawn (τ (let loop () (sync entry3) (loop))))
(list (call 1)
(raised-or (exn cml rpc) (call2 0))
(call2 5)
(raised-or (exn cml rpc) (call3 0))
(call3 7)))))))))
((eta 0.0) (memory #(12582912 3178992 1048576)) (stdout "") (stderr ""))
test/rpc/select-loop: pass
(define (test/rpc/select-loop _)
(⊦= '((get 0) (get 5) stopped)
(cdr (run/log
(τ (receive
(call-get get-evt)
(make-rpc/in (λ (ignored st) st))
(receive
(call-add add-evt)
(make-rpc/in-out (λ (x st) (values (void) (+ x st))))
(let* ((stop (make-channel))
(server
(spawn (τ (let loop ((st 0))
(select/case
((get-evt st) () (loop st))
((add-evt st) (new) (loop new))
((recv-evt stop) () (note! 'stopped))))))))
(note! (list 'get (call-get #f)))
(call-add 2)
(call-add 3)
(note! (list 'get (call-get #f)))
(send stop #t)
(sync (join-evt server))
(cml/shutdown)))))))))
((eta 0.0) (memory #(12582912 3166544 1048576)) (stdout "") (stderr ""))
test/rpc/exceptions: pass
(define (test/rpc/exceptions _)
(⊦= '((raised "no zero" 0) (ok 1/2) (raised "no zero" 0) (ok 1/4) served)
(cdr (run/log
(τ (receive
(call entry-evt)
(make-rpc (λ (x) (if (zero? x) (error "no zero" x) (/ 1 x))))
(let1 (server
(spawn (τ (for-each (λ (i) (sync entry-evt)) (iota 4)))))
(for-each
(λ (x)
(note! (condition-case
(list 'ok (call x))
(e (exn) (list 'raised (message-of e) x)))))
'(0 2 0 4))
(sync (join-evt server))
(note! 'served)
(cml/shutdown)))))))
(⊦= '(caught boom)
(run/value
(τ (receive
(call entry-evt)
(make-rpc (λ (x) (abort x)))
(spawn (τ (sync entry-evt)))
(handle-exceptions e (list 'caught e) (call 'boom)))))))
((eta 0.0) (memory #(12582912 3166088 1048576)) (stdout "") (stderr ""))
test/rpc/exceptions/state: pass
(define (test/rpc/exceptions/state _)
(⊦= '((1 raised 3) 3)
(run/value
(τ (receive
(call entry-evt)
(make-rpc/in-out
(λ (x acc)
(when (negative? x) (error "negative" x))
(values (+ x acc) (+ x acc))))
(let* ((final #f)
(server
(spawn (τ (let loop ((acc 0) (n 3))
(if (zero? n)
(set! final acc)
(loop (sync (entry-evt acc)) (sub1 n))))))))
(let1 (replies
(map (λ (x) (condition-case (call x) ((exn) 'raised)))
'(1 -5 2)))
(sync (join-evt server))
(list replies final)))))))
(⊦= '((server raised "odd") raised)
(cdr (run/log
(τ (receive
(call entry-evt)
(make-rpc/out (λ (x) (error "odd" x)))
(let1 (server
(spawn (τ (condition-case
(sync entry-evt)
(e (exn)
(note! (list 'server 'raised (message-of e))))))))
(let1 (r (condition-case (call 1) ((exn) 'raised)))
(sync (join-evt server))
(note! r))
(cml/shutdown))))))))
((eta 0.0) (memory #(12582912 3166776 1048576)) (stdout "") (stderr ""))
Tracing and watching threads#
Concurrent ML library: TraceCML
Trace modules, thread watching and uncaught exception handlers, a port of TraceCML with its bugs fixed.
test/modules: pass
(define (test/modules _)
(let* ((a (trace-module trace-module/root "lib-a"))
(b (trace-module a "b"))
(c (trace-module b "c")))
(⊨ (trace-module? a))
(⊭ (trace-module? "/lib-a/"))
(⊭ (trace-module? (make-channel)))
(⊦= "/" (trace-module-name trace-module/root))
(⊦= "/lib-a/" (trace-module-name a))
(⊦= "/lib-a/b/c/" (trace-module-name c))
(⊦= "#<trace-module /lib-a/b/>" (->string b))
(⊦= "/ThreadWatcher/" (trace-module-name trace-watcher))
(⊨ (eq? b (trace-module a "b")))
(⊨ (eq? b (trace-module "/lib-a/" "b")))
(⊨ (eq? c (trace-module-of "/lib-a/b/c/")))
(⊨ (eq? c (trace-module-of "lib-a//b/c")))
(⊨ (eq? trace-module/root (trace-module-of "/")))
(⊨ (eq? trace-module/root (trace-module-of "")))
(⊦raises (exn cml no-such-module) (trace-module-of "/lib-a/nope/"))
(⊦raises (exn cml trace) (trace-module a "x/y"))
(⊦raises (exn cml trace) (trace-module a ""))
(⊦= '(raised ok)
(run/value
(τ (list (raised-or
(exn cml no-such-module)
(trace-module-of "/nope/"))
(and (eq? c (trace-module-of "/lib-a/b/c")) 'ok)))))))
((eta 0.0) (memory #(12582912 3193096 1048576)) (stdout "") (stderr ""))
test/on-off-hierarchy: pass
(define (test/on-off-hierarchy _)
(let* ((a (trace-module trace-module/root "lib-h"))
(b (trace-module a "b"))
(c (trace-module b "c"))
(d (trace-module a "d"))
(flags (τ (map tracing? (list a b c d)))))
(⊦= '(#f #f #f #f) (flags))
(trace-on! b)
(⊦= '(#f #t #t #f) (flags))
(⊨ (tracing? (trace-module c "new")))
(⊭ (tracing? (trace-module d "new")))
(trace-on! a)
(⊦= '(#t #t #t #t) (flags))
(trace-off! b)
(⊦= '(#t #f #f #t) (flags))
(trace-off! "/lib-h/")
(trace-on-only! b)
(⊦= '(#f #t #f #f) (flags))
(⊦= '(("/lib-h/" . #f)
("/lib-h/d/" . #f)
("/lib-h/d/new/" . #f)
("/lib-h/b/" . #t)
("/lib-h/b/c/" . #f)
("/lib-h/b/c/new/" . #f))
(map (λ (p) (cons (trace-module-name (car p)) (cdr p)))
(trace-status a)))
(⊨ (tracing? trace-watcher))
(trace-off! a)))
((eta 0.0) (memory #(12582912 3191240 1048576)) (stdout "") (stderr ""))
test/trace-output: pass
(define (test/trace-output _)
(let* ((m (trace-module trace-module/root "lib-out"))
(calls 0)
(msg (τ (set! calls (add1 calls)) (list "x=" 1 " " 'y "\n"))))
(⊦= "" (with-output-to-string (τ (trace m msg))))
(⊦= 0 calls)
(dynamic-wind
(τ (trace-on! m))
(τ (⊦= "x=1 y\n" (with-output-to-string (τ (trace m msg))))
(⊦= "x=1 y\n"
(with-error-output-to-string
(τ (parameterize ((trace-to 'err)) (trace m msg)))))
(⊦= ""
(with-output-to-string
(τ (parameterize ((trace-to 'null)) (trace m msg)))))
(⊦= "plain"
(call-with-output-string
(λ (port)
(parameterize ((trace-to port)) (trace m (τ "plain"))))))
(⊦= 50000
(string-length
(call-with-output-string
(λ (port)
(parameterize
((trace-to port))
(trace m (τ (make-list 50000 "a"))))))))
(⊦= 3 calls)
(⊦raises (exn cml trace) (trace-to 42))
(⊦= "t1\nt2\nmain\n"
(with-output-to-string
(τ (run-cml
(τ (join-all
(map (λ (s) (spawn (τ (trace m (τ (list s "\n"))))))
'("t1" "t2")))
(trace m (τ '("main\n")))
(cml/shutdown))))))
(⊦= '("to-ch 1" "to-mb 2")
(run/value
(τ (let ((ch (make-channel)) (mb (make-mailbox)))
(spawn (τ (parameterize
((trace-to ch))
(trace m (τ '("to-ch " 1))))))
(parameterize ((trace-to mb)) (trace m (τ '("to-mb " 2))))
(list (recv ch) (mailbox-recv mb))))))
(let1 (mb (make-mailbox))
(parameterize ((trace-to mb)) (trace m msg))))
(τ (trace-off! m)))))
((eta 0.007) (memory #(12582912 3293232 1048576)) (stdout "") (stderr ""))
test/trace-file: pass
(define (test/trace-file _)
(let* ((m (trace-module trace-module/root "lib-file"))
(file (create-temporary-file ".txt")))
(dynamic-wind
void
(τ (trace-on! m)
(parameterize
((trace-to file))
(run-cml
(τ (trace m (τ '("one\n")))
(trace m (τ '("two\n")))
(cml/shutdown)))
(run-cml (τ (trace m (τ '("three\n"))) (cml/shutdown))))
(⊦= "one\ntwo\nthree\n"
(with-input-from-file file (τ (read-string #f))))
(let* ((err #f)
(out (with-output-to-string
(τ (set! err
(with-error-output-to-string
(τ (parameterize
((trace-to "/nonexistent-dir/x/trace.txt"))
(trace m (τ '("lost\n")))
(trace m (τ '("found\n")))))))))))
(⊦= "lost\nfound\n" out)
(⊨ (and (substring-index "unable to open" err) #t))))
(τ (trace-close-files!) (trace-off! m) (delete-file* file)))))
((eta 0.001) (memory #(12582912 3294608 1048576)) (stdout "") (stderr ""))
test/trace-file-closed-after-the-servers: pass
(define (test/trace-file-closed-after-the-servers _)
(let* ((m (trace-module trace-module/root "lib-file-servers"))
(file (create-temporary-file ".txt")))
(dynamic-wind
void
(τ (trace-on! m)
(cml/log-server!
"trace-test-server"
void
(τ (trace m (τ '("server shut\n")))))
(parameterize
((trace-to file))
(run-cml (τ (trace m (τ '("in run\n"))) (cml/shutdown))))
(⊦= '() %trace-files)
(⊦= "in run\nserver shut\n"
(with-input-from-file file (τ (read-string #f)))))
(τ (cml/unlog-server! "trace-test-server")
(trace-close-files!)
(trace-off! m)
(delete-file* file)))))
((eta 0.0) (memory #(12582912 3158400 1048576)) (stdout "") (stderr ""))
test/watch: pass
(define (test/watch _)
(let1 (out (with-output-to-string
(τ (⊦= '(#t #f done)
(cdr (run/log
(τ (let* ((c (make-channel)) (t (spawn (τ (recv c)))))
(watch "worker" t)
(note! (watched? t))
(send c 'die)
(cml/yield)
(note! (watched? t))
(note! 'done)
(cml/shutdown)))))))))
(⊦= "WARNING! Watched thread worker [000003] has died.\n" out)))
((eta 0.0) (memory #(12582912 3160304 1048576)) (stdout "") (stderr ""))
test/watch/many-threads: pass
(define (test/watch/many-threads _)
(define (cost n)
(let1 (t0 (current-process-milliseconds))
(⊦= ""
(with-output-to-string
(τ (run-cml
(τ (let* ((iv (make-ivar))
(ts (map (λ (i) (spawn (τ (ivar-get iv)))) (iota n))))
(for-each (λ (t) (watch "w" t)) ts)
(unless (every watched? ts) (print "not watched"))
(ivar-put! iv #t)
(join-all ts)
(cml/yield)
(when (any watched? ts) (print "still watched"))
(cml/shutdown)))))))
(- (current-process-milliseconds) t0)))
(dynamic-wind
(τ (trace-off! trace-watcher))
(τ (let* ((small (max 20 (cost 400))) (large (cost 4000)))
(⊨ (< large (* 30 small)))))
(τ (trace-on! trace-watcher))))
((eta 0.069) (memory #(25165824 3897616 1048576)) (stdout "") (stderr ""))
test/unwatch: pass
(define (test/unwatch _)
(⊦= ""
(with-output-to-string
(τ (run-cml
(τ (let* ((c (make-channel))
(t1 (spawn (τ (recv c))))
(t2 (spawn (τ (recv c)))))
(watch "t1" t1)
(cml/yield)
(unwatch t1)
(watch "t2" t2)
(unwatch t2)
(unwatch t2)
(send c 1)
(send c 2)
(cml/sleep 0)
(cml/yield)
(cml/shutdown)))))))
(dynamic-wind
(τ (trace-off! trace-watcher))
(τ (⊦= ""
(with-output-to-string
(τ (run-cml
(τ (watch "quiet" (spawn void))
(cml/yield)
(cml/shutdown)))))))
(τ (trace-on! trace-watcher))))
((eta 0.0) (memory #(25165824 3163160 1048576)) (stdout "") (stderr ""))
test/watch/after-death: pass
(define (test/watch/after-death _)
(let1 (out (with-output-to-string
(τ (⊦= '(#f #f #f)
(run/value
(τ (let* ((ts (map (λ (i) (spawn (τ (cml/yield)))) '(1 2 3)))
(c (make-channel))
(u (spawn (τ (recv c)))))
(for-each (λ (t n) (watch n t)) ts '("a" "b" "c"))
(watch "u" u)
(watch "u-again" u)
(join-all ts)
(cml/yield)
(unwatch (car ts))
(send c 'go)
(sync (join-evt u))
(cml/yield)
(map watched? (list (car ts) u (cadr ts))))))))))
(⊦= '("a" "b" "c" "u-again")
(map (λ (l) (cadddr (string-split l " ")))
(filter
(λ (l) (substring-index "WARNING" l))
(string-split out "\n"))))))
((eta 0.0) (memory #(25165824 3164112 1048576)) (stdout "") (stderr ""))
test/watch/exception: pass
(define (test/watch/exception _)
(let* ((err #f)
(out (with-output-to-string
(τ (set! err
(with-error-output-to-string
(τ (run-cml
(τ (let1 (t (spawn (τ (cml/yield) (error "crash"))))
(watch "crasher" t)
(sync (join-evt t))
(cml/yield)
(cml/shutdown)))))))))))
(⊨ (and (substring-index "Watched thread crasher" out) #t))
(⊨ (and (substring-index "crash" err) #t))))
((eta 0.0) (memory #(25165824 3165464 1048576)) (stdout "") (stderr ""))
test/uncaught/default: pass
(define (test/uncaught/default _)
(⊨ (eq? trace-exn-handler (default-exn-handler)))
(let1 (err (with-error-output-to-string
(τ (⊦= '(joined)
(cdr (run/log
(τ (sync (join-evt (spawn (τ (error "kaboom" 7)))))
(note! 'joined)
(cml/shutdown))))))))
(⊨ (and (substring-index "kaboom" err) #t))
(⊨ (and (substring-index "[000003]" err) #t)))
(let1 (err (with-error-output-to-string
(τ (uncaught-default-handler
(current-tid)
(make-property-condition 'exn 'message "direct")))))
(⊨ (and (substring-index "[-000001] died of an uncaught exception" err)
#t))
(⊨ (and (substring-index "direct" err) #t))))
((eta 0.0) (memory #(25165824 3166168 1048576)) (stdout "") (stderr ""))
test/uncaught/handlers: pass
(define (test/uncaught/handlers _)
(dynamic-wind
void
(τ (add-uncaught-handler!
(λ (tid e)
(and (condition? e)
((condition-predicate 'type) e)
(note! (list 'h1 (tid->string tid)))
#t)))
(add-uncaught-handler!
(λ (tid e) (and (symbol? e) (note! (list 'h2 e)) #t)))
(add-uncaught-handler!
(λ (tid e) (and (eq? e 'bad-handler) (error "handler fails"))))
(set-uncaught-handler!
(λ (tid e) (note! (list 'default (if (symbol? e) e (message-of e))))))
(⊦= '((h2 sym)
j1
(h1 "[000004]")
j2
(default "other")
j3
(default bad-handler)
j4)
(cdr (run/log
(τ (let1 (die (λ (thunk j)
(sync (join-evt (spawn thunk)))
(note! j)))
(die (τ (abort 'sym)) 'j1)
(die (τ (car 1)) 'j2)
(die (τ (error "other")) 'j3)
(die (τ (abort 'bad-handler)) 'j4)
(cml/shutdown))))))
(set-uncaught-handler! (λ (tid e) (error "default fails")))
(⊦= '(alive)
(cdr (run/log
(τ (sync (join-evt (spawn (τ (error "x")))))
(note! 'alive)
(cml/shutdown)))))
(⊦= '(own)
(cdr (run/log
(τ (parameterize
((default-exn-handler (λ (e) (note! 'own))))
(sync (join-evt (spawn (τ (abort 'sym))))))
(cml/shutdown))))))
reset-uncaught-handlers!)
(let1 (err (with-error-output-to-string
(τ (run-cml
(τ (sync (join-evt (spawn (τ (error "after reset")))))
(cml/shutdown))))))
(⊨ (and (substring-index "after reset" err) #t))))
((eta 0.0) (memory #(25165824 3167296 1048576)) (stdout "") (stderr ""))
I/O: ports#
Concurrent ML: events on ports, processes and sockets
Input events read without blocking the process into a per-port side buffer that only a commit consumes, see the IO section of aux.cml.scm.
test/line-reader/slow-writer: pass
(define (test/line-reader/slow-writer _)
(⊦= '(timeout "hello" "world" "tail" #!eof)
(with-pipe
(λ (p out)
(log-of
(τ (spawn (τ (fd-write! out "hel")
(cml/sleep 0.05)
(fd-write! out "lo\nwor")
(cml/sleep 0.02)
(fd-write! out "ld\ntail")
(file-close out)))
(note! (sync/timeout (input-line-evt p) 0.02 'timeout))
(note! (sync (input-line-evt p)))
(note! (sync (input-line-evt p)))
(note! (sync (input-line-evt p)))
(note! (sync (input-line-evt p)))))))))
((eta 0.001) (memory #(12582912 3866616 1048576)) (stdout "") (stderr ""))
test/stdio-buffered-data: pass
(define (test/stdio-buffered-data _)
(⊦= '(#\a "bc" "de" #\f "ghi" #\j timeout)
(with-pipe
(λ (p out)
(fd-write! out "abc\ndefghij")
(let1 (c (read-char p))
(log-of
(τ (note! c)
(note! (sync/timeout (input-line-evt p) 1 'stuck))
(note! (sync/timeout (input-string-evt p 2) 1 'stuck))
(note! (sync/timeout (input-char-evt p) 1 'stuck))
(note! (sync/timeout (input-string-evt p 3) 1 'stuck))
(note! (sync/timeout (input-char-evt p) 1 'stuck))
(note! (sync/timeout (input-char-evt p) 0.02 'timeout)))))))))
((eta 0.002) (memory #(12582912 3858936 1048576)) (stdout "") (stderr ""))
test/char-peek-string-all: pass
(define (test/char-peek-string-all _)
(⊦= '(#\x #\x "" "yz1" "23\nrest\n" "" #!eof #!eof #!eof)
(with-pipe
(λ (p out)
(fd-write! out "xyz123\nrest\n")
(file-close out)
(log-of
(τ (note! (sync (peek-char-evt p)))
(note! (sync (input-char-evt p)))
(note! (sync (input-string-evt p 0)))
(note! (sync (input-string-evt p 3)))
(note! (sync (input-all-evt p)))
(note! (sync (input-all-evt p)))
(note! (sync (input-char-evt p)))
(note! (sync (peek-char-evt p)))
(note! (sync (input-line-evt p)))))))))
((eta 0.0) (memory #(12582912 3866448 1048576)) (stdout "") (stderr ""))
test/eof-partial-string: pass
(define (test/eof-partial-string _)
(⊦= '("ab" #!eof)
(with-pipe
(λ (p out)
(fd-write! out "ab")
(file-close out)
(log-of
(τ (note! (sync (input-string-evt p 5)))
(note! (sync (input-string-evt p 5)))))))))
((eta 0.0) (memory #(12582912 3870576 1048576)) (stdout "") (stderr ""))
test/select-two-pipes: pass
(define (test/select-two-pipes _)
(⊦= '((two "b") (one "a") (two "c"))
(with-pipe
(λ (p1 out1)
(with-pipe
(λ (p2 out2)
(define (either)
(select
(wrap (input-line-evt p1) (λ (l) (list 'one l)))
(wrap (input-line-evt p2) (λ (l) (list 'two l)))))
(log-of
(τ (spawn (τ (cml/sleep 0.02) (fd-write! out2 "b\n")))
(note! (either))
(fd-write! out1 "a\n")
(note! (either))
(fd-write! out2 "c\n")
(note! (either))))))))))
((eta 0.001) (memory #(12582912 3869408 1048576)) (stdout "") (stderr ""))
test/losing-branch-keeps-input: pass
(define (test/losing-branch-keeps-input _)
(define (lose-then-read mk data)
(with-pipe
(λ (p out)
(fd-write! out data)
(log-of
(τ (note! (select (mk p) (always-evt 'lost)))
(cml/sleep 0.01)
(note! (sync (mk p))))))))
(⊦= '(lost #\a) (lose-then-read input-char-evt "ab"))
(⊦= '(lost "ab") (lose-then-read input-line-evt "ab\nc\n"))
(⊦= '(lost "abc") (lose-then-read (λ (p) (input-string-evt p 3)) "abcdef")))
((eta 0.001) (memory #(12582912 3875936 1048576)) (stdout "") (stderr ""))
test/line-terminators: pass
(define (test/line-terminators _)
(⊦= '("GET / HTTP/1.0" "Host: x" "" "a" "b" "c" #!eof)
(with-pipe
(λ (p out)
(fd-write! out "GET / HTTP/1.0\r\nHost: x\r\n\r\na\rb\nc\r")
(file-close out)
(log-of
(τ (for-each
(λ (i) (note! (sync (input-line-evt p))))
(iota 7)))))))
(⊦= '("x" "y")
(with-pipe
(λ (p out)
(log-of
(τ (spawn (τ (fd-write! out "x\r")
(cml/sleep 0.02)
(fd-write! out "\ny\n")))
(note! (sync (input-line-evt p)))
(note! (sync (input-line-evt p)))))))))
((eta 0.001) (memory #(12582912 3862480 1048576)) (stdout "") (stderr ""))
test/readers-share-a-port: pass
(define (test/readers-share-a-port _)
(⊦= '((a "1") (b "2") (a "3"))
(with-pipe
(λ (p out)
(log-of
(τ (let ((ra (make-mailbox)) (rb (make-mailbox)))
(spawn (τ (mailbox-send! ra (sync (input-line-evt p)))
(mailbox-send! ra (sync (input-line-evt p)))))
(spawn (τ (mailbox-send! rb (sync (input-line-evt p)))))
(fd-write! out "1\n2\n")
(note! (list 'a (mailbox-recv ra)))
(note! (list 'b (mailbox-recv rb)))
(fd-write! out "3\n")
(note! (list 'a (mailbox-recv ra))))))))))
((eta 0.0) (memory #(12582912 3863808 1048576)) (stdout "") (stderr ""))
test/port-survives-a-run-boundary: pass
(define (test/port-survives-a-run-boundary _)
(⊦= '((timeout) ("abcd"))
(with-pipe
(λ (p out)
(fd-write! out "ab")
(list (log-of
(τ (note! (sync/timeout (input-line-evt p) 0.01 'timeout))
(spawn (τ (sync (input-line-evt p))))
(cml/sleep 0.01)))
(begin
(fd-write! out "cd\n")
(log-of
(τ (note! (sync/timeout (input-line-evt p) 1 'stuck))))))))))
((eta 0.001) (memory #(12582912 3875912 1048576)) (stdout "") (stderr ""))
test/select-input-vs-timeout-keeps-waiting: pass
(define (test/select-input-vs-timeout-keeps-waiting _)
(⊦= '(timeout "late")
(with-pipe
(λ (p out)
(log-of
(τ (note! (sync/timeout (input-line-evt p) 0.01 'timeout))
(spawn (τ (cml/sleep 0.03) (fd-write! out "late\n")))
(note! (sync (input-line-evt p)))))))))
((eta 0.001) (memory #(12582912 3871832 1048576)) (stdout "") (stderr ""))
test/read-errors-reach-the-syncing-thread: pass
(define (test/read-errors-reach-the-syncing-thread _)
(⊦= '(raised)
(receive
(in out)
(create-pipe)
(let1 (p (open-input-file* in))
(close-input-port p)
(file-close out)
(log-of
(τ (note! (condition-case
(sync (input-char-evt p))
(e () 'raised)))))))))
((eta 0.001) (memory #(12582912 3878216 1048576)) (stdout "") (stderr ""))
test/writer-and-reader-over-a-full-pipe: pass
(define (test/writer-and-reader-over-a-full-pipe _)
(let1 (big (make-string 200000 #\z))
(receive
(in out)
(create-pipe)
(let* ((p (open-input-file* in))
(o (open-output-file* out))
(log (log-of
(τ (let1 (w (spawn (τ (sync (write-string-evt o big))
(note! 'written)
(close-output-port o))))
(let1 (s (sync (input-all-evt p)))
(note! (list (string-length s) (equal? s big)))))))))
(close-input-port p)
(⊦= '(written (200000 #t)) log)))))
((eta 0.05) (memory #(12582912 4278616 1048576)) (stdout "") (stderr ""))
test/output-evt: pass
(define (test/output-evt _)
(receive
(in out)
(create-pipe)
(let* ((p (open-input-file* in))
(o (open-output-file* out))
(log (log-of
(τ (note! (eq? o (sync (output-evt o))))
(sync (write-string-evt o "λ-line\n"))
(note! (sync (input-line-evt p)))))))
(close-input-port p)
(close-output-port o)
(⊦= '(#t "λ-line") log))))
((eta 0.0) (memory #(12582912 4081088 1048576)) (stdout "") (stderr ""))
test/commit-survives-an-escape: pass
(define (test/commit-survives-an-escape _)
(⊦= '(#\a #\b)
(with-pipe
(λ (p out)
(fd-write! out "ab")
(map (λ (i)
(call-with-current-continuation
(λ (k) (run-cml (τ (k (sync (input-char-evt p))))))))
'(1 2))))))
((eta 0.0) (memory #(12582912 4084920 1048576)) (stdout "") (stderr ""))
test/forcing-raises-releases-the-port: pass
(define (test/forcing-raises-releases-the-port _)
(⊦= '(raised #\a "bc")
(with-pipe
(λ (p out)
(fd-write! out "abc\n")
(log-of
(τ (note! (condition-case
(select
(input-char-evt p)
(guard (τ (error "guard failed"))))
(e () 'raised)))
(note! (sync/timeout (input-char-evt p) 1 'stuck))
(note! (sync/timeout (input-line-evt p) 1 'stuck))))))))
((eta 0.001) (memory #(12582912 4088224 1048576)) (stdout "") (stderr ""))
test/input-evt: pass
(define (test/input-evt _)
(⊦= '("abc" timeout "defgh" "rest" "xy" #!eof lost "12" "34")
(with-pipe
(λ (p out)
(fd-write! out "abc")
(log-of
(τ (note! (sync/timeout (input-evt p) 1 'stuck))
(note! (sync/timeout (input-evt p) 0.02 'timeout))
(spawn (τ (cml/sleep 0.01)
(fd-write! out "def")
(fd-write! out "gh\nrest")))
(note! (sync/timeout (input-line-evt p) 1 'stuck))
(note! (sync/timeout (input-evt p) 1 'stuck))
(fd-write! out "xy")
(file-close out)
(note! (sync/timeout (input-evt p) 1 'stuck))
(note! (sync/timeout (input-evt p) 1 'stuck))
(let1 (p (open-input-string "12"))
(note! (select (input-evt p) (always-evt 'lost)))
(cml/sleep 0.01)
(note! (sync (input-evt p))))
(let* ((ch (make-channel)) (in (open-channel-input-port ch)))
(spawn (τ (send ch "3") (send ch "4") (send ch #!eof)))
(note! (let loop ((acc ""))
(let1 (s (sync/timeout (input-evt in) 1 'stuck))
(if (string? s) (loop (string-append acc s)) acc))))))))))
(let1 (r (with-pipe
(λ (p out)
(log-of
(τ (let* ((o (open-output-file* out))
(w (spawn (τ (sync (write-string-evt o (make-string 10000 #\z)))
(close-output-port o)))))
(let1 (all-in?
(sync/timeout
(wrap (join-evt w) (λ ignored #t))
0.5
#f))
(let loop ((n 0) (k 0) (first #f))
(let1 (s (sync/timeout (input-evt p) 1 'stuck))
(if (string? s)
(loop (+ n (string-length s))
(add1 k)
(or first (string-length s)))
(note! (list n k s first all-in?))))))))))))
(⊦= 10000 (list-ref (car r) 0))
(⊦= #!eof (list-ref (car r) 2))
(⊨ (>= (list-ref (car r) 3) 4096))
(when (list-ref (car r) 4) (⊨ (<= 2 (list-ref (car r) 1) 3)))))
((eta 0.004) (memory #(12582912 3928336 1048576)) (stdout "") (stderr ""))
test/stream-drain-leaves-the-file-status-flags-alone: pass
(define (test/stream-drain-leaves-the-file-status-flags-alone _)
(define (nonblock? fd)
(not (zero? (bitwise-and open/nonblock (file-control fd fcntl/getfl)))))
(⊦= '(("abc\n" #f #f (#f)) (() #f #f))
(with-pipe
(λ (p out)
(fd-write! out "abc\n")
(let* ((fd (port->fileno p))
(drain (%stream-drainer fd #t 'utf-8))
(seen '()))
(list (receive
(acc eof? more?)
(drain p
(λ (c)
(unless
(memv (nonblock? fd) seen)
(set! seen (cons (nonblock? fd) seen)))
#f))
(list (list->string (reverse acc)) eof? more? seen))
(receive
(acc eof? more?)
(drain p (λ (c) #f))
(list acc eof? more?))))))))
((eta 0.001) (memory #(12582912 3933584 1048576)) (stdout "") (stderr ""))
test/port-encodings: pass
(define (test/port-encodings _)
(define (direct f enc)
(with-input-from-file
f
(τ (let* ((a (read-line)) (b (read-line)) (c (read-line))) (list a b c)))
enc))
(define (events f enc)
(let1 (p (open-input-file f enc))
(begin1
(run/value
(τ (let* ((a (sync (input-line-evt p)))
(b (sync (input-line-evt p)))
(c (sync (input-line-evt p))))
(list a b c))))
(close-input-port p))))
(let1 (f (create-temporary-file))
(with-output-to-file f (τ (display "café\nÿx\n")) #:latin-1)
(⊦= '("café" "ÿx" #!eof) (direct f #:latin-1))
(⊦= (direct f #:latin-1) (events f #:latin-1))
(⊦= (direct f #:binary) (events f #:binary))
(delete-file f)))
((eta 0.001) (memory #(12582912 3887128 1048576)) (stdout "") (stderr ""))
test/utf-8-split-across-writes: pass
(define (test/utf-8-split-across-writes _)
(⊦= '("café€𝄞" #!eof)
(with-pipe
(λ (p out)
(log-of
(τ (spawn (τ (for-each
(λ (bytes)
(file-write out (apply bytevector bytes))
(cml/sleep 0.01))
'((99 97 102 195)
(169 226)
(130 172 240 157)
(132 158 10)))
(file-close out)))
(note! (sync (input-line-evt p)))
(note! (sync (input-line-evt p)))))))))
((eta 0.001) (memory #(12582912 3888000 1048576)) (stdout "") (stderr ""))
test/utf-8-split-after-a-direct-read: pass
(define (test/utf-8-split-after-a-direct-read _)
(⊦= '("café!" "x" #!eof)
(with-pipe
(λ (p out)
(fd-write! out "greeting\n")
(read-line p)
(file-write out (bytevector 99 97 102 195))
(log-of
(τ (spawn (τ (cml/sleep 0.05)
(file-write out (bytevector 169 33 10 120 10))
(file-close out)))
(note! (sync/timeout (input-line-evt p) 1 'stuck))
(note! (sync/timeout (input-line-evt p) 1 'stuck))
(note! (sync/timeout (input-line-evt p) 1 'stuck)))))))
(⊦= '("rest" "é" timeout)
(with-pipe
(λ (p out)
(fd-write! out "greeting\nrest\né\n")
(read-line p)
(log-of
(τ (note! (sync/timeout (input-line-evt p) 1 'stuck))
(note! (sync/timeout (input-line-evt p) 1 'stuck))
(note! (sync/timeout (input-line-evt p) 0.02 'timeout))))))))
((eta 0.0) (memory #(12582912 3881832 1048576)) (stdout "") (stderr ""))
test/waiting-reader-does-not-leak: pass
(define (test/waiting-reader-does-not-leak _)
(define p (make-input-port (λ () #!eof) (λ () #f) void))
(define (used) (gc #t) (vector-ref (memory-statistics) 1))
(let1 (r (run/log
(τ (spawn (τ (sync (input-char-evt p))))
(cml/sleep 0.2)
(let1 (m0 (used))
(cml/sleep 0.6)
(cml/shutdown (- (used) m0))))))
(⊨ (< (car r) 100000))))
((eta 0.024) (memory #(12582912 3885464 1048576)) (stdout "") (stderr ""))
test/poll-raises-releases-the-port: pass
(define (test/poll-raises-releases-the-port _)
(⊦= '(raised "line1" "line2")
(with-pipe
(λ (p out)
(fd-write! out "line1\nline2\n")
(log-of
(τ (note! (condition-case
(select
(input-line-evt p)
(mvar-take-evt 'not-an-mvar))
(e () 'raised)))
(note! (sync/timeout (input-line-evt p) 1 'stuck))
(note! (sync/timeout (input-line-evt p) 1 'stuck))))))))
((eta 0.0) (memory #(12582912 3890592 1048576)) (stdout "") (stderr ""))
test/concurrent-write-string-evts-do-not-interleave: pass
(define (test/concurrent-write-string-evts-do-not-interleave _)
(let1 (big (string-append (make-string 100000 #\A) "\n"))
(receive
(in out)
(create-pipe)
(let* ((p (open-input-file* in))
(o (open-output-file* out))
(log (log-of
(τ (spawn (τ (sync (write-string-evt o big))
(note! 'a-written)))
(cml/sleep 0.05)
(spawn (τ (sync (write-string-evt o "BBBB\n"))
(note! 'b-written)))
(cml/sleep 0.05)
(do ((i 0 (add1 i)))
((= i 2))
(let1 (l (sync/timeout (input-line-evt p) 30 'timeout))
(note! (if (member l '("BBBB" timeout))
l
(list (string-length l)
(and (memv #\B (string->list l)) #t))))))
(note! (sync/timeout
(input-line-evt p)
0.1
'nothing-more))))))
(close-input-port p)
(close-output-port o)
(⊦= '((100000 #f) "BBBB" nothing-more)
(filter (λ (x) (not (memq x '(a-written b-written)))) log))))))
((eta 0.031) (memory #(12582912 3985256 1048576)) (stdout "") (stderr ""))
test/input-events-on-a-closed-port: pass
(define (test/input-events-on-a-closed-port _)
(define (message thunk)
(condition-case
(begin (thunk) 'not-raised)
(e (exn) (get-condition-property e 'exn 'message))))
(receive
(r w)
(create-pipe)
(let ((in (open-input-file* r)) (out (open-output-file* w)))
(⊦= '("one" "port is closed" #t "secret" "port is closed" #t "more")
(log-of
(τ (display "one\n" out)
(flush-output out)
(note! (sync (input-line-evt in)))
(close-input-port in)
(close-output-port out)
(receive
(r2 w2)
(create-pipe)
(let1 (in2 (open-input-file* r2))
(fd-write! w2 "secret\n")
(note! (message
(τ (sync/timeout (input-line-evt in) 0.2 'timeout))))
(note! (equal? (list r w) (list r2 w2)))
(note! (sync/timeout (input-line-evt in2) 0.2 'timeout))
(let* ((res (make-ivar))
(t (spawn (τ (ivar-put!
res
(message (τ (sync (input-line-evt in2)))))))))
(cml/sleep 0.02)
(close-input-port in2)
(file-close w2)
(receive
(r3 w3)
(create-pipe)
(let1 (in3 (open-input-file* r3))
(fd-write! w3 "more\n")
(note! (sync/timeout
(ivar-get-evt res)
0.5
'still-waiting))
(note! (equal? (list r2 w2) (list r3 w3)))
(note! (sync/timeout (input-line-evt in3) 0.2 'timeout))
(close-input-port in3)
(file-close w3))))))))))))
((eta 0.001) (memory #(12582912 3996768 1048576)) (stdout "") (stderr ""))
test/closed-port-with-a-reused-descriptor: pass
(define (test/closed-port-with-a-reused-descriptor _)
(define (message thunk)
(condition-case
(begin (thunk) 'not-raised)
(e (exn) (get-condition-property e 'exn 'message))))
(receive
(r w)
(create-pipe)
(let ((in (open-input-file* r)) (new '()))
(⊦= '("port is closed" #t)
(log-of
(τ (let1 (res (make-ivar))
(spawn (τ (ivar-put!
res
(message (τ (sync (input-line-evt in)))))))
(cml/sleep 0.02)
(close-input-port in)
(receive
(r2 w2)
(create-pipe)
(set! new (list r2 w2))
(note! (sync/timeout
(ivar-get-evt res)
1
'still-waiting))
(note! (= r r2)))))))
(file-close w)
(for-each file-close new)))
(receive
(r w)
(create-pipe)
(let ((out (open-output-file* w)) (new '()))
(⊦= '("port is closed" #t)
(log-of
(τ (let1 (res (make-ivar))
(spawn (τ (ivar-put!
res
(message
(τ (sync (write-string-evt out (make-string 200000 #\x))))))))
(cml/sleep 0.05)
(close-output-port out)
(receive
(r2 w2)
(create-pipe)
(set! new (list r2 w2))
(note! (sync/timeout
(ivar-get-evt res)
1
'still-waiting))
(note! (= w r2)))))))
(file-close r)
(for-each file-close new))))
((eta 0.003) (memory #(12582912 3882192 1048576)) (stdout "") (stderr ""))
test/closed-port-with-a-reused-descriptor-before-the-first-write: pass
(define (test/closed-port-with-a-reused-descriptor-before-the-first-write _)
(define (message thunk)
(condition-case
(begin (thunk) 'not-raised)
(e (exn) (get-condition-property e 'exn 'message))))
(define (fill! fd)
(let1 (flags (file-control fd fcntl/getfl))
(file-control fd fcntl/setfl (bitwise-ior flags open/nonblock))
(let loop ()
(when (positive?
(condition-case
(file-write fd (make-bytevector 4096 65))
(e () 0)))
(loop)))
(file-control fd fcntl/setfl flags)))
(for-each
(λ (evt)
(for-each
(λ (reuse?)
(receive
(r w)
(create-pipe)
(fill! w)
(let ((out (open-output-file* w)) (new '()))
(⊦= (if reuse? '("port is closed" #t) '("port is closed"))
(log-of
(τ (let1 (res (make-ivar))
(spawn (τ (ivar-put! res (message (τ (sync (evt out)))))))
(cml/sleep 0.05)
(close-output-port out)
(if reuse?
(receive
(r2 w2)
(create-pipe)
(set! new (list r2 w2))
(note! (sync/timeout
(ivar-get-evt res)
1
'still-waiting))
(note! (= w r2)))
(note! (sync/timeout
(ivar-get-evt res)
1
'still-waiting)))))))
(file-close r)
(for-each file-close new))))
'(#t #f)))
(list (λ (o) (write-string-evt o "hello")) output-evt)))
((eta 0.003) (memory #(12582912 3887168 1048576)) (stdout "") (stderr ""))
test/input-in-small-pieces-in-linear-time: pass
(define (test/input-in-small-pieces-in-linear-time _)
(define (reading-time n make-evt)
(let1 (t0 (current-process-milliseconds))
(⊦= (* 10 n)
(run/value
(τ (let1 (ch (make-channel))
(spawn (τ (for-each
(λ (i) (send ch "abcdefghi."))
(iota n))
(send ch "\n")
(send ch #!eof)))
(string-length
(sync (make-evt (open-channel-input-port ch))))))))
(- (current-process-milliseconds) t0)))
(for-each
(λ (make-evt)
(let* ((small (max 20 (reading-time 500 make-evt)))
(large (reading-time 4000 make-evt)))
(⊨ (< large (* 20 small)))))
(list input-line-evt
(λ (p)
(wrap (input-all-evt p)
(λ (s) (substring s 0 (sub1 (string-length s)))))))))
((eta 0.123) (memory #(12582912 4011920 1048576)) (stdout "") (stderr ""))
test/a-flooding-stream-does-not-freeze-the-others: pass
(define (test/a-flooding-stream-does-not-freeze-the-others _)
(receive
(in out pid)
(cml/execute "/bin/sh" '("-c" "yes | tr -d '\\n'"))
(let1 (log (log-of
(τ (spawn (τ (let loop ()
(cml/sleep 0.01)
(note! 'tick)
(loop))))
(note! (sync/timeout (input-line-evt in) 0.5 'timeout))
(note! (sync/timeout (input-all-evt in) 0.2 'timeout)))))
(process-signal pid)
(close-input-port in)
(close-output-port out)
(process-wait pid)
(⊦= '(timeout timeout)
(filter symbol? (filter (λ (x) (not (eq? x 'tick))) log)))
(⊨ (>= (count (λ (x) (eq? x 'tick)) log) 3)))))
((eta 0.713) (memory #(27262984 7039568 1048576)) (stdout "") (stderr ""))
test/write-to-a-closed-pipe-raises: pass
(define (test/write-to-a-closed-pipe-raises _)
(receive
(in out)
(create-pipe)
(file-close in)
(let1 (o (open-output-file* out))
(⊦= (list errno/pipe errno/pipe errno/pipe)
(run/value
(τ (map (λ (s)
(condition-case
(begin (sync (write-string-evt o s)) 'no-error)
(e (exn i/o)
(get-condition-property e 'exn 'errno #f))))
(list "line\n" "again\n" (make-string 1000 #\x))))))
(condition-case (close-output-port o) (ignored () (void)))))
(for-each
(λ (enc)
(receive
(in out)
(create-pipe)
(file-close in)
(let1 (o (open-output-file* out))
(set! (port-encoding o) enc)
(⊦= (list enc errno/pipe errno/pipe)
(cons enc
(run/value
(τ (map (λ (s)
(condition-case
(begin (sync (write-string-evt o s)) 'no-error)
(e (exn i/o)
(get-condition-property e 'exn 'errno #f))))
(list "line\n" "café\n"))))))
(condition-case (close-output-port o) (ignored () (void))))))
'(latin-1 binary))
(let* ((s "aé€")
(message (λ (e) (get-condition-property e 'exn 'message #f)))
(outcome
(λ (enc write!)
(receive
(in out)
(create-pipe)
(let ((i (open-input-file* in)) (o (open-output-file* out)))
(set! (port-encoding o) enc)
(let1 (raised (write! o))
(condition-case
(close-output-port o)
(ignored () (void)))
(set! (port-encoding i) 'binary)
(let1 (bytes (let loop ((acc '()))
(let1 (b (read-byte i))
(if (eof-object? b)
(reverse acc)
(loop (cons b acc))))))
(close-input-port i)
(list raised bytes))))))))
(for-each
(λ (enc)
(⊦= (outcome
enc
(λ (o)
(condition-case
(begin (display s o) (flush-output o) #f)
(e (exn) (message e)))))
(outcome
enc
(λ (o)
(run/value
(τ (condition-case
(begin (sync (write-string-evt o s)) #f)
(e (exn) (message e)))))))))
'(latin-1 binary utf-8))
(⊦= '(#f (97 195 169 226 130 172))
(outcome
'utf-8
(λ (o) (run/value (τ (sync (write-string-evt o s)) #f)))))))
((eta 0.001) (memory #(27262984 6946096 1048576)) (stdout "") (stderr ""))
test/writer-dying-with-the-write-lock-releases-it: pass
(define (test/writer-dying-with-the-write-lock-releases-it _)
(receive
(in out)
(create-pipe)
(let ((p (open-input-file* in)) (o (open-output-file* out)) (armed #f))
(let1 (log (log-of
(τ (parameterize
((default-exn-handler (λ (e) (note! 'a-died))))
(spawn (τ (dynamic-wind
(τ (when armed
(set! armed #f)
(error "before thunk raised")))
(τ (sync (write-string-evt o (make-string 100000 #\A)))
(note! 'a-written))
void))))
(cml/sleep 0.05)
(set! armed #t)
(spawn (τ (sync (write-string-evt o "BBBB\n"))
(note! 'b-written)))
(note! (let1 (l (sync/timeout (input-line-evt p) 2 'timeout))
(if (string? l)
(substring l (- (string-length l) 4))
l))))))
(close-input-port p)
(close-output-port o)
(⊨ (and (memq 'a-died log) #t))
(⊨ (and (memq 'b-written log) #t))
(⊨ (and (member "BBBB" log) #t))
(⊨ (not (memq 'a-written log)))))))
((eta 0.018) (memory #(27262984 7172088 1048576)) (stdout "") (stderr ""))
test/idle-waiters-cost-nothing: pass
(define (test/idle-waiters-cost-nothing _)
(let* ((pipes (map (λ (i)
(receive
(r w)
(create-pipe)
(cons (open-input-file* r) w)))
(iota 400)))
(cpu (run/value
(τ (for-each
(λ (pw) (spawn (τ (sync (input-line-evt (car pw))))))
pipes)
(cml/sleep 0.1)
(let1 (t0 (cpu-ms)) (cml/sleep 1) (- (cpu-ms) t0))))))
(for-each (λ (pw) (close-input-port (car pw)) (file-close (cdr pw))) pipes)
(⊨ (< cpu 60))))
((eta 0.019) (memory #(27262984 6436808 1048576)) (stdout "") (stderr ""))
test/bad-arguments: pass
(define (test/bad-arguments _)
(define (message thunk)
(condition-case
(begin (thunk) 'not-raised)
(e (exn) (get-condition-property e 'exn 'message))))
(⊦= "bad count"
(message (τ (input-string-evt (open-input-string "abcdef") 2.0))))
(⊦= "bad count"
(message (τ (input-string-evt (open-input-string "abcdef") -1))))
(⊦= "cml: bad io spec, expected (fd input|output)"
(message (τ (poll-evt (list 0.0 'input)))))
(⊦= "ab"
(run/value
(τ (sync (input-string-evt (open-input-string "abcdef") 2))))))
((eta 0.0) (memory #(27262984 6447024 1048576)) (stdout "") (stderr ""))
test/polls-find-the-available-input: pass
(define (test/polls-find-the-available-input _)
(define (misses port q)
(run-cml
(τ (let loop ((i 0) (misses 0))
(if (= i 1000)
(cml/shutdown misses)
(loop (add1 i)
(if (eq? 'none (sync/timeout (input-char-evt port) 0 'none))
(add1 misses)
misses)))))
#:quantum
q))
(for-each
(λ (q) (⊦= 0 (misses (open-input-string (make-string 2000 #\a)) q)))
'(8 64 1000))
(with-pipe
(λ (p out) (fd-write! out (make-string 2000 #\b)) (⊦= 0 (misses p 64))))
(⊦= '(none #\x "yz" none)
(with-pipe
(λ (p out)
(log-of
(τ (note! (sync/timeout (input-char-evt p) 0 'none))
(fd-write! out "xyz\n")
(note! (sync/timeout (input-char-evt p) 0 'none))
(note! (sync/timeout (input-line-evt p) 0 'none))
(note! (sync/timeout (input-char-evt p) 0 'none))))))))
((eta 0.046) (memory #(27262984 6448056 1048576)) (stdout "") (stderr ""))
test/not-running: pass
(define (test/not-running _)
(with-pipe
(λ (p out) (⊦raises (exn cml not-running) (sync (input-char-evt p)))))
(⊦raises (exn cml not-running) (system-evt "true"))
(⊦raises (exn) (input-char-evt 42)))
((eta 0.0) (memory #(27262984 6453144 1048576)) (stdout "") (stderr ""))
I/O: processes#
test/system-evt/exit-codes: pass
(define (test/system-evt/exit-codes _)
(⊦= '(0 3 127 137 0)
(log-of
(τ (note! (cml/system "true"))
(note! (sync (system-evt "exit 3")))
(note! (cml/system "a-command-that-does-not-exist 2>/dev/null"))
(note! (cml/system "kill -9 $$"))
(let1 (e (system-evt "true")) (sync e) (note! (sync e)))))))
((eta 0.004) (memory #(27262984 6444584 1048576)) (stdout "") (stderr ""))
test/bad-arguments: pass
(define (test/bad-arguments _)
(⊦= '(raised raised raised raised raised raised)
(run/value
(τ (map (λ (thunk)
(condition-case
(begin (thunk) 'no-raise)
(e (exn) 'raised)))
(list (τ (system-evt 42))
(τ (cml/system 'true))
(τ (cml/execute 'ls))
(τ (cml/execute "ls" "-l"))
(τ (cml/execute "ls" '(1 2)))
(τ (cml/execute "ls" '() '(42)))))))))
((eta 0.0) (memory #(27262984 6460976 1048576)) (stdout "") (stderr ""))
test/system-evt/eager-and-selectable: pass
(define (test/system-evt/eager-and-selectable _)
(let1 (log (log-of
(τ (let1 (e (system-evt "sleep 0.05"))
(note! (select
(wrap e (λ (c) (list 'done c)))
(wrap (timeout-evt 0.01) (λ ignored 'timeout))))
(note! (sync e))))))
(⊦= '(timeout 0) log)))
((eta 0.002) (memory #(27262984 6449944 1048576)) (stdout "") (stderr ""))
test/system-evt/concurrent: pass
(define (test/system-evt/concurrent _)
(let1 (log (log-of
(τ (let* ((t0 (cml/now))
(es (map (λ (i)
(system-evt
(string-append
"sleep 0.1; exit "
(number->string i))))
(iota 3))))
(note! (map sync es))
(note! (< (- (cml/now) t0) 0.25))))))
(⊦= '((0 1 2) #t) log)))
((eta 0.003) (memory #(27262984 6452864 1048576)) (stdout "") (stderr ""))
test/execute/cat: pass
(define (test/execute/cat _)
(let1 (log (log-of
(τ (receive
(in out pid)
(cml/execute "cat")
(sync (write-string-evt out "ping\npong\n"))
(note! (sync (input-line-evt in)))
(note! (sync/timeout (input-line-evt in) 1 'stuck))
(close-output-port out)
(note! (sync (input-line-evt in)))
(note! (sync (process-evt pid)))
(close-input-port in)))))
(⊦= '("ping" "pong" #!eof (#t 0)) log)))
((eta 0.002) (memory #(27262984 4093968 1048576)) (stdout "") (stderr ""))
test/execute/close-on-exec: pass
(define (test/execute/close-on-exec _)
(⊦= '((#t 0) #!eof (#t 0))
(log-of
(τ (receive
(in1 out1 pid1)
(cml/execute "cat")
(receive
(in2 out2 pid2)
(cml/execute "cat")
(close-output-port out1)
(note! (sync/timeout (process-evt pid1) 2 'still-running))
(note! (sync/timeout (input-line-evt in1) 1 'no-eof))
(close-output-port out2)
(note! (sync (process-evt pid2)))
(close-input-port in1)
(close-input-port in2)))))))
((eta 0.002) (memory #(27262984 4096320 1048576)) (stdout "") (stderr ""))
test/process-evt/reaped-after-the-run: pass
(define (test/process-evt/reaped-after-the-run _)
(let1 (pid #f)
(⊦= 'gave-up
(run-cml
(τ (set! pid (process-run "sleep" '("0.05")))
(cml/shutdown
(sync/timeout (process-evt pid) 0.005 'gave-up)))))
(run-cml (τ (cml/sleep 0.2) (cml/shutdown)))
(⊦= 'reaped
(condition-case
(receive
(p normal? code)
(process-wait (process-id pid) #t)
(list p normal? code))
(ignored () 'reaped)))))
((eta 0.002) (memory #(13631492 4100768 1048576)) (stdout "") (stderr ""))
test/execute/closed-stdio: pass
(define (test/execute/closed-stdio _)
(flush-output (current-output-port))
(let ((in0 (duplicate-fileno 0)) (out1 (duplicate-fileno 1)))
(file-close 0)
(file-close 1)
(let1 (log (dynamic-wind
void
(τ (log-of
(τ (receive
(in out pid)
(cml/execute "cat")
(sync (write-string-evt out "ping\n"))
(note! (sync/timeout (input-line-evt in) 2 'stuck))
(close-output-port out)
(note! (sync (process-evt pid)))
(close-input-port in)))))
(τ (duplicate-fileno in0 0)
(duplicate-fileno out1 1)
(file-close in0)
(file-close out1))))
(⊦= '("ping" (#t 0)) log))))
((eta 0.001) (memory #(13631492 4096536 1048576)) (stdout "") (stderr ""))
test/system-evt/bin-sh: pass
(define (test/system-evt/bin-sh _)
(let1 (shell (get-environment-variable "SHELL"))
(set-environment-variable! "SHELL" "/bin/false")
(let1 (log (dynamic-wind
void
(τ (log-of (τ (note! (cml/system "exit 3")))))
(τ (if shell
(set-environment-variable! "SHELL" shell)
(unset-environment-variable! "SHELL")))))
(⊦= '(3) log))))
((eta 0.001) (memory #(13631492 4097432 1048576)) (stdout "") (stderr ""))
test/execute/missing-program: pass
(define (test/execute/missing-program _)
(⊦= '((#t 128))
(log-of
(τ (receive
(in out pid)
(cml/execute "/nonexistent/program")
(note! (sync (process-evt pid)))
(close-input-port in)
(close-output-port out))))))
((eta 0.001) (memory #(13631492 4099344 1048576)) (stdout "") (stderr ""))
test/execute/path-and-argv0: pass
(define (test/execute/path-and-argv0 _)
(⊦= '("/bin/sh" "sh" (#t 0) (#t 0))
(log-of
(τ (receive
(in1 out1 pid1)
(cml/execute "/bin/sh" '("-c" "echo $0"))
(receive
(in2 out2 pid2)
(cml/execute "sh" '("-c" "echo $0"))
(note! (sync (input-line-evt in1)))
(note! (sync (input-line-evt in2)))
(note! (sync (process-evt pid1)))
(note! (sync (process-evt pid2)))
(for-each close-input-port (list in1 in2))
(for-each close-output-port (list out1 out2))))))))
((eta 0.003) (memory #(13631492 4104616 1048576)) (stdout "") (stderr ""))
test/execute/in-env: pass
(define (test/execute/in-env _)
(⊦= '("LANG=C\nFOO=a=b\n" (#t 0) "" (#t 0) "set\n" (#t 0) (#t 128) raised)
(log-of
(τ (define (run cmd args . env)
(receive
(in out pid)
(apply cml/execute cmd args env)
(close-output-port out)
(note! (sync (input-all-evt in)))
(note! (sync (process-evt pid)))
(close-input-port in)))
(run "/usr/bin/env" '() '("LANG=C" "FOO=a=b"))
(run "/usr/bin/env" '() '())
(run "/bin/sh" '("-c" "echo ${PATH:+set}"))
(receive
(in out pid)
(cml/execute "env" '() '("A=b"))
(note! (sync (process-evt pid)))
(close-input-port in)
(close-output-port out))
(note! (condition-case
(cml/execute "/usr/bin/env" '() '("NOEQUALS"))
(e (exn) 'raised)))))))
((eta 0.004) (memory #(13631492 4113600 1048576)) (stdout "") (stderr ""))
test/children-start-with-sigpipe-default: pass
(define (test/children-start-with-sigpipe-default _)
(⊦= '(141 (#f 13) 0)
(log-of
(τ (note! (cml/system "kill -PIPE $$; exit 0"))
(receive
(in out pid)
(cml/execute "/bin/sh" '("-c" "kill -PIPE $$; exit 0"))
(note! (sync (process-evt pid)))
(close-input-port in)
(close-output-port out))
(note! (sync/timeout
(system-evt
"(while :; do echo x; done 2>/dev/null) | head -1 >/dev/null")
5
'hung))))))
((eta 0.002) (memory #(13631492 4111976 1048576)) (stdout "") (stderr ""))
I/O: TCP#
test/echo-server: pass
(define (test/echo-server _)
(let* ((l (tcp-listen 0 4 "127.0.0.1"))
(port (tcp-listener-port l))
(log (log-of
(τ (define (serve in out)
(let loop ()
(let1 (line (sync (input-line-evt in)))
(if (eof-object? line)
(begin
(close-input-port in)
(close-output-port out))
(begin
(sync (write-string-evt
out
(string-append "echo " line "\n")))
(loop))))))
(define (client name)
(let* ((io (sync (tcp-connect-evt "127.0.0.1" port)))
(in (car io))
(out (cadr io)))
(let1 (replies
(map (λ (i)
(sync (write-string-evt
out
(string-append name (number->string i) "\n")))
(sync (input-line-evt in)))
(iota 3)))
(close-output-port out)
(let1 (end (sync (input-line-evt in)))
(close-input-port in)
(list name replies end)))))
(spawn (τ (let loop ()
(let1 (io (sync (tcp-accept-evt l)))
(spawn (τ (serve (car io) (cadr io))))
(loop)))))
(let* ((r1 (make-result)) (r2 (make-result)))
(spawn (τ (result-put! r1 (client "a"))))
(spawn (τ (result-put! r2 (client "b"))))
(note! (result-get r1))
(note! (result-get r2)))))))
(tcp-close l)
(⊦= '(("a" ("echo a0" "echo a1" "echo a2") #!eof)
("b" ("echo b0" "echo b1" "echo b2") #!eof))
log)))
((eta 0.002) (memory #(13631492 4150296 1048576)) (stdout "") (stderr ""))
test/write-string-evt-sends-a-message-at-once: pass
(define (test/write-string-evt-sends-a-message-at-once _)
(let* ((l (tcp-listen 0 4 "127.0.0.1"))
(port (tcp-listener-port l))
(line (make-string 1000 #\m))
(r (run/value
(τ (spawn (τ (let* ((io (sync (tcp-accept-evt l)))
(in (car io))
(out (cadr io)))
(let loop ()
(let1 (s (sync (input-line-evt in)))
(unless
(eof-object? s)
(sync (write-string-evt out (string-append s "\n")))
(loop))))
(close-input-port in)
(close-output-port out))))
(let* ((io (sync (tcp-connect-evt "127.0.0.1" port)))
(in (car io))
(out (cadr io))
(t0 (current-process-milliseconds))
(ok (every-reply
(λ ()
(sync (write-string-evt out (string-append line "\n")))
(equal? line (sync (input-line-evt in)))))))
(let1 (ms (- (current-process-milliseconds) t0))
(close-output-port out)
(close-input-port in)
(list ok ms)))))))
(tcp-close l)
(⊨ (car r))
(⊨ (< (cadr r) 400))))
((eta 0.01) (memory #(13631492 4196880 1048576)) (stdout "") (stderr ""))
test/accept-evt/timeout-then-accept: pass
(define (test/accept-evt/timeout-then-accept _)
(let* ((l (tcp-listen 0 4 "127.0.0.1"))
(port (tcp-listener-port l))
(log (log-of
(τ (note! (sync/timeout (tcp-accept-evt l) 0.02 'nobody))
(spawn (τ (cml/sleep 0.02)
(let1 (io (sync (tcp-connect-evt "127.0.0.1" port)))
(sync (write-string-evt (cadr io) "hi\n"))
(close-output-port (cadr io))
(close-input-port (car io)))))
(let1 (io (sync (tcp-accept-evt l)))
(note! (sync (input-line-evt (car io))))
(note! (sync (input-line-evt (car io))))
(close-input-port (car io))
(close-output-port (cadr io)))))))
(tcp-close l)
(⊦= '(nobody "hi" #!eof) log)))
((eta 0.001) (memory #(13631492 4181464 1048576)) (stdout "") (stderr ""))
test/accept-evt/losing-branch: pass
(define (test/accept-evt/losing-branch _)
(let* ((l (tcp-listen 0 4 "127.0.0.1"))
(port (tcp-listener-port l))
(log (log-of
(τ (receive
(cin cout)
(tcp-connect "127.0.0.1" port)
(let1 (got (map (λ (i)
(select
(wrap (tcp-accept-evt l)
(λ (io)
(close-input-port (car io))
(close-output-port (cadr io))
'accepted))
(always-evt 'other)))
(iota 4)))
(note! (count (λ (x) (eq? x 'accepted)) got))
(note! (tcp-accept-ready? l)))
(close-input-port cin)
(close-output-port cout))))))
(tcp-close l)
(⊦= '(1 #f) log)))
((eta 0.0) (memory #(13631492 4183288 1048576)) (stdout "") (stderr ""))
test/accept-evt/two-acceptors-one-connection: pass
(define (test/accept-evt/two-acceptors-one-connection _)
(let* ((l (tcp-listen 0 4 "127.0.0.1")) (port (tcp-listener-port l)))
(receive
(cin cout)
(tcp-connect "127.0.0.1" port)
(let1 (r (run/log
(τ (for-each
(λ (tag)
(spawn (τ (let1 (io (sync (tcp-accept-evt l)))
(note! 'accepted)
(close-input-port (car io))
(close-output-port (cadr io))))))
'(a b))
(spawn (τ (cml/sleep 0.2) (cml/shutdown 'watchdog)))
(cml/exit))
#:quantum
1))
(close-input-port cin)
(close-output-port cout)
(tcp-close l)
(⊦= '(watchdog accepted) r)))))
((eta 0.001) (memory #(13631492 4186456 1048576)) (stdout "") (stderr ""))
test/utf-8-split-across-segments: pass
(define (test/utf-8-split-across-segments _)
(let* ((l (tcp-listen 0 4 "127.0.0.1"))
(port (tcp-listener-port l))
(log (parameterize
((tcp-read-timeout 2000))
(log-of
(τ (let* ((c (sync (tcp-connect-evt "127.0.0.1" port)))
(s (sync (tcp-accept-evt l)))
(fd (port->fileno (cadr c)))
(ticks 0))
(spawn (τ (let loop ()
(cml/sleep 0.01)
(set! ticks (add1 ticks))
(loop))))
(spawn (τ (file-write fd (bytevector 99 97 102 195))
(cml/sleep 0.1)
(file-write fd (bytevector 169 33 10))))
(note! (condition-case
(sync (input-line-evt (car s)))
(e () 'blocked)))
(note! (> ticks 3))
(for-each close-input-port (list (car c) (car s)))
(for-each
close-output-port
(list (cadr c) (cadr s)))))))))
(tcp-close l)
(⊦= '("café!" #t) log)))
((eta 0.001) (memory #(13631492 4098512 1048576)) (stdout "") (stderr ""))
I/O: channel ports#
test/round-trip: pass
(define (test/round-trip _)
(⊦= '(("hello" "world" "!" #!eof #!eof) ("hello\nworld" "\n!" #!eof))
(begin
(list (log-of
(τ (let* ((ch (make-channel))
(out (open-channel-output-port ch))
(in (open-channel-input-port ch)))
(spawn (τ (display "hello\nworld" out)
(flush-output out)
(display "\n!" out)
(close-output-port out)))
(note! (read-line in))
(note! (read-line in))
(note! (read-line in))
(note! (read-line in))
(note! (read-char in)))))
(log-of
(τ (let* ((ch (make-channel))
(out (open-channel-output-port ch)))
(spawn (τ (display "hello\nworld" out)
(flush-output out)
(display "\n!" out)
(close-output-port out)))
(note! (recv ch))
(note! (recv ch))
(note! (recv ch)))))))))
((eta 0.0) (memory #(13631492 4104584 1048576)) (stdout "") (stderr ""))
test/blocked-channel-port-reader-is-a-deadlock: pass
(define (test/blocked-channel-port-reader-is-a-deadlock _)
(define (status-in-a-child thunk)
(let1 (pid (process-fork
(τ (emergency-exit (if (eq? 'failure (run-cml thunk)) 0 1)))))
(let loop ((i 0))
(receive
(p normal? code)
(process-wait pid #t)
(cond ((not (zero? p))
(if (and normal? (zero? code))
'failure
(list 'exited code)))
((< i 1000) (sleep-ms 10) (loop (add1 i)))
(else (process-signal pid) (process-wait pid) 'hung))))))
(⊦= 'failure
(status-in-a-child
(τ (read-line (open-channel-input-port (make-channel))))))
(⊦= 'failure
(status-in-a-child
(τ (sync (input-line-evt (open-channel-input-port (make-channel)))))))
(⊦= 'failure
(status-in-a-child
(τ (let1 (in (open-channel-input-port (make-channel)))
(spawn (τ (sync (input-char-evt in))))
(sync (input-all-evt in)))))))
((eta 0.002) (memory #(13631492 4105448 1048576)) (stdout "") (stderr ""))
test/events-on-channel-ports: pass
(define (test/events-on-channel-ports _)
(⊦= '(timeout "abcd" #\e "fg" #!eof)
(log-of
(τ (let* ((ch (make-channel)) (in (open-channel-input-port ch)))
(spawn (τ (send ch "ab")
(cml/sleep 0.03)
(send ch "")
(send ch #\c)
(send ch "d\ne")
(send ch "fg")
(send ch #!eof)))
(note! (sync/timeout (input-line-evt in) 0.01 'timeout))
(note! (sync (input-line-evt in)))
(note! (sync (input-char-evt in)))
(note! (sync (input-all-evt in)))
(note! (sync (input-line-evt in))))))))
((eta 0.001) (memory #(13631492 4111664 1048576)) (stdout "") (stderr ""))
test/write-string-evt-is-a-rendezvous: pass
(define (test/write-string-evt-is-a-rendezvous _)
(⊦= '(timeout "buffered+now" sent)
(log-of
(τ (let* ((ch (make-channel)) (out (open-channel-output-port ch)))
(display "buffered+" out)
(note! (sync/timeout (write-string-evt out "now") 0.01 'timeout))
(spawn (τ (note! (recv ch))))
(sync (write-string-evt out "now"))
(note! 'sent))))))
((eta 0.0) (memory #(13631492 4113936 1048576)) (stdout "") (stderr ""))
test/write-string-evt-and-a-flush: pass
(define (test/write-string-evt-and-a-flush _)
(⊦= '("abcqX" ok flushed nothing)
(log-of
(τ (let* ((ch (make-channel)) (out (open-channel-output-port ch)))
(display "abc" out)
(let ((f (spawn (τ (cml/yield)
(display "q" out)
(flush-output out))))
(a (spawn (τ (note! (condition-case
(begin (sync (write-string-evt out "X")) 'ok)
(e () 'raised)))))))
(note! (recv ch))
(note! (sync/timeout
(wrap (join-evt f) (λ ignored 'flushed))
0.5
'flush-stuck))
(sync (join-evt a))
(note! (sync/timeout (recv-evt ch) 0.05 'nothing))))))))
((eta 0.0) (memory #(13631492 4110648 1048576)) (stdout "") (stderr ""))
test/channel-output-port-never-sends-empty-strings: pass
(define (test/channel-output-port-never-sends-empty-strings _)
(⊦= '(done done "ab" "c" #!eof nothing)
(log-of
(τ (let* ((ch (make-channel)) (out (open-channel-output-port ch)))
(note! (sync/timeout
(wrap (write-string-evt out "") (λ ignored 'done))
0.2
'blocked))
(spawn (τ (sync (write-string-evt out ""))
(display "" out)
(flush-output out)
(sync (write-string-evt out "ab"))
(sync (write-string-evt out ""))
(display "c" out)
(sync (write-string-evt out ""))
(flush-output out)
(close-output-port out)))
(note! (sync/timeout
(wrap (write-string-evt out "") (λ ignored 'done))
0.2
'blocked))
(let loop ()
(let1 (v (sync/timeout (recv-evt ch) 0.2 'nothing))
(note! v)
(unless (symbol? v) (loop)))))))))
((eta 0.001) (memory #(13631492 4117888 1048576)) (stdout "") (stderr ""))
test/flush-keeps-a-threads-output-in-order: pass
(define (test/flush-keeps-a-threads-output-in-order _)
(⊦= '(("Xs" nothing) "Xsyz")
(log-of
(τ (let* ((ch (make-channel)) (out (open-channel-output-port ch)))
(spawn (τ (display "X" out) (sync (write-string-evt out "s"))))
(cml/yield)
(spawn (τ (flush-output out)))
(cml/yield)
(note! (let* ((a (recv ch))
(b (sync/timeout (recv-evt ch) 0.05 'nothing)))
(list a b)))
(let* ((ch (make-channel))
(out (open-channel-output-port ch))
(in (open-channel-input-port ch)))
(spawn (τ (display "X" out)
(sync (write-string-evt out "s"))
(display "yz" out)
(close-output-port out)))
(cml/yield)
(spawn (τ (flush-output out)))
(cml/yield)
(note! (sync (input-all-evt in)))))))))
((eta 0.0) (memory #(13631492 4118448 1048576)) (stdout "") (stderr ""))
test/output-events-on-a-closed-port: pass
(define (test/output-events-on-a-closed-port _)
(⊦= '("a" #!eof raised raised raised nothing-sent)
(log-of
(τ (let* ((ch (make-channel)) (o (open-channel-output-port ch)))
(spawn (τ (display "a" o) (close-output-port o)))
(note! (recv ch))
(note! (recv ch))
(note! (condition-case
(begin (display "b" o) 'ok)
(e () 'raised)))
(note! (condition-case
(begin (sync (write-string-evt o "late")) 'ok)
(e () 'raised)))
(note! (condition-case
(begin (sync (output-evt o)) 'ok)
(e () 'raised)))
(note! (sync/timeout (recv-evt ch) 0.02 'nothing-sent)))))))
((eta 0.001) (memory #(13631492 4127800 1048576)) (stdout "") (stderr ""))
test/peek-char-then-events: pass
(define (test/peek-char-then-events _)
(⊦= '(#\a #t #\a #\b)
(log-of
(τ (let* ((ch (make-channel)) (in (open-channel-input-port ch)))
(spawn (τ (send ch "a") (cml/sleep 0.1) (send ch "b")))
(note! (peek-char in))
(note! (char-ready? in))
(note! (sync/timeout (input-char-evt in) 0.02 'stuck))
(note! (read-char in)))))))
((eta 0.0) (memory #(13631492 4119096 1048576)) (stdout "") (stderr ""))
test/bad-value-reaches-the-syncing-thread: pass
(define (test/bad-value-reaches-the-syncing-thread _)
(⊦= '(raised #\g)
(log-of
(τ (let* ((ch (make-channel)) (in (open-channel-input-port ch)))
(spawn (τ (cml/sleep 0.01)
(send ch 'not-a-string)
(send ch "good")))
(note! (condition-case
(sync/timeout (input-char-evt in) 1 'stuck)
(e (exn) 'raised)))
(note! (sync/timeout (input-char-evt in) 1 'stuck)))))))
((eta 0.0) (memory #(13631492 4115408 1048576)) (stdout "") (stderr ""))
test/bad-value-keeps-the-chars-before-it: pass
(define (test/bad-value-keeps-the-chars-before-it _)
(define (run mk n)
(log-of
(τ (let* ((ch (make-channel)) (in (open-channel-input-port ch)))
(spawn (τ (send ch "hello ")))
(spawn (τ (send ch 42)))
(spawn (τ (send ch "world\n")))
(cml/sleep 0.05)
(for-each
(λ (i)
(note! (condition-case
(sync/timeout (mk in) 1 'stuck)
(e (exn) 'raised))))
(iota n))))))
(⊦= '(raised "hello world") (run input-line-evt 2))
(⊦= '(#\h #\e #\l #\l #\o #\space raised #\w) (run input-char-evt 8)))
((eta 0.001) (memory #(13631492 4118064 1048576)) (stdout "") (stderr ""))
test/dropped-ports-are-collected: pass
(define (test/dropped-ports-are-collected _)
(define (collected? make)
(let1 (ws (map (λ (i) (weak-cons (make) #f)) (iota 100)))
(gc #t)
(>= (count (λ (w) (bwp-object? (car w))) ws) 90)))
(define (custom-port) (make-input-port (τ #\a) (τ #t) void))
(⊨ (collected? (τ (open-channel-input-port (make-channel)))))
(⊨ (collected? (τ (open-channel-output-port (make-channel)))))
(⊨ (collected? (τ (let1 (p (custom-port)) (%port-driver p) p))))
(⊨ (collected?
(τ (let1 (p (open-channel-input-port (make-channel)))
(run-cml (τ (sync/timeout (input-char-evt p) 0 #f)))
p))))
(let1 (most (let loop ((i 0) (most 0))
(if (= i 2000)
most
(begin
(%port-driver
(let1 (p (custom-port))
(##sys#set-port-data! p (list 'its-own))
p))
(when (zero? (remainder i 50)) (gc #t))
(loop (add1 i)
(max most
(length (%port-table-others %pdrivers))))))))
(⊨ (< most 200)))
(receive
(in out)
(create-pipe)
(let1 (p (open-input-file* in))
(%port-driver p)
(close-input-port p)
(file-close out)))
(receive
(in out)
(create-pipe)
(let1 (p (open-input-file* in))
(%port-driver p)
(⊦= 1 (length (vector-ref (%port-table-fds %pdrivers) in)))
(close-input-port p)
(file-close out))))
((eta 0.168) (memory #(13631492 4120192 1048576)) (stdout "") (stderr ""))
test/ports-without-a-descriptor-carry-their-state: pass
(define (test/ports-without-a-descriptor-carry-their-state _)
(define (custom-port)
(make-input-port
(let1 (s (open-input-string "b\n")) (τ (read-char s)))
(τ #t)
void))
(let ((s (open-input-string "a\n")) (c (custom-port)) (o (custom-port)))
(##sys#set-port-data! o (list 'its-own))
(⊦= '("a" "b" "b")
(run/value (τ (map (λ (p) (sync (input-line-evt p))) (list s c o)))))
(⊨ (%pdriver? (%own-port-state s)))
(⊨ (%pdriver? (%own-port-state c)))
(⊦= '(its-own) (##sys#slot o 9))
(⊭ (%own-port-state o))
(⊭ (memq s (map car (%port-table-others %pdrivers))))
(⊭ (memq c (map car (%port-table-others %pdrivers))))
(⊨ (and (memq o (map car (%port-table-others %pdrivers))) #t))))
((eta 0.0) (memory #(13631492 4125008 1048576)) (stdout "") (stderr ""))
test/many-open-ports-in-linear-time: pass
(define (test/many-open-ports-in-linear-time _)
(define (cost n)
(let1 (t0 (current-process-milliseconds))
(⊦= n
(run/value
(τ (let1 (ports (map (λ (i)
(let1 (ch (make-channel))
(cons (open-channel-input-port ch)
(open-channel-output-port ch))))
(iota n)))
(for-each
(λ (p)
(sync (output-evt (cdr p)))
(sync/timeout (input-char-evt (car p)) 0 #f))
ports)
(length ports)))))
(- (current-process-milliseconds) t0)))
(let* ((small (max 20 (cost 400))) (large (cost 4000)))
(⊨ (< large (* 30 small)))))
((eta 0.062) (memory #(27262984 10709472 1048576)) (stdout "") (stderr ""))
test/pipe-to-channel-bridge: pass
(define (test/pipe-to-channel-bridge _)
(⊦= '("one" "two" #!eof)
(with-pipe
(λ (p out)
(log-of
(τ (let* ((ch (make-channel))
(cin (open-channel-input-port ch))
(cout (open-channel-output-port ch)))
(spawn (τ (let loop ()
(let1 (l (sync (input-line-evt p)))
(if (eof-object? l)
(close-output-port cout)
(begin
(display l cout)
(newline cout)
(flush-output cout)
(loop)))))))
(spawn (τ (fd-write! out "one\ntw")
(cml/sleep 0.01)
(fd-write! out "o\n")
(file-close out)))
(note! (sync (input-line-evt cin)))
(note! (sync (input-line-evt cin)))
(note! (sync (input-line-evt cin))))))))))
((eta 0.0) (memory #(27262984 4111320 1048576)) (stdout "") (stderr ""))