Split promises into their own library

Makefile

@@ -1,2 +1,2 @@
 all:
-	dune build @runtest @install
+	dune build @runtest @all

dune-project

@@ -3,7 +3,7 @@
 (generate_opam_files true)
 (source (github ocaml-multicore/eioio))
 (license ISC)
-(authors "Anil Madhavapeddy")
+(authors "Anil Madhavapeddy" "Thomas Leonard")
 (maintainers "anil@recoil.org")
 (package
  (name eunix)
@@ -11,3 +11,7 @@
  (description "An effect-based IO implementation for multicore OCaml with fibres.")
  (depends
 ))
+(package
+ (name promise)
+ (synopsis "effect-based promises")
+ (description "An effect-based implementation of promises."))

eunix.opam

@@ -4,7 +4,7 @@ synopsis: "effect-based direct-style IO for OCaml"
 description:
   "An effect-based IO implementation for multicore OCaml with fibres."
 maintainer: ["anil@recoil.org"]
-authors: ["Anil Madhavapeddy"]
+authors: ["Anil Madhavapeddy" "Thomas Leonard"]
 license: "ISC"
 homepage: "https://github.com/ocaml-multicore/eioio"
 bug-reports: "https://github.com/ocaml-multicore/eioio/issues"

lib_eioio/dune

@@ -2,5 +2,4 @@
  (name eunix)
  (public_name eunix)
  (modules eunix zzz)
- (libraries unix uring bheap logs fmt bigstringaf))
+ (libraries promise unix uring bheap logs fmt bigstringaf))
-

lib_eioio/eunix.ml

@@ -17,48 +17,6 @@
 let src = Logs.Src.create "eunix" ~doc:"Effect-based IO system"
 module Log = (val Logs.src_log src : Logs.LOG)
 
-module Promise = struct
-  type 'a state =
-    | Unresolved of ((('a, exn) result -> unit) Queue.t)
-    | Fulfilled of 'a
-    | Broken of exn
-
-  type 'a t = 'a state ref
-
-  type 'a u = 'a t
-
-  effect Await : 'a t -> 'a
-
-  let create () =
-    let t = ref (Unresolved (Queue.create ())) in
-    t, t
-
-  let await t =
-    perform (Await t)
-
-  let fulfill t v =
-    match !t with
-    | Broken ex -> Fmt.failwith "Can't fulfill already-broken promise: %a" Fmt.exn ex
-    | Fulfilled _ -> Fmt.failwith "Can't fulfill already-fulfilled promise"
-    | Unresolved q ->
-      t := Fulfilled v;
-      Queue.iter (fun f -> f (Ok v)) q
-
-  let break t ex =
-    match !t with
-    | Broken orig -> Fmt.failwith "Can't break already-broken promise: %a -> %a" Fmt.exn orig Fmt.exn ex
-    | Fulfilled _ -> Fmt.failwith "Can't break already-fulfilled promise (with %a)" Fmt.exn ex
-    | Unresolved q ->
-      t := Broken ex;
-      Queue.iter (fun f -> f (Error ex)) q
-
-  let state t =
-    match !t with
-    | Unresolved _ -> `Unresolved
-    | Fulfilled x -> `Fulfilled x
-    | Broken ex -> `Broken ex
-end
-
 type amount = Exactly of int | Upto of int
 
 type rw_req = {
@@ -292,18 +250,13 @@ let run ?(queue_depth=64) ?(block_size=4096) main =
     | effect (Sleep d) k ->
       Zzz.sleep sleep_q d k;
       schedule st
-    | effect (Promise.Await p) k ->
+    | effect (Promise.Await q) k ->
-      begin match !p with
+      let when_resolved = function
-        | Fulfilled v -> continue k v
+        | Ok v -> enqueue_thread st k v
-        | Broken ex -> discontinue k ex
+        | Error ex -> enqueue_failed_thread st k ex
-        | Unresolved q ->
+      in
-          let when_resolved = function
+      Promise.add_waiter q when_resolved;
-            | Ok v -> enqueue_thread st k v
+      schedule st
-            | Error ex -> enqueue_failed_thread st k ex
-          in
-          Queue.add when_resolved q;
-          schedule st
-      end
     | effect (Fork f) k ->
       let promise, resolver = Promise.create () in
       enqueue_thread st k promise;

lib_eioio/eunix.mli

@@ -16,33 +16,6 @@
 
 type t
 
-module Promise : sig
-  type 'a t
-  (** An ['a t] is a promise for a value of type ['a]. *)
-
-  type 'a u
-  (** An ['a u] is a resolver for a promise of type ['a]. *)
-
-  val create : unit -> 'a t * 'a u
-  (** [create ()] is a fresh promise/resolver pair.
-      The promise is initially unresolved. *)
-
-  val await : 'a t -> 'a
-  (** [await t] blocks until [t] is resolved.
-      If [t] is already resolved then this returns immediately.
-      If [t] is broken, it raises the exception. *)
-
-  val fulfill : 'a u -> 'a -> unit
-  (** [fulfill u v] successfully resolves [u]'s promise with the value [v].
-      Any threads waiting for the result will be added to the run queue. *)
-
-  val break : 'a u -> exn -> unit
-  (** [break u ex] resolves [u]'s promise with the exception [ex].
-      Any threads waiting for the result will be added to the run queue. *)
-
-  val state : 'a t -> [ `Fulfilled of 'a | `Broken of exn | `Unresolved ]
-end
-
 (** {1 Fibre functions} *)
 
 val fork : (unit -> 'a) -> 'a Promise.t

lib_promise/dune

@@ -0,0 +1,4 @@
+(library
+ (name promise)
+ (public_name promise)
+ (flags (:standard -w -50)))

lib_promise/promise.ml

@@ -0,0 +1,46 @@
+type 'a waiters = (('a, exn) result -> unit) Queue.t
+
+type 'a state =
+  | Unresolved of 'a waiters
+  | Fulfilled of 'a
+  | Broken of exn
+
+type 'a t = 'a state ref
+
+type 'a u = 'a t
+
+effect Await : 'a waiters -> 'a
+
+let create () =
+  let t = ref (Unresolved (Queue.create ())) in
+  t, t
+
+let await t =
+  match !t with
+  | Fulfilled x -> x
+  | Broken ex -> raise ex
+  | Unresolved q ->
+    perform (Await q)
+
+let fulfill t v =
+  match !t with
+  | Broken ex -> invalid_arg ("Can't fulfill already-broken promise: " ^ Printexc.to_string ex)
+  | Fulfilled _ -> invalid_arg "Can't fulfill already-fulfilled promise"
+  | Unresolved q ->
+    t := Fulfilled v;
+    Queue.iter (fun f -> f (Ok v)) q
+
+let break t ex =
+  match !t with
+  | Broken orig -> invalid_arg (Printf.sprintf "Can't break already-broken promise: %s -> %s"
+                                  (Printexc.to_string orig) (Printexc.to_string ex))
+  | Fulfilled _ -> invalid_arg (Printf.sprintf "Can't break already-fulfilled promise (with %s)"
+                                  (Printexc.to_string ex))
+  | Unresolved q ->
+    t := Broken ex;
+    Queue.iter (fun f -> f (Error ex)) q
+
+let state t = !t
+
+let add_waiter waiters cb =
+  Queue.add cb waiters

lib_promise/promise.mli

@@ -0,0 +1,39 @@
+type 'a t
+(** An ['a t] is a promise for a value of type ['a]. *)
+
+type 'a u
+(** An ['a u] is a resolver for a promise of type ['a]. *)
+
+val create : unit -> 'a t * 'a u
+(** [create ()] is a fresh promise/resolver pair.
+    The promise is initially unresolved. *)
+
+val await : 'a t -> 'a
+(** [await t] blocks until [t] is resolved.
+    If [t] is already resolved then this returns immediately.
+    If [t] is broken, it raises the exception. *)
+
+val fulfill : 'a u -> 'a -> unit
+(** [fulfill u v] successfully resolves [u]'s promise with the value [v].
+    Any threads waiting for the result will be added to the run queue. *)
+
+val break : 'a u -> exn -> unit
+(** [break u ex] resolves [u]'s promise with the exception [ex].
+    Any threads waiting for the result will be added to the run queue. *)
+
+type 'a waiters
+
+type 'a state =
+  | Unresolved of 'a waiters
+  | Fulfilled of 'a
+  | Broken of exn
+
+val state : 'a t -> 'a state
+
+(** {2 Provider API} *)
+
+val add_waiter : 'a waiters -> (('a, exn) result -> unit) -> unit
+
+effect Await : 'a waiters -> 'a
+(** Performed when the user calls [await] on an unresolved promise.
+    The handler should add itself to the list of waiters and block until its callback is invoked. *)

promise.opam

@@ -0,0 +1,28 @@
+# This file is generated by dune, edit dune-project instead
+opam-version: "2.0"
+synopsis: "effect-based promises"
+description: "An effect-based implementation of promises."
+maintainer: ["anil@recoil.org"]
+authors: ["Anil Madhavapeddy" "Thomas Leonard"]
+license: "ISC"
+homepage: "https://github.com/ocaml-multicore/eioio"
+bug-reports: "https://github.com/ocaml-multicore/eioio/issues"
+depends: [
+  "dune" {>= "2.7"}
+  "odoc" {with-doc}
+]
+build: [
+  ["dune" "subst"] {dev}
+  [
+    "dune"
+    "build"
+    "-p"
+    name
+    "-j"
+    jobs
+    "@install"
+    "@runtest" {with-test}
+    "@doc" {with-doc}
+  ]
+]
+dev-repo: "git+https://github.com/ocaml-multicore/eioio.git"

tests/eurcp_lib.ml

@@ -18,7 +18,7 @@ let copy_file infd outfd insize block_size =
        let len = min block_size remaining in
        let thread = U.fork (fun () -> read_then_write_chunk infd outfd file_offset len) in
        copy_block (file_offset + len);
-       U.Promise.await thread
+       Promise.await thread
   in
   copy_block 0
 

tests/test.ml

@@ -14,29 +14,35 @@ let state t =
        | `Fulfilled x -> Fmt.pf f "fulfilled:%a" (Alcotest.pp t) x
     )
 
+let get_state p =
+  match Promise.state p with
+  | Unresolved _ -> `Unresolved
+  | Broken ex -> `Broken ex
+  | Fulfilled x -> `Fulfilled x
+
 let test_promise () =
   Eunix.run @@ fun () ->
   let p, r = Promise.create () in
-  Alcotest.(check (state string)) "Initially unresolved" (Promise.state p) `Unresolved;
+  Alcotest.(check (state string)) "Initially unresolved" (get_state p) `Unresolved;
   let thread = Eunix.fork (fun () -> Promise.await p) in
   Promise.fulfill r "ok";
-  Alcotest.(check (state string)) "Resolved OK" (Promise.state p) (`Fulfilled "ok");
+  Alcotest.(check (state string)) "Resolved OK" (get_state p) (`Fulfilled "ok");
-  Alcotest.(check (state string)) "Thread unresolved" (Promise.state thread) `Unresolved;
+  Alcotest.(check (state string)) "Thread unresolved" (get_state thread) `Unresolved;
   yield ();
-  Alcotest.(check (state string)) "Thread resolved" (Promise.state thread) @@ `Fulfilled "ok";
+  Alcotest.(check (state string)) "Thread resolved" (get_state thread) @@ `Fulfilled "ok";
   let result = Promise.await thread in
   Alcotest.(check string) "Await result" result "ok"
 
 let test_promise_exn () =
   Eunix.run @@ fun () ->
   let p, r = Promise.create () in
-  Alcotest.(check (state reject)) "Initially unresolved" (Promise.state p) `Unresolved;
+  Alcotest.(check (state reject)) "Initially unresolved" (get_state p) `Unresolved;
   let thread = Eunix.fork (fun () -> Promise.await p) in
   Promise.break r (Failure "test");
-  Alcotest.(check (state reject)) "Broken" (Promise.state p) @@ `Broken (Failure "test");
+  Alcotest.(check (state reject)) "Broken" (get_state p) @@ `Broken (Failure "test");
-  Alcotest.(check (state reject)) "Thread unresolved" (Promise.state thread) `Unresolved;
+  Alcotest.(check (state reject)) "Thread unresolved" (get_state thread) `Unresolved;
   yield ();
-  Alcotest.(check (state reject)) "Thread broken" (Promise.state thread) @@ `Broken (Failure "test");
+  Alcotest.(check (state reject)) "Thread broken" (get_state thread) @@ `Broken (Failure "test");
   match Promise.await thread with
   | `Cant_happen -> assert false
   | exception (Failure msg) -> Alcotest.(check string) "Await result" msg "test"
@@ -58,7 +64,7 @@ let test_poll_add () =
   Eunix.await_writable w;
   let sent = Unix.write w (Bytes.of_string "!") 0 1 in
   assert (sent = 1);
-  let result = Eunix.Promise.await thread in
+  let result = Promise.await thread in
   Alcotest.(check string) "Received data" "!" result
 
 let test_poll_add_busy () =
@@ -69,9 +75,9 @@ let test_poll_add_busy () =
   Eunix.yield ();
   let sent = Unix.write w (Bytes.of_string "!!") 0 2 in
   assert (sent = 2);
-  let a = Eunix.Promise.await a in
+  let a = Promise.await a in
   Alcotest.(check string) "Received data" "!" a;
-  let b = Eunix.Promise.await b in
+  let b = Promise.await b in
   Alcotest.(check string) "Received data" "!" b
 
 let () =