Merge pull request #567 from talex5/single-waiter

Move Waiters out of Eio_core

Author: talex5

lib_eio/core/eio__core.ml

@@ -7,7 +7,6 @@ module Private = struct
   module Suspend = Suspend
   module Cells = Cells
   module Broadcast = Broadcast
-  module Waiters = Waiters
   module Ctf = Ctf
   module Fiber_context = Cancel.Fiber_context
   module Debug = Debug

lib_eio/core/eio__core.mli

@@ -706,51 +706,6 @@ module Private : sig
         because you need to unlock a mutex if cancelled). *)
   end
 
-  (** A queue of fibers waiting for an event. *)
-  module Waiters : sig
-    type 'a t
-    (* A queue of fibers waiting for something.
-       Note: an [_ t] is not thread-safe itself.
-       To use share it between domains, the user is responsible for wrapping it in a mutex. *)
-
-    val create : unit -> 'a t
-
-    val wake_all : 'a t -> 'a -> unit
-    (** [wake_all t] calls (and removes) all the functions waiting on [t].
-        If [t] is shared between domains, the caller must hold the mutex while calling this. *)
-
-    val wake_one : 'a t -> 'a -> [`Ok | `Queue_empty]
-    (** [wake_one t] is like {!wake_all}, but only calls (and removes) the first waiter in the queue.
-        If [t] is shared between domains, the caller must hold the mutex while calling this. *)
-
-    val is_empty : 'a t -> bool
-    (** [is_empty t] checks whether there are any functions waiting on [t].
-        If [t] is shared between domains, the caller must hold the mutex while calling this,
-        and the result is valid until the mutex is released. *)
-
-    val await :
-      mutex:Mutex.t option ->
-      'a t -> Ctf.id -> 'a
-    (** [await ~mutex t id] suspends the current fiber and adds its continuation to [t].
-        When the waiter is woken, the fiber is resumed and returns the result.
-        If [t] can be used from multiple domains:
-        - [mutex] must be set to the mutex to use to unlock it.
-        - [mutex] must be already held when calling this function, which will unlock it before blocking.
-        When [await] returns, [mutex] will have been unlocked.
-        @raise Cancel.Cancelled if the fiber's context is cancelled *)
-
-    val await_internal :
-      mutex:Mutex.t option ->
-      'a t -> Ctf.id -> Fiber_context.t ->
-      (('a, exn) result -> unit) -> unit
-    (** [await_internal ~mutex t id ctx enqueue] is like [await], but the caller has to suspend the fiber.
-        This also allows wrapping the [enqueue] function.
-        Calls [enqueue (Error (Cancelled _))] if cancelled.
-        Note: [enqueue] is called from the triggering domain,
-              which is currently calling {!wake_one} or {!wake_all}
-              and must therefore be holding [mutex]. *)
-  end
-
   module Debug : sig
     val traceln :
       ?__POS__:string * int * int * int ->

lib_eio/core/single_waiter.ml

@@ -1,4 +1,6 @@
-(* A simplified version of [Waiters] that can only handle one waiter and is not thread-safe. *)
+(* Allows a single fiber to wait to be notified by another fiber in the same domain.
+   If multiple fibers need to wait at once, or the notification comes from another domain,
+   this can't be used. *)
 
 type 'a t = {
   mutable wake : ('a, exn) result -> unit;

lib_eio/core/switch.ml

@@ -4,7 +4,7 @@ type t = {
   mutable daemon_fibers : int;
   mutable exs : (exn * Printexc.raw_backtrace) option;
   on_release : (unit -> unit) Lwt_dllist.t;
-  waiter : unit Waiters.t;              (* The main [top]/[sub] function may wait here for fibers to finish. *)
+  waiter : unit Single_waiter.t;              (* The main [top]/[sub] function may wait here for fibers to finish. *)
   cancel : Cancel.t;
 }
 
@@ -68,7 +68,7 @@ let dec_fibers t =
   if t.daemon_fibers > 0 && t.fibers = t.daemon_fibers then
     Cancel.cancel t.cancel Exit;
   if t.fibers = 0 then
-    Waiters.wake_all t.waiter ()
+    Single_waiter.wake t.waiter (Ok ())
 
 let with_op t fn =
   inc_fibers t;
@@ -92,7 +92,7 @@ let rec await_idle t =
   (* Wait for fibers to finish: *)
   while t.fibers > 0 do
     Ctf.note_try_read t.id;
-    Waiters.await ~mutex:None t.waiter t.id
+    Single_waiter.await t.waiter t.id
   done;
   (* Call on_release handlers: *)
   let queue = Lwt_dllist.create () in
@@ -125,7 +125,7 @@ let create cancel =
     fibers = 1;         (* The main function counts as a fiber *)
     daemon_fibers = 0;
     exs = None;
-    waiter = Waiters.create ();
+    waiter = Single_waiter.create ();
     on_release = Lwt_dllist.create ();
     cancel;
   }

lib_eio/core/waiters.mli

@@ -38,16 +38,16 @@ let rec wake_one t v =
 
 let is_empty = Lwt_dllist.is_empty
 
-let await_internal ~mutex (t:'a t) id (ctx:Cancel.fiber_context) enqueue =
-  match Cancel.Fiber_context.get_error ctx with
+let await_internal ~mutex (t:'a t) id ctx enqueue =
+  match Fiber_context.get_error ctx with
   | Some ex ->
     Option.iter Mutex.unlock mutex;
     enqueue (Error ex)
   | None ->
     let resolved_waiter = ref Hook.null in
     let finished = Atomic.make false in
     let enqueue x =
-      Ctf.note_read ~reader:id ctx.tid;
+      Ctf.note_read ~reader:id (Fiber_context.tid ctx);
       enqueue x
     in
     let cancel ex =
@@ -56,7 +56,7 @@ let await_internal ~mutex (t:'a t) id (ctx:Cancel.fiber_context) enqueue =
         enqueue (Error ex)
       )
     in
-    Cancel.Fiber_context.set_cancel_fn ctx cancel;
+    Fiber_context.set_cancel_fn ctx cancel;
     let waiter = { enqueue; finished } in
     match mutex with
     | None ->

lib_eio/core/hook.ml renamed to lib_eio/hook.ml

@@ -0,0 +1,42 @@
+(** A queue of fibers waiting for an event. *)
+type 'a t
+(* A queue of fibers waiting for something.
+   Note: an [_ t] is not thread-safe itself.
+   To use share it between domains, the user is responsible for wrapping it in a mutex. *)
+
+val create : unit -> 'a t
+
+val wake_all : 'a t -> 'a -> unit
+(** [wake_all t] calls (and removes) all the functions waiting on [t].
+    If [t] is shared between domains, the caller must hold the mutex while calling this. *)
+
+val wake_one : 'a t -> 'a -> [`Ok | `Queue_empty]
+(** [wake_one t] is like {!wake_all}, but only calls (and removes) the first waiter in the queue.
+    If [t] is shared between domains, the caller must hold the mutex while calling this. *)
+
+val is_empty : 'a t -> bool
+(** [is_empty t] checks whether there are any functions waiting on [t].
+    If [t] is shared between domains, the caller must hold the mutex while calling this,
+    and the result is valid until the mutex is released. *)
+
+val await :
+  mutex:Mutex.t option ->
+  'a t -> Ctf.id -> 'a
+(** [await ~mutex t id] suspends the current fiber and adds its continuation to [t].
+    When the waiter is woken, the fiber is resumed and returns the result.
+    If [t] can be used from multiple domains:
+    - [mutex] must be set to the mutex to use to unlock it.
+    - [mutex] must be already held when calling this function, which will unlock it before blocking.
+    When [await] returns, [mutex] will have been unlocked.
+    @raise Cancel.Cancelled if the fiber's context is cancelled *)
+
+val await_internal :
+  mutex:Mutex.t option ->
+  'a t -> Ctf.id -> Fiber_context.t ->
+  (('a, exn) result -> unit) -> unit
+(** [await_internal ~mutex t id ctx enqueue] is like [await], but the caller has to suspend the fiber.
+    This also allows wrapping the [enqueue] function.
+    Calls [enqueue (Error (Cancelled _))] if cancelled.
+    Note: [enqueue] is called from the triggering domain,
+          which is currently calling {!wake_one} or {!wake_all}
+          and must therefore be holding [mutex]. *)