Something went wrong. Try again.
The Monad language. Dependent types, functional programming compiled with LLVM. Hobby project. monad-lang.org
dependent-types language compiler programming-language functional-programming
Something went wrong. Try again.
8.2 kB · 264 lines
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265// Structured concurrency tests: all, race, scoped, sleepIO, cancel_all//// One test here, `test_sleepIO_fibers_run_concurrently`, asserts on// ELAPSED TIME, so it fails on any implementation that serializes// fibers. The Rust host is such an implementation and is lazy on// purpose; this file runs under the compiled backend's runner, not the// host's. See that test's own comment for the numbers and the reason.use lib::concurrent::fiber {Fiber, FiberError, await_fiber, cancel_fiber, forkIO}use lib::concurrent::combine { Duration, FiberScope, Timeout, all, race, scope_fork, scoped, sleepIO, timeout,}use lib::io {}use lib::list {List.length}open IO {current_time_nano}// Helper thunksdef io_10_action (_ : Unit) : IO I64 := IO.pure 10def io_20_action (_ : Unit) : IO I64 := IO.pure 20def io_30_action (_ : Unit) : IO I64 := IO.pure 30def io_42_action (_ : Unit) : IO I64 := IO.pure 42def ok_is (r : Result FiberError I64) (v : I64) : Bool := match r { Result.ok x => x == v, Result.err _ => false }// `all` collects in reverse await order; only the count is checked.def ok_count (r : Result FiberError (List I64)) (n : I64) : Bool := match r { Result.ok xs => List.length xs == n, Result.err _ => false }// race: concrete specializeddef cancel_i64 (fibers : List (Fiber I64)) : IO Unit := match fibers { List.empty => IO.pure Unit.unit, List.cons f rest => do { cancel_fiber f; cancel_i64 rest } }// Tests#[test]def test_all_single : IO Bool { let f <- forkIO io_42_action; let fibers := List.cons f List.empty; let results <- all fibers; return (ok_count results 1)}#[test]def test_all_two : IO Bool { let f1 <- forkIO io_10_action; let f2 <- forkIO io_20_action; let fibers := List.cons f1 (List.cons f2 List.empty); let results <- all fibers; return (ok_count results 2)}#[test]def test_all_three : IO Bool { let f1 <- forkIO io_10_action; let f2 <- forkIO io_20_action; let f3 <- forkIO io_30_action; let fibers := List.cons f1 (List.cons f2 (List.cons f3 List.empty)); let results <- all fibers; return (ok_count results 3)}#[test]def test_all_empty : IO Bool { let fibers := (List.empty : List (Fiber I64)); let results <- all fibers; return (ok_count results 0)}def err_is_cancelled (r : Result FiberError A) : Bool := match r { Result.ok _ => false, Result.err e => match e { FiberError.cancelled => true, FiberError.never_started => false } }/// `all` stops at the first fiber with no result and reports why, rather/// than handing back a list with a made-up element. Both backends: the/// cancelled fiber is still sleeping (compiled) or never ran (host).#[test]def test_all_reports_a_cancelled_fiber : IO Bool { let f1 <- forkIO io_10_action; let f2 <- forkIO slow_10_action; cancel_fiber f2; let results <- all [f1, f2]; return (err_is_cancelled results)}/// A race whose only contestant was cancelled answers `err cancelled`/// once that fiber stops, instead of a value it never produced.#[test]def test_race_of_a_cancelled_fiber_is_err : IO Bool { let f <- forkIO slow_10_action; cancel_fiber f; let result <- race [f]; return (err_is_cancelled result)}def slow_10_action (_ : Unit) : IO I64 := do { sleepIO 400; return 10}/// The first fiber to FINISH wins, not the first in the list: the head/// sleeps 400 ms, so the instant one must win, and well before the head/// would have finished. Compiled only, like the timing test below: the/// host runs fibers lazily and in order, so its head always wins.#[test]def test_race_first_to_finish : IO Bool { let start <- current_time_nano; let slow <- forkIO slow_10_action; let fast <- forkIO io_20_action; let result <- race [slow, fast]; let stop <- current_time_nano; let elapsed_ms := I64.div (I64.sub stop start) 1000000; return (ok_is result 20 && elapsed_ms < 300)}#[test]def test_race_single : IO Bool { let f1 <- forkIO io_30_action; let fibers := List.cons f1 List.empty; let result <- race fibers; return (ok_is result 30)}#[test]def test_cancel_all_smoke : IO Bool { let f1 <- forkIO io_10_action; let f2 <- forkIO io_20_action; let fibers := List.cons f1 (List.cons f2 List.empty); cancel_i64 fibers; return true}// scoped: creates scope, runs action, auto-cancels on exitdef fork_io_42_scoped (s : FiberScope) : IO (Fiber I64) := scope_fork s io_42_actiondef scoped_fork_await_42 (s : FiberScope) : IO (Result FiberError I64) := do { let f <- fork_io_42_scoped s; await_fiber f}#[test]def test_scoped_runs_action : IO Bool { let result <- scoped scoped_fork_await_42; return (ok_is result 42)}// scoped: auto-cancels on exit (callback returns without awaiting)def scoped_forget (s : FiberScope) : IO I64 := do { scope_fork s io_42_action; return 0}#[test]def test_scoped_cancels : IO Bool { let result <- scoped scoped_forget; return (result == 0)}// sleepIO: basic smoke test#[test]def test_sleepIO_smoke : IO Bool { sleepIO 10; return true}// The one test here that only a genuinely CONCURRENT implementation can// pass, and the reason `forkIO` must actually start its thread at the// fork rather than defer the action to the first `await`.//// Every other test in this file -- and every test in `fiber_test.mo` --// passes just as well on a sequential implementation, because each one// awaits the fiber it just forked: a `forkIO` that returns a deferred// thunk, and an `await_fiber` that runs it inline, satisfies them all.// That is exactly what the Rust host does on purpose (see// `core/src/core_native.rs`'s `fork_io`, "cooperative/lazy, not real// OS-thread concurrency"), so those tests cannot tell the backends// apart. This one can: four fibers are forked BEFORE any of them is// awaited, so the four sleeps overlap only if the four threads exist at// the same time.//// concurrent -> the sleeps overlap -> ~500 ms// sequential -> the sleeps are a sum -> ~2000 ms//// The bound is 1200 ms: 2.4x the concurrent expectation, so a loaded or// slow machine cannot flake it, and well under the sequential sum, so a// serialized implementation cannot squeak past it. The 500 ms sleep is// long enough that thread startup -- a `pthread_create` plus Boehm's// thread registration, tens of microseconds, with the 128 MB stack// committed lazily -- cannot move the result either way. `_r1`..`_r4`// are bound rather than discarded only so the awaits read as an// ordered sequence of four; their values are all `Unit`.def sleep_500_action (_ : Unit) : IO Unit := sleepIO 500#[test]def test_sleepIO_fibers_run_concurrently : IO Bool { let start <- current_time_nano; let f1 <- forkIO sleep_500_action; let f2 <- forkIO sleep_500_action; let f3 <- forkIO sleep_500_action; let f4 <- forkIO sleep_500_action; let _r1 <- await_fiber f1; let _r2 <- await_fiber f2; let _r3 <- await_fiber f3; let _r4 <- await_fiber f4; let finish <- current_time_nano; return (I64.lt (I64.sub finish start) 1200000000)}/// The waiter is first in the race list, so the lazy host runs it and/// reports `finished`; the compiled backend races it against the timer and/// the work finishes first either way.#[test]def test_timeout_finished_before_expiry : IO Bool := do { let f <- forkIO (fn (_ : Unit) => do { sleepIO 10; return 7i64 }); let d : Duration := { millis := 300 }; let r <- timeout d f; return (match r { Result.ok t => match t { Timeout.finished inner => ok_is inner 7, Timeout.expired => false }, Result.err _ => false })}/// Compiled-only: the host's poll runs the waiter to completion, so it can/// never observe the timer winning.#[test]def test_timeout_expires : IO Bool := do { let f <- forkIO (fn (_ : Unit) => do { sleepIO 300; return 7i64 }); let d : Duration := { millis := 20 }; let r <- timeout d f; return (match r { Result.ok t => match t { Timeout.finished _ => false, Timeout.expired => true }, Result.err _ => false })}