open Lwt.Syntax

let rec worker recv_task f send_result =

let* task = Lwt_stream.get recv_task in

match task with

| None ->

send_result None;

Lwt.return ()

| Some data ->

let* result = f data in

send_result (Some result);

let* () = Lwt.pause () in

worker recv_task f send_result

let spawn_domain_worker f =

let recv_task, send_task = Lwt_stream.create () in

let recv_result, send_result = Lwt_stream.create () in

let dw =

Domain.spawn (fun () ->

Lwt_unix.init_domain ();

Lwt_main.run (

let* () = Lwt.pause () in

worker recv_task f send_result

)

)

in

send_task, dw, recv_result

let simulate_work data =

let* () = Lwt.pause () in

Lwt.return (String.length data)

let input = [""; "adsf"; "lkjh"; "lkjahsdflkjahdlfkjha"; "0"; ""; ""; ""; ""; ""; "adf"; "ASDSKJLHDAS"; "WPOQIEU"; "DSFALKHJ"; ""; ""; ""; ""; "SD"; "SD"; "SAD; SD;SD"; "ad"; "...."]

let expected_result = List.fold_left (fun acc s -> acc + String.length s) 0 input

let main () =

let send_task1, dw1, recv_result1 = spawn_domain_worker simulate_work in

let send_task2, dw2, recv_result2 = spawn_domain_worker simulate_work in

let l =

Lwt_unix.init_domain ();

Lwt_main.run (

let* () = Lwt.pause () in

let () = (* push work *)

List.iteri

(fun idx s -> if idx mod 3 = 0 then send_task1 (Some s) else send_task2 (Some s))

input

in

send_task1 None;

send_task2 None;

let* lengths1 = Lwt_stream.fold (+) recv_result1 0

and* lengths2 = Lwt_stream.fold (+) recv_result2 0

in

Lwt.return (lengths1 + lengths2)

)

in

let () = Domain.join dw1 in

let () = Domain.join dw2 in

let _code =

if l = expected_result then begin

0

end else begin

Printf.printf "domain-workers: ×\n";

flush_all ();

exit 1

end

in

()

let () =

for _ = 1 to 1000 do

main ()

done;

Printf.printf "domain-workers: ✓\n";

flush_all ();

exit 0

(names basic domainworkers domainworkers2 movingpromises unixpipe preempting)