Add Event.from_computation and a Promise.completed event

Author: polytypic

lib/picos_structured/dune

@@ -1,7 +1,11 @@
 (library
  (name picos_structured)
  (public_name picos.structured)
- (libraries picos backoff multicore-magic))
+ (libraries
+  (re_export picos.sync)
+  picos
+  backoff
+  multicore-magic))
 
 (mdx
  (enabled_if

lib/picos_structured/picos_structured.mli

@@ -13,6 +13,7 @@
     ]} *)
 
 open Picos
+open Picos_sync
 
 (** {1 Modules} *)
 
@@ -171,6 +172,10 @@ module Promise : sig
       {{!Control.Terminate} [Terminate]} exception in case the promise has been
       canceled. *)
 
+  val completed : 'a t -> 'a Event.t
+  (** [completed promise] returns an {{!Picos_sync.Event} event} that can be
+      committed to once the promise has completed. *)
+
   val is_running : 'a t -> bool
   (** [is_running promise] determines whether the completion of the promise is
       still pending. *)

lib/picos_structured/promise.ml

@@ -1,9 +1,11 @@
 open Picos
+open Picos_sync
 
 type 'a t = 'a Computation.t
 
 let of_value = Computation.returned
 let await = Computation.await
+let completed = Event.from_computation
 let is_running = Computation.is_running
 
 let try_terminate ?callstack t =

lib/picos_sync/event.ml

@@ -14,52 +14,73 @@ type ('a, 'r) id = Yes : ('a, 'a) id | No : ('a, 'r) id
 
 let rec request_1_as :
     type a r. (_ -> r) Computation.t -> (a -> r) -> (a, r) id -> a t -> _ =
- fun computation to_result id -> function
-  | Request { request } -> request computation to_result
-  | Choose ts -> request_n_as computation to_result id ts
+ fun target to_result id -> function
+  | Request { request } -> request target to_result
+  | Choose ts -> request_n_as target to_result id ts
   | Wrap { event; fn } ->
       let to_result =
         match id with No -> fun x -> to_result (fn x) | Yes -> fn
       in
-      request_1_as computation to_result No event
+      request_1_as target to_result No event
 
 and request_n_as :
     type a r. (_ -> r) Computation.t -> (a -> r) -> (a, r) id -> a t list -> _ =
- fun computation to_result id -> function
+ fun target to_result id -> function
   | [] -> ()
   | t :: ts ->
-      request_1_as computation to_result id t;
-      request_n_as computation to_result id ts
+      request_1_as target to_result id t;
+      request_n_as target to_result id ts
 
 type ('a, _) tycon = Id : ('a, 'a t) tycon | List : ('a, 'a t list) tycon
 
 let sync_as : type a n. n -> (a, n) tycon -> a =
  fun t n ->
-  let computation = Computation.create ~mode:`LIFO () in
+  let target = Computation.create ~mode:`LIFO () in
   match
     match n with
-    | Id -> request_1_as computation Fun.id Yes t
-    | List -> request_n_as computation Fun.id Yes t
+    | Id -> request_1_as target Fun.id Yes t
+    | List -> request_n_as target Fun.id Yes t
   with
   | () ->
-      if Computation.is_running computation then begin
+      if Computation.is_running target then begin
         let t = Trigger.create () in
-        if Computation.try_attach computation t then
+        if Computation.try_attach target t then
           match Trigger.await t with
           | None -> ()
           | Some exn_bt ->
-              if Computation.try_cancel computation exn_bt then
-                Exn_bt.raise exn_bt
+              if Computation.try_cancel target exn_bt then Exn_bt.raise exn_bt
       end;
-      Computation.await computation ()
+      Computation.await target ()
   | exception exn ->
       let exn_bt = Exn_bt.get exn in
-      Computation.cancel computation exn_bt;
+      Computation.cancel target exn_bt;
       Exn_bt.raise exn_bt
 
 let guard create_event =
-  let request computation to_result =
-    request_1_as computation to_result No (create_event ())
+  let request target to_result =
+    request_1_as target to_result No (create_event ())
+  in
+  Request { request }
+
+let[@alert "-handler"] from_computation source =
+  let request target to_result =
+    let result () = to_result (Computation.await source) in
+    if Computation.is_running source then begin
+      let propagator =
+        Trigger.from_action result target @@ fun _ result target ->
+        Computation.return target result
+      in
+      if Computation.try_attach source propagator then begin
+        let detacher =
+          Trigger.from_action propagator source @@ fun _ propagator source ->
+          Computation.detach source propagator
+        in
+        if not (Computation.try_attach target detacher) then
+          Computation.detach source propagator
+      end
+      else Computation.return target result
+    end
+    else Computation.return target result
   in
   Request { request }
 

lib/picos_sync/picos_sync.mli

@@ -254,6 +254,13 @@ module Event : sig
   val from_request : 'a request -> 'a t
   (** [from_request { request }] creates an {{!Event} event} from the request
       function. *)
+
+  val from_computation : 'a Computation.t -> 'a t
+  (** [from_computation source] creates an {{!Event} event} that can be
+      committed to once the given [source] computation has completed.
+
+      ℹ️ Committing to some other event does not cancel the [source]
+      computation. *)
 end
 
 (** {1 Examples}

test/test_structured.ml

@@ -161,6 +161,17 @@ let test_can_wait_promises () =
   in
   assert (Promise.await promise = 42)
 
+let test_can_select_promises () =
+  Test_scheduler.run ~max_domains:2 @@ fun () ->
+  Bundle.join_after @@ fun bundle ->
+  let a =
+    Bundle.fork_as_promise bundle @@ fun () ->
+    Control.sleep ~seconds:0.1;
+    42
+  and b = Bundle.fork_as_promise bundle @@ fun () -> Event.select [] in
+  assert (Event.select [ Promise.completed a; Promise.completed b ] = 42);
+  Bundle.terminate bundle
+
 let test_any_and_all_errors () =
   [ Run.all; Run.any ]
   |> List.iter @@ fun run_op ->
@@ -255,6 +266,7 @@ let () =
         Alcotest.test_case "error in promise terminates" `Quick
           test_error_in_promise_terminates;
         Alcotest.test_case "can wait promises" `Quick test_can_wait_promises;
+        Alcotest.test_case "can select promises" `Quick test_can_select_promises;
         Alcotest.test_case "any and all errors" `Quick test_any_and_all_errors;
         Alcotest.test_case "any and all returns" `Quick test_any_and_all_returns;
         Alcotest.test_case "race any" `Quick test_race_any;

test/test_sync.ml

@@ -203,6 +203,34 @@ let test_lazy_cancelation () =
   | _ -> assert false
   | exception Exit -> Trigger.signal to_await
 
+let test_event_basics () =
+  Test_scheduler.run ~max_domains:2 @@ fun () ->
+  assert (
+    Event.select
+      [
+        Event.from_computation (Computation.create ());
+        Event.from_computation (Computation.returned 51) |> Event.map (( + ) 50);
+      ]
+    = 101);
+  begin
+    let c = Computation.create () in
+    Computation.cancel_after c ~seconds:0.1 (Exn_bt.get_callstack 0 Not_found);
+    match Event.sync (Event.from_computation c) with
+    | () -> assert false
+    | exception Not_found -> ()
+  end;
+  begin
+    match
+      [
+        Event.guard (fun () -> raise Exit);
+        Event.from_computation (Computation.returned 42);
+      ]
+      |> Event.choose |> Event.sync
+    with
+    | _ -> assert false
+    | exception Exit -> ()
+  end
+
 let () =
   [
     ( "Mutex and Condition",
@@ -217,5 +245,6 @@ let () =
         Alcotest.test_case "basics" `Quick test_lazy_basics;
         Alcotest.test_case "cancelation" `Quick test_lazy_cancelation;
       ] );
+    ("Event", [ Alcotest.test_case "basics" `Quick test_event_basics ]);
   ]
   |> Alcotest.run "Picos_sync"