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 ""))
Categories: scheme
See also
The (aux anthropic) module
The Anthropic Messages API, offline
The (aux hakaru10) modules
Random sources, distributions, the language, exact inference and examples
The (aux kanren) modules
Arithmetic, dates, presentation of answers and a union-find miniKanren
The (aux games dice-of-doom) module
Dice of Doom
The (aux machine) modules
Environments, the SC machine and the SECD machine
The (aux schemer seasoned) module
The Seasoned Schemer
The (aux continuation) modules
Classic operators, escapers, amb and Collatz sequences
The (aux category) modules
Monads: list, maybe, probability, ziplist, difference lists and state
The (aux timsort) module
Sorting and the trace buffer
The (aux fds) modules
Functional queues and union-find
The (aux sxml) module
Rendering SXML to HTML
The (aux variable) module
Logic variables introduced by fresh
The (aux tabling) module
Memoization and tabled definitions
The (aux commons) module
Learning tests for CHICKEN Scheme
Promises
delay and make-promise
The Ackermann function
On unary numbers, as lists
Concurrent ML for CHICKEN Scheme
Porting SML/NJ CML as the (aux cml) library
CHICKEN Scheme
Versions and implementation notes
The SBRAL data structure
Skew Binary Random-Access Lists
Pattern matching
Scheme macrology: a generalized pattern matcher
The (aux hansei) modules
Inference, random sources, lazy lists, logic programming and example models
The (aux nondeterministic) module
Learning tests for Scheme
The (aux stream) modules
Streams, streams after SICP, and sequences and power series
The (aux unittest) module
Bootstrapping a unittest framework
Scheme lang: Chez, 48 and Chicken
Some docs, refs and containers, respectively.
The Reasoned Schemer
miniKanren, microKanren and some implementations
Aggregation predicates for µkanren
Groups, window functions, sets and enumerations
miniKanren, live and untagged
Quine generation via relational interpreters
Pure, declarative, and constructive arithmetic relations
A declarative pearl
Final shift for call/cc
Direct implementation of shift and reset
Call with Current Continuation Patterns
Cogen in six lines
The Seasoned Schemer
The Little Schemer
Adding generic functions to Scheme
Handling control