I was looking at elm’s Task module and was wondering if I could hide some of the implementation details in my code which heavily uses Eio for parallelism. It turns out this can be done:
(Not sure about the word choice of “IO” here, but it’s what elm used internally, so I ran with it while writing this).
Here is a usage example:
let count_primes n =
let is_prime k =
k > 1
&&
let rec go d = d * d > k || (k mod d <> 0 && go (d + 1)) in
go 2
in
let c = ref 0 in
for k = 2 to n do
if is_prime k then incr c
done;
!c
let test_pool_parallel ~env ~sw () =
let pool =
Eio.Executor_pool.create ~sw ~domain_count:2 (Eio.Stdenv.domain_mgr env)
in
let module T = Task.Make_pool_task (struct
let sw = sw
let pool = pool
end) in
let open T in
let timed_count_primes n =
let t0 = Unix.gettimeofday () in
let r = count_primes n in
(r, Unix.gettimeofday () -. t0)
in
let task_a =
let@ () = run in
timed_count_primes 2_000_000
in
let task_b =
let@ () = run in
timed_count_primes 2_000_000
in
let pipeline =
let+ a, time_a = task_a and+ b, time_b = task_b in
(a + b, time_a +. time_b)
in
let start = Unix.gettimeofday () in
let result = Eio.Promise.await (perform Fun.id pipeline) in
let elapsed = Unix.gettimeofday () -. start in
match result with
| Error exn -> Format.eprintf "pool job raised: %s" (Printexc.to_string exn)
| Ok (_total, sequential_estimate) ->
assert (elapsed < sequential_estimate *. 0.75);
Format.printf "It's fast!@."
let () =
let@ env = Eio_main.run in
let@ sw = Eio.Switch.run ?name:None in
test_pool_parallel ~env ~sw ()
This seems to work, but I wonder where it lands on the [gimmick ↔ useful] spectrum
Is there anything about this that screams “bad idea”? Any pitfalls?
I gather you are working in the constraints of eio. But to me it just screams bureaucracy and wrong abstractions. Just for fun I tried to write your example with the newly released affect library, here’s the result:
open Affect
let count_primes n = …
let timed_count_primes n = …
let main () =
Fun.Async.main ~domain_count:2 @@ fun () ->
let start = Unix.gettimeofday () in
let task_a = Fun.Async.call (fun () -> timed_count_primes 2_000_000) in
let task_b = Fun.Async.call (fun () -> timed_count_primes 2_000_000) in
let a, time_a = Fun.Async.get task_a in
let b, time_b = Fun.Async.get task_b in
let total, sequential_estimate = a + b, time_a +. time_b in
let elapsed = Unix.gettimeofday () -. start in
if elapsed < sequential_estimate *. 0.75
then Format.printf "It's fast!@."
let () = if !Sys.interactive then () else main ()
This is very nice, I’d like to adopt the library. We were using eio for it’s IO capabilities, so it was natural to reach for it’s concurrency capabilities as well, but keeping track of everything manually became a pain. Are there any considerations for running Fun.Async.main within Eio_main.run or vice versa?
That’s uncharted territory :–) But I suspect that simple nested runs would lead to obscure behaviours or disasters one way or the other. For a tighter and more thoughtful integration, the effects are exposed for handling though, see this cookbook entry.
I must also show an example with miou (but as far as the scheduler proposes by default a pool of domains, it’s much more easy to describe in OCaml what you want to do):
let main () =
Miou.run ~domains:2 @@ fun () ->
let start = Unix.gettimeofday () in
let domains = Miou.Domain.all () in
let task_a = Miou.call ~pin:List.(hd domains) (fun () -> timed_count_primes 2_000_000) in
let task_b = Miou.call ~pin:List.(hd (tl domains)) (fun () -> timed_count_primes 2_000_000) in
let a, time_a = Miou.await_exn task_a in
let b, time_b = Miou.await_exn task_b in
let total, sequential_estimate = a + b, time_a +. time_b in
let elapsed = Unix.gettimeofday () -. start in
if elapsed < sequential_estimate *. 0.75
then Format.printf "It's fast!@."
let () = if !Sys.interactive then () else main ()
parallel also exists to propose something like the fork-join pattern (used by httpcats for instance to launch http servers on multiple domains).
I reinvented the wheel, but made it square. For now I think the best solution will be to find a more ergonomic way to invoke parallelism on top of eio, and adopting an eio-compatible scheduler if that fails.
let () =
let open Lwt.Syntax in
let pool = Lwt_domain.setup_pool 2 in
Lwt_main.run (
let start = Unix.gettimeofday () in
let+ a, time_a = Lwt_domain.detach pool timed_count_primes 2_000_000
and+ b, time_b = Lwt_domain.detach pool timed_count_primes 2_000_000
in
let _total, sequential_estimate = a + b, time_a +. time_b in
let elapsed = Unix.gettimeofday () -. start in
if elapsed < sequential_estimate *. 0.75
then Format.printf "It's fast!@."
)
The boilerplate is:
setup_pool for setup
and+ for concurrent execution of the two tasks
detach for parallel execution of the computation
[EDIT] lwt_domain is the work of @sudha , I’m doing a little bit of packaging and release work for that project, but she should be given the credits for this library.
The other story is the concurrency model. I note that both miou and lwt require you to have specific scheduling instructions and/or setup scheduling structures which is not compositional (more on this on the affect design notes and concurrency model). And don’t get me started on their cancellation model :–)
To actually answer your question about Eio, I think you can just use submit_fork for this to look similar to the other examples.
let count_primes n =
let is_prime k =
k > 1 &&
let rec go d = d * d > k || (k mod d <> 0 && go (d + 1)) in
go 2
in
let c = ref 0 in
for k = 2 to n do
if is_prime k then incr c
done;
!c
let timed_count_primes n =
let t0 = Unix.gettimeofday () in
let r = count_primes n in
(r, Unix.gettimeofday () -. t0)
let main () =
Eio_main.run @@ fun env ->
Eio.Switch.run @@ fun sw ->
let pool =
Eio.Executor_pool.create ~sw ~domain_count:2 (Eio.Stdenv.domain_mgr env)
in
let start = Unix.gettimeofday () in
let task_a =
Eio.Executor_pool.submit_fork ~sw pool ~weight:1.0 (fun () ->
timed_count_primes 2_000_000)
in
let task_b =
Eio.Executor_pool.submit_fork ~sw pool ~weight:1.0 (fun () ->
timed_count_primes 2_000_000)
in
let a, time_a = Eio.Promise.await_exn task_a in
let b, time_b = Eio.Promise.await_exn task_b in
let _total, sequential_estimate = (a + b, time_a +. time_b) in
let elapsed = Unix.gettimeofday () -. start in
if elapsed < sequential_estimate *. 0.75 then Format.printf "It's fast!@."
I doubt that reintroducing a monadic task abstraction would be ergonomic over a direct-style library. I’ve not had the time to look at affect yet, but I agree with Daniel that cancellation semantics are where things get complex!