Repository: ocaml-multicore/picos
Pull request #204 (2024-08-05 13:37:07)

Add an implicitly propagated flock of fibers
This PR adds to the `picos.structured` library a new `Flock` module, which essentially provides a more convenient interface to the existing `Bundle` mechanism by passing the underlying `Bundle` through `FLS`.

The naming is a bit tongue in cheek.  The idea is that, unlike a "[bundle](https://www.merriam-webster.com/dictionary/bundle)", a "[flock](https://www.merriam-webster.com/dictionary/flock)" (like a flock of birds) is something that you cannot really hold in your hands.   Flock is also a term used in conjuction with fibers.


Patch:
diff --git a/bench/bench_bounded_q.ml b/bench/bench_bounded_q.ml
index 0276536..ec48684 100644
--- a/bench/bench_bounded_q.ml
+++ b/bench/bench_bounded_q.ml
@@ -118,8 +118,8 @@ let run_one ~budgetf ~n_adders ~n_takers ?(n_msgs = 10 * Util.iter_factor) () =
   in
   let wrap _ () = Scheduler.run in
   let work domain_index () =
-    Bundle.join_after @@ fun bundle ->
-    Bundle.fork bundle yielder;
+    Flock.join_after @@ fun () ->
+    Flock.fork yielder;
     if domain_index < n_adders then
       let rec work () =
         let n = Countdown.alloc n_msgs_to_add ~domain_index ~batch:100 in
@@ -129,7 +129,7 @@ let run_one ~budgetf ~n_adders ~n_takers ?(n_msgs = 10 * Util.iter_factor) () =
           done;
           work ()
         end
-        else Bundle.terminate bundle
+        else Flock.terminate ()
       in
       work ()
     else
@@ -141,7 +141,7 @@ let run_one ~budgetf ~n_adders ~n_takers ?(n_msgs = 10 * Util.iter_factor) () =
           done;
           work ()
         end
-        else Bundle.terminate bundle
+        else Flock.terminate ()
       in
       work ()
   in
diff --git a/bench/bench_cancel_after.ml b/bench/bench_cancel_after.ml
index af54630..7896275 100644
--- a/bench/bench_cancel_after.ml
+++ b/bench/bench_cancel_after.ml
@@ -45,7 +45,7 @@ let run_async ~budgetf ~n_domains () =
   let init _ = Countdown.non_atomic_set n_ops_todo n_ops in
   let wrap _ () = Scheduler.run in
   let work domain_index () =
-    Bundle.join_after @@ fun bundle ->
+    Flock.join_after @@ fun () ->
     let queue = Queue.create () in
     let exit = ref false in
 
@@ -56,7 +56,7 @@ let run_async ~budgetf ~n_domains () =
           Computation.wait computation;
           awaiter ()
     in
-    Bundle.fork bundle awaiter;
+    Flock.fork awaiter;
 
     let rec work () =
       let n = Countdown.alloc n_ops_todo ~domain_index ~batch:10 in
diff --git a/bench/bench_mutex.ml b/bench/bench_mutex.ml
index 35da445..eae138d 100644
--- a/bench/bench_mutex.ml
+++ b/bench/bench_mutex.ml
@@ -26,7 +26,7 @@ let run_one ~budgetf ~n_fibers ~use_domains () =
   in
   let wrap _ () = Scheduler.run in
   let work domain_index () =
-    Bundle.join_after @@ fun bundle ->
+    Flock.join_after @@ fun () ->
     let rec work () =
       let n = Countdown.alloc n_ops_todo ~domain_index ~batch in
       if n <> 0 then
@@ -46,13 +46,13 @@ let run_one ~budgetf ~n_fibers ~use_domains () =
         loop n
     in
     if use_domains then begin
-      Bundle.fork bundle yielder;
+      Flock.fork yielder;
       work ();
-      Bundle.terminate bundle
+      Flock.terminate ()
     end
     else
       for _ = 1 to n_fibers do
-        Bundle.fork bundle work
+        Flock.fork work
       done
   in
 
diff --git a/bench/bench_yield.ml b/bench/bench_yield.ml
index 80440de..34131bb 100644
--- a/bench/bench_yield.ml
+++ b/bench/bench_yield.ml
@@ -12,14 +12,14 @@ let run_one ~budgetf ~n_fibers () =
   let init _ = () in
   let wrap _ () = Scheduler.run in
   let work _ () =
-    Bundle.join_after @@ fun bundle ->
+    Flock.join_after @@ fun () ->
     let main () =
       for _ = 1 to n_yields_per_fiber do
         Control.yield ()
       done
     in
     for _ = 1 to n_fibers do
-      Bundle.fork bundle main
+      Flock.fork main
     done
   in
   let config =
diff --git a/lib/picos/picos.mli b/lib/picos/picos.mli
index 8927fe8..20b501b 100644
--- a/lib/picos/picos.mli
+++ b/lib/picos/picos.mli
@@ -153,11 +153,11 @@ module Trigger : sig
 
       {[
         run begin fun () ->
-          Bundle.join_after @@ fun bundle ->
+          Flock.join_after @@ fun () ->
 
           let trigger = Trigger.create () in
 
-          Bundle.fork bundle begin fun () ->
+          Flock.fork begin fun () ->
             Trigger.signal trigger
           end;
 
@@ -428,20 +428,20 @@ module Computation : sig
 
       {[
         run begin fun () ->
-          Bundle.join_after @@ fun bundle ->
+          Flock.join_after @@ fun () ->
 
           let computation =
             Computation.create ()
           in
 
-          let canceler = Bundle.fork_as_promise bundle @@ fun () ->
+          let canceler = Flock.fork_as_promise @@ fun () ->
             Fiber.sleep ~seconds:1.0;
 
             Computation.cancel computation
               (Exn_bt.get_callstack 0 Exit)
           in
 
-          Bundle.fork bundle begin fun () ->
+          Flock.fork begin fun () ->
             let rec fib i =
               Computation.check
                 computation;
diff --git a/lib/picos_finally/picos_finally.mli b/lib/picos_finally/picos_finally.mli
index 71c2432..96ba0e8 100644
--- a/lib/picos_finally/picos_finally.mli
+++ b/lib/picos_finally/picos_finally.mli
@@ -85,7 +85,7 @@ val move : 'r instance -> ('r -> 'a) -> 'a
 
     {[
       let recursive_server server_fd =
-        Bundle.join_after @@ fun bundle ->
+        Flock.join_after @@ fun () ->
 
         (* recursive server *)
         let rec accept () =
@@ -96,12 +96,12 @@ val move : 'r instance -> ('r -> 'a) -> 'a
           in
 
           (* fork to accept other clients *)
-          Bundle.fork bundle accept;
+          Flock.fork accept;
 
           (* handle this client... omitted *)
           ()
         in
-        Bundle.fork bundle accept
+        Flock.fork accept
     ]}
 
     {2 Looping server}
@@ -114,7 +114,7 @@ val move : 'r instance -> ('r -> 'a) -> 'a
 
     {[
       let looping_server server_fd =
-        Bundle.join_after @@ fun bundle ->
+        Flock.join_after @@ fun () ->
 
         (* loop to accept clients *)
         while true do
@@ -125,7 +125,7 @@ val move : 'r instance -> ('r -> 'a) -> 'a
           in
 
           (* fork to handle this client *)
-          Bundle.fork bundle @@ fun () ->
+          Flock.fork @@ fun () ->
             let@ client_fd = move client_fd in
 
             (* handle client... omitted *)
@@ -142,13 +142,13 @@ val move : 'r instance -> ('r -> 'a) -> 'a
 
     {[
       let move_from_child_to_parent () =
-        Bundle.join_after @@ fun bundle ->
+        Flock.join_after @@ fun () ->
 
         (* for communicating a resource *)
         let shared_ivar = Ivar.create () in
 
         (* fork a child that creates a resource *)
-        Bundle.fork bundle begin fun () ->
+        Flock.fork begin fun () ->
           let pretend_release () = ()
           and pretend_acquire () = () in
 
diff --git a/lib/picos_select/picos_select.mli b/lib/picos_select/picos_select.mli
index a8c41a1..a4df5cc 100644
--- a/lib/picos_select/picos_select.mli
+++ b/lib/picos_select/picos_select.mli
@@ -186,8 +186,8 @@ val check_configured : unit -> unit
           assert (w = 1)
         in
 
-        Bundle.join_after begin fun bundle ->
-          Bundle.fork bundle begin fun () ->
+        Flock.join_after begin fun () ->
+          Flock.fork begin fun () ->
             while true do
               Event.select [
                 Picos_select.on msg_inn1 `R
@@ -217,7 +217,7 @@ val check_configured : unit -> unit
           write1 msg_out2;
           read1 syn_inn;
 
-          Bundle.terminate bundle
+          Flock.terminate ()
         end
       Inn1
       Inn2
diff --git a/lib/picos_stdio/picos_stdio.mli b/lib/picos_stdio/picos_stdio.mli
index bdf0faf..7efd659 100644
--- a/lib/picos_stdio/picos_stdio.mli
+++ b/lib/picos_stdio/picos_stdio.mli
@@ -718,8 +718,8 @@ end
         Unix.set_nonblock syn_i;
         Unix.set_nonblock syn_o;
 
-        Bundle.join_after begin fun bundle ->
-          Bundle.fork bundle begin fun () ->
+        Flock.join_after begin fun () ->
+          Flock.fork begin fun () ->
             let bytes = Bytes.create 100 in
             while true do
               let n =
@@ -753,7 +753,7 @@ end
           send_string "Hello, world!";
           send_string "POSIX with OCaml";
 
-          Bundle.terminate bundle
+          Flock.terminate ()
         end
       Hello, world!
       POSIX with OCaml
diff --git a/lib/picos_structured/bundle.ml b/lib/picos_structured/bundle.ml
index 8ca94ac..f82f9a3 100644
--- a/lib/picos_structured/bundle.ml
+++ b/lib/picos_structured/bundle.ml
@@ -2,83 +2,109 @@ open Picos
 
 let[@inline never] completed () = invalid_arg "already completed"
 
-type t = {
-  num_fibers : int Atomic.t;
-  bundle : unit Computation.t;
-  errors : Control.Errors.t;
-  finished : Trigger.t;
-}
-
-let terminate ?callstack t =
-  Computation.cancel t.bundle (Control.terminate_bt ?callstack ())
-
-let terminate_after ?callstack t ~seconds =
-  Computation.cancel_after t.bundle ~seconds
+type _ tdt =
+  | Nothing : [> `Nothing ] tdt
+  | Bundle : {
+      num_fibers : int Atomic.t;
+      bundle : unit Computation.t;
+      errors : Control.Errors.t;
+      finished : Trigger.t;
+    }
+      -> [> `Bundle ] tdt
+
+let flock_key : [ `Bundle | `Nothing ] tdt Fiber.FLS.key =
+  Fiber.FLS.new_key (Constant Nothing)
+
+type t = [ `Bundle ] tdt
+
+let terminate ?callstack (Bundle r : t) =
+  Computation.cancel r.bundle (Control.terminate_bt ?callstack ())
+
+let terminate_after ?callstack (Bundle r : t) ~seconds =
+  Computation.cancel_after r.bundle ~seconds
     (Control.terminate_bt ?callstack ())
 
-let error ?callstack t (exn_bt : Exn_bt.t) =
+let error ?callstack (Bundle r as t : t) (exn_bt : Exn_bt.t) =
   if exn_bt.Exn_bt.exn != Control.Terminate then begin
     terminate ?callstack t;
-    Control.Errors.push t.errors exn_bt
+    Control.Errors.push r.errors exn_bt
   end
 
-let decr t =
-  let n = Atomic.fetch_and_add t.num_fibers (-1) in
+let decr (Bundle r : t) =
+  let n = Atomic.fetch_and_add r.num_fibers (-1) in
   if n = 1 then begin
-    Computation.finish t.bundle;
-    Trigger.signal t.finished
+    Computation.finish r.bundle;
+    Trigger.signal r.finished
   end
 
-let await t fiber packed canceler =
+type _ pass = FLS : unit pass | Arg : t pass
+
+let await (type a) (Bundle r as t : t) fiber packed canceler outer
+    (pass : a pass) =
   decr t;
   Fiber.set_computation fiber packed;
   let forbid = Fiber.exchange fiber ~forbid:true in
-  Trigger.await t.finished |> ignore;
+  Trigger.await r.finished |> ignore;
   Fiber.set fiber ~forbid;
+  begin
+    match pass with FLS -> Fiber.FLS.set fiber flock_key outer | Arg -> ()
+  end;
   let (Packed parent) = packed in
   Computation.detach parent canceler;
-  Control.Errors.check t.errors;
+  Control.Errors.check r.errors;
   Fiber.check fiber
 
-let join_after fn =
+let join_after_pass (type a) (fn : a -> _) (pass : a pass) =
   (* The sequence of operations below ensures that nothing is leaked. *)
-  let t =
+  let (Bundle r as t : t) =
     let num_fibers = Atomic.make 1 in
     let bundle = Computation.create ~mode:`LIFO () in
     let errors = Control.Errors.create () in
     let finished = Trigger.create () in
-    { num_fibers; bundle; errors; finished }
+    Bundle { num_fibers; bundle; errors; finished }
   in
   let fiber = Fiber.current () in
+  let outer =
+    match pass with FLS -> Fiber.FLS.get fiber flock_key | Arg -> Nothing
+  in
   let (Packed parent as packed) = Fiber.get_computation fiber in
-  let bundle = Computation.Packed t.bundle in
-  let canceler = Computation.attach_canceler ~from:parent ~into:t.bundle in
+  let bundle = Computation.Packed r.bundle in
+  let canceler = Computation.attach_canceler ~from:parent ~into:r.bundle in
   (* Ideally there should be no poll point betweem [attach_canceler] and the
      [match ... with] below. *)
-  Fiber.set_computation fiber bundle;
-  match fn t with
+  match
+    Fiber.set_computation fiber bundle;
+    fn (match pass with FLS -> Fiber.FLS.set fiber flock_key t | Arg -> t)
+  with
   | value ->
-      await t fiber packed canceler;
+      await t fiber packed canceler outer pass;
       value
   | exception exn ->
       let exn_bt = Exn_bt.get exn in
       error t exn_bt;
-      await t fiber packed canceler;
+      await t fiber packed canceler outer pass;
       Exn_bt.raise exn_bt
 
-let rec incr t backoff =
-  let before = Atomic.get t.num_fibers in
+let join_after fn = join_after_pass fn Arg
+
+let rec incr (Bundle r as t : t) backoff =
+  let before = Atomic.get r.num_fibers in
   if before = 0 then completed ()
-  else if not (Atomic.compare_and_set t.num_fibers before (before + 1)) then
+  else if not (Atomic.compare_and_set r.num_fibers before (before + 1)) then
     incr t (Backoff.once backoff)
 
-let fork_as_promise t thunk =
+let fork_as_promise_pass (type a) (Bundle r as t : t) thunk (pass : a pass) =
   (* The sequence of operations below ensures that nothing is leaked. *)
   incr t Backoff.default;
   let child = Computation.create ~mode:`LIFO () in
   try
-    let canceler = Computation.attach_canceler ~from:t.bundle ~into:child in
+    let canceler = Computation.attach_canceler ~from:r.bundle ~into:child in
     let main () =
+      begin
+        match pass with
+        | FLS -> Fiber.FLS.set (Fiber.current ()) flock_key t
+        | Arg -> ()
+      end;
       begin
         match thunk () with
         | value -> Computation.return child value
@@ -87,7 +113,7 @@ let fork_as_promise t thunk =
             Computation.cancel child exn_bt;
             error t exn_bt
       end;
-      Computation.detach t.bundle canceler;
+      Computation.detach r.bundle canceler;
       decr t
     in
     Fiber.spawn ~forbid:false child [ main ];
@@ -98,10 +124,11 @@ let fork_as_promise t thunk =
     decr t;
     raise canceled_exn
 
+let fork_as_promise t thunk = fork_as_promise_pass t thunk Arg
 let fork t thunk = fork_as_promise t thunk |> ignore
 
 (* *)
 
-let is_running t = Computation.is_running t.bundle
-let unsafe_incr t = Atomic.incr t.num_fibers
-let unsafe_reset t = Atomic.set t.num_fibers 1
+let is_running (Bundle r : t) = Computation.is_running r.bundle
+let unsafe_incr (Bundle r : t) = Atomic.incr r.num_fibers
+let unsafe_reset (Bundle r : t) = Atomic.set r.num_fibers 1
diff --git a/lib/picos_structured/flock.ml b/lib/picos_structured/flock.ml
new file mode 100644
index 0000000..e5e9d77
--- /dev/null
+++ b/lib/picos_structured/flock.ml
@@ -0,0 +1,17 @@
+open Picos
+
+let[@inline never] no_flock () = invalid_arg "no flock"
+
+let get () =
+  match Fiber.FLS.get (Fiber.current ()) Bundle.flock_key with
+  | Nothing -> no_flock ()
+  | Bundle _ as t -> t
+
+let terminate_after ?callstack ~seconds () =
+  Bundle.terminate_after ?callstack (get ()) ~seconds
+
+let terminate ?callstack () = Bundle.terminate ?callstack (get ())
+let error ?callstack exn_bt = Bundle.error (get ()) ?callstack exn_bt
+let fork_as_promise thunk = Bundle.fork_as_promise_pass (get ()) thunk FLS
+let fork action = fork_as_promise action |> ignore
+let join_after fn = Bundle.join_after_pass fn Bundle.FLS
diff --git a/lib/picos_structured/picos_structured.ml b/lib/picos_structured/picos_structured.ml
index c589586..8b65b82 100644
--- a/lib/picos_structured/picos_structured.ml
+++ b/lib/picos_structured/picos_structured.ml
@@ -1,4 +1,5 @@
 module Control = Control
 module Promise = Promise
 module Bundle = Bundle
+module Flock = Flock
 module Run = Run
diff --git a/lib/picos_structured/picos_structured.mli b/lib/picos_structured/picos_structured.mli
index 7d2fa6c..2c962b4 100644
--- a/lib/picos_structured/picos_structured.mli
+++ b/lib/picos_structured/picos_structured.mli
@@ -138,8 +138,9 @@ module Promise : sig
   (** [of_value value] returns a constant completed promise that returns the
       given [value].
 
-      ℹ️ Promises can also be {{!Bundle.fork_as_promise} created} in the scope of
-      a {!Bundle}. *)
+      ℹ️ Promises can also be created in the scope of a
+      {{!Bundle.fork_as_promise} [Bundle]} or a {{!Flock.fork_as_promise}
+      [Flock]}. *)
 
   val await : 'a t -> 'a
   (** [await promise] awaits until the promise has completed and either returns
@@ -181,7 +182,7 @@ module Promise : sig
 end
 
 module Bundle : sig
-  (** A dynamic bundle of fibers guaranteed to be joined at the end.
+  (** An explicit dynamic bundle of fibers guaranteed to be joined at the end.
 
       Bundles allow you to conveniently structure or delimit concurrency into
       nested scopes.  After a bundle returns or raises an exception, no fibers
@@ -244,6 +245,64 @@ module Bundle : sig
       {{!fork_as_promise} [fork_as_promise bundle action |> ignore]}. *)
 end
 
+module Flock : sig
+  (** An implicit dynamic flock of fibers guaranteed to be joined at the end.
+
+      This is essentially a very thin convenience wrapper for an implicitly
+      propagated {!Bundle}. *)
+
+  val join_after : (unit -> 'a) -> 'a
+  (** [join_after scope] creates a new flock for fibers, calls [scope] after
+      setting current flock to the new flock, and restores the previous flock,
+      if any after [scope] exits.  The flock will be implicitly propagated to
+      all fibers {{!fork} forked} into the flock.  A call of [join_after]
+      returns or raises only after [scope] has returned or raised and all
+      {{!fork} forked} fibers have terminated.  If [scope] raises an exception,
+      {!error} will be called.
+
+      ℹ️ When [scope] returns normally, {!terminate} will not be called
+      implicitly. *)
+
+  val terminate : ?callstack:int -> unit -> unit
+  (** [terminate ()] cancels all of the {{!fork} forked} fibers using the
+      {{!Control.Terminate} [Terminate]} exception.  After [terminate] has been
+      called, no new fibers can be forked to the current flock.
+
+      The optional [callstack] argument specifies the number of callstack
+      entries to capture with the {{!Control.Terminate} [Terminate]} exception.
+      The default is [0].
+
+      ℹ️ Calling [terminate] at the end of a flock can be a convenient way to
+      cancel any background fibers started by the flock.
+
+      ℹ️ Calling [terminate] does not raise the {{!Control.Terminate}
+      [Terminate]} exception, but blocking operations after [terminate] will
+      raise the exception to propagate cancelation unless {{!Control.protect}
+      propagation of cancelation is forbidden}. *)
+
+  val terminate_after : ?callstack:int -> seconds:float -> unit -> unit
+  (** [terminate_after ~seconds ()] arranges to {!terminate} the current flock
+      after the specified timeout in [seconds]. *)
+
+  val error : ?callstack:int -> Exn_bt.t -> unit
+  (** [error exn_bt] first calls {!terminate} and then adds the exception with
+      backtrace to the list of exceptions to be raised, unless the exception is
+      the {{!Control.Terminate} [Terminate]} exception, which is not considered
+      to signal an error by itself.
+
+      The optional [callstack] argument is passed to {!terminate}. *)
+
+  val fork_as_promise : (unit -> 'a) -> 'a Promise.t
+  (** [fork_as_promise thunk] spawns a new fiber to the current flock that will
+      run the given [thunk].  The result of the [thunk] will be written to the
+      {{!Promise} promise}.  If the [thunk] raises an exception, {!error} will
+      be called with that exception. *)
+
+  val fork : (unit -> unit) -> unit
+  (** [fork action] is equivalent to
+      {{!fork_as_promise} [fork_as_promise action |> ignore]}. *)
+end
+
 module Run : sig
   (** Operations for running fibers in specific patterns. *)
 
@@ -300,17 +359,17 @@ end
 
     {[
       let main () =
-        Bundle.join_after begin fun bundle ->
+        Flock.join_after begin fun bundle ->
           let promise =
-            Bundle.fork_as_promise bundle
-            @@ fun () -> Control.block ()
+            Flock.fork_as_promise @@ fun () ->
+            Control.block ()
           in
 
-          Bundle.fork bundle begin fun () ->
+          Flock.fork begin fun () ->
             Promise.await promise
           end;
 
-          Bundle.fork bundle begin fun () ->
+          Flock.fork begin fun () ->
             let condition = Condition.create ()
             and mutex = Mutex.create () in
             Mutex.protect mutex begin fun () ->
@@ -320,27 +379,27 @@ end
             end
           end;
 
-          Bundle.fork bundle begin fun () ->
+          Flock.fork begin fun () ->
             Event.sync (Event.choose [])
           end;
 
-          Bundle.fork bundle begin fun () ->
+          Flock.fork begin fun () ->
             let latch = Latch.create 1 in
             Latch.await latch
           end;
 
-          Bundle.fork bundle begin fun () ->
+          Flock.fork begin fun () ->
             let ivar = Ivar.create () in
             Ivar.read ivar
           end;
 
-          Bundle.fork bundle begin fun () ->
+          Flock.fork begin fun () ->
             let stream = Stream.create () in
             Stream.read (Stream.tap stream)
             |> ignore
           end;
 
-          Bundle.fork bundle begin fun () ->
+          Flock.fork begin fun () ->
             let@ inn, out = finally
               Unix.close_pair @@ fun () ->
               Unix.socketpair ~cloexec:true
@@ -354,7 +413,7 @@ end
             assert (n = 1)
           end;
 
-          Bundle.fork bundle begin fun () ->
+          Flock.fork begin fun () ->
             let a_month =
               60.0 *. 60.0 *. 24.0 *. 30.0
             in
@@ -364,7 +423,7 @@ end
           (* Let the children get stuck *)
           Control.yield ();
 
-          Bundle.terminate bundle
+          Flock.terminate ()
         end
     ]}
 
@@ -373,8 +432,8 @@ end
     library and the {{!Picos_stdio.Unix} [Unix]} module comes from the
     {!Picos_stdio} library.  They do not come from the standard OCaml libraries.
 
-    The above program creates a {{!Bundle} bundle} of fibers and {{!Bundle.fork}
-    forks} several fibers to the bundle that all block in various ways.  In
+    The above program creates a {{!Flock} flock} of fibers and {{!Flock.fork}
+    forks} several fibers to the flock that all block in various ways.  In
     detail,
 
     - {!Control.block} never returns,
@@ -393,9 +452,9 @@ end
       is never written to, and the
     - {!Control.sleep} call would return only after about a month.
 
-    Fibers forked to a bundle can be canceled in various ways.  In the above
-    program we call {!Bundle.terminate} to cancel all of the fibers and
-    effectively close the bundle.  This allows the program to return normally
+    Fibers forked to a flock can be canceled in various ways.  In the above
+    program we call {!Flock.terminate} to cancel all of the fibers and
+    effectively close the flock.  This allows the program to return normally
     immediately and without leaking or leaving anything in an invalid state:
 
     {[
@@ -404,19 +463,19 @@ end
     ]}
 
     Now, the point of the above example isn't that you should just call
-    {{!Bundle.terminate} [terminate]} when your program gets stuck. 😅
+    {{!Flock.terminate} [terminate]} when your program gets stuck. 😅
 
     What the above example hopefully demonstrates is that concurrent
     abstractions like mutexes and condition variables, asynchronous IO
     libraries, and others can be designed to support cancelation.
 
     Cancelation is a control flow mechanism that allows structured concurrent
-    abstractions, like the {!Bundle} abstraction, to (hopefully) gracefully tear
+    abstractions, like the {!Flock} abstraction, to (hopefully) gracefully tear
     down concurrent fibers in case of errors.  Indeed, one of the basic ideas
-    behind the {!Bundle} abstraction is that in case any fiber forked to the
-    bundle raises an unhandled exception, the whole bundle will be terminated
-    and the error will raised from the bundle, which allows you to understand
-    what went wrong, instead of having to debug a program that mysteriously gets
+    behind the {!Flock} abstraction is that in case any fiber forked to the
+    flock raises an unhandled exception, the whole flock will be terminated and
+    the error will raised from the flock, which allows you to understand what
+    went wrong, instead of having to debug a program that mysteriously gets
     stuck, for example.
 
     Cancelation can also, with some care, be used as a mechanism to terminate
@@ -432,12 +491,12 @@ end
 
     {[
       let many_errors () =
-        Bundle.join_after @@ fun bundle ->
+        Flock.join_after @@ fun () ->
 
         let latch = Latch.create 1 in
 
         let fork_raising exn =
-          Bundle.fork bundle begin fun () ->
+          Flock.fork begin fun () ->
             Control.protect begin fun () ->
               Latch.await latch
             end;
@@ -474,7 +533,7 @@ end
       let run_server server_fd =
         Unix.listen server_fd 8;
 
-        Bundle.join_after begin fun bundle ->
+        Flock.join_after begin fun () ->
           while true do
             let@ client_fd =
               instantiate Unix.close @@ fun () ->
@@ -483,7 +542,7 @@ end
             in
 
             (* Fork a fiber for client *)
-            Bundle.fork bundle begin fun () ->
+            Flock.fork begin fun () ->
               let@ client_fd =
                 move client_fd
               in
@@ -558,17 +617,17 @@ end
           Unix.getsockname server_fd
         in
 
-        Bundle.join_after begin fun bundle ->
+        Flock.join_after begin fun () ->
           (* Start server *)
           let server =
-            Bundle.fork_as_promise bundle
-            @@ fun () -> run_server server_fd
+            Flock.fork_as_promise @@ fun () ->
+            run_server server_fd
           in
 
           (* Run clients concurrently *)
-          Bundle.join_after begin fun bundle ->
+          Flock.join_after begin fun () ->
             for _ = 1 to 5 do
-              Bundle.fork bundle @@ fun () ->
+              Flock.fork @@ fun () ->
                 run_client server_addr
             done
           end;
@@ -596,6 +655,6 @@ end
 
     As an exercise, you might want to refactor the server to avoid
     {{!Finally.move} moving} the file descriptors and use a {{!Finally.let@}
-    recursive} accept loop instead.  You could also {{!Bundle.terminate}
-    terminate the whole bundle} at the end instead of just terminating the
+    recursive} accept loop instead.  You could also {{!Flock.terminate}
+    terminate the whole flock} at the end instead of just terminating the
     server. *)
diff --git a/lib/picos_structured/run.ml b/lib/picos_structured/run.ml
index af42ac3..e7e7935 100644
--- a/lib/picos_structured/run.ml
+++ b/lib/picos_structured/run.ml
@@ -20,10 +20,10 @@ let wrap_any t main =
     Bundle.decr t
 
 let run actions wrap =
-  Bundle.join_after @@ fun t ->
+  Bundle.join_after @@ fun (Bundle r as t : Bundle.t) ->
   try
     let mains = List.map (wrap t) actions in
-    Fiber.spawn ~forbid:false t.bundle mains
+    Fiber.spawn ~forbid:false r.bundle mains
   with exn ->
     Bundle.unsafe_reset t;
     raise exn
diff --git a/lib/picos_sync/picos_sync.mli b/lib/picos_sync/picos_sync.mli
index f870763..6d353d7 100644
--- a/lib/picos_sync/picos_sync.mli
+++ b/lib/picos_sync/picos_sync.mli
@@ -502,8 +502,8 @@ end
           Bounded_q.create ~capacity:3
         in
 
-        Bundle.join_after begin fun bundle ->
-          Bundle.fork bundle begin fun () ->
+        Flock.join_after begin fun () ->
+          Flock.fork begin fun () ->
             while true do
               Printf.printf "Popped %d\n%!"
                 (Bounded_q.pop bq)
@@ -519,7 +519,7 @@ end
 
           Control.yield ();
 
-          Bundle.terminate bundle
+          Flock.terminate ()
         end;
 
         Printf.printf "Pushing %d\n%!" 101;


Test patch:
diff --git a/test/test_lwt_unix.ml b/test/test_lwt_unix.ml
index 2111bbf..7782422 100644
--- a/test/test_lwt_unix.ml
+++ b/test/test_lwt_unix.ml
@@ -3,9 +3,9 @@ open Picos_structured
 let basics () =
   Lwt_main.run @@ Picos_lwt_unix.run
   @@ fun () ->
-  Bundle.join_after @@ fun bundle ->
+  Flock.join_after @@ fun () ->
   let child =
-    Bundle.fork_as_promise bundle @@ fun () ->
+    Flock.fork_as_promise @@ fun () ->
     while true do
       Picos_lwt.await (Lwt_unix.sleep 0.01)
     done
diff --git a/test/test_picos_lwt_unix_with_cohttp.ml b/test/test_picos_lwt_unix_with_cohttp.ml
index 6962450..503a458 100644
--- a/test/test_picos_lwt_unix_with_cohttp.ml
+++ b/test/test_picos_lwt_unix_with_cohttp.ml
@@ -7,12 +7,12 @@ open Cohttp_lwt_unix
 let await = Picos_lwt.await
 
 let main () =
-  Bundle.join_after @@ fun bundle ->
+  Flock.join_after @@ fun () ->
   (* We use the [Picos_structured] library for structured concurrency. *)
 
   (* First we start the server. *)
   let server =
-    Bundle.fork_as_promise bundle @@ fun () ->
+    Flock.fork_as_promise @@ fun () ->
     let callback _conn req body =
       let uri = req |> Request.uri |> Uri.to_string in
       let meth = req |> Request.meth |> Code.string_of_method in
diff --git a/test/test_schedulers.ml b/test/test_schedulers.ml
index ec11f7b..f646e91 100644
--- a/test/test_schedulers.ml
+++ b/test/test_schedulers.ml
@@ -36,8 +36,8 @@ let test_current () =
   Test_scheduler.run ~max_domains:2 @@ fun () ->
   let fiber_parent = Fiber.current () in
   let fiber_child = ref fiber_parent in
-  Bundle.join_after @@ fun bundle ->
-  Bundle.fork bundle (fun () -> fiber_child := Fiber.current ());
+  Flock.join_after @@ fun () ->
+  Flock.fork (fun () -> fiber_child := Fiber.current ());
   while fiber_parent == !fiber_child do
     Control.yield ()
   done
diff --git a/test/test_select.ml b/test/test_select.ml
index 7b8846b..c888106 100644
--- a/test/test_select.ml
+++ b/test/test_select.ml
@@ -12,9 +12,9 @@ let test_intr () =
   in
   let main () =
     Picos_threaded.run @@ fun () ->
-    Bundle.join_after @@ fun bundle ->
+    Flock.join_after @@ fun () ->
     for _ = 1 to 10 do
-      Bundle.fork bundle @@ fun () ->
+      Flock.fork @@ fun () ->
       for _ = 1 to 1_000 do
         let req = Picos_select.Intr.req ~seconds:0.000_001 in
         match Unix.read inn (Bytes.create 1) 0 1 with
diff --git a/test/test_server_and_client.ml b/test/test_server_and_client.ml
index 2570636..436d38e 100644
--- a/test/test_server_and_client.ml
+++ b/test/test_server_and_client.ml
@@ -37,7 +37,7 @@ let main () =
         Condition.signal condition;
         Unix.listen socket 8;
         Printf.printf "  Server listening\n%!";
-        Bundle.join_after @@ fun bundle ->
+        Flock.join_after @@ fun () ->
         while true do
           let@ client =
             instantiate Unix.close @@ fun () ->
@@ -46,7 +46,7 @@ let main () =
           in
           Printf.printf "  Server accepted client\n%!";
 
-          Bundle.fork bundle @@ fun () ->
+          Flock.fork @@ fun () ->
           let@ client = move client in
           set_nonblock client;
           let bytes = Bytes.create n in
@@ -71,7 +71,7 @@ let main () =
         Condition.signal condition;
         Unix.listen socket 8;
         Printf.printf "  Server listening\n%!";
-        Bundle.join_after @@ fun bundle ->
+        Flock.join_after @@ fun () ->
         let rec accept () =
           let@ client =
             finally Unix.close @@ fun () ->
@@ -79,7 +79,7 @@ let main () =
             Unix.accept ~cloexec:true socket |> fst
           in
           Printf.printf "  Server accepted client\n%!";
-          Bundle.fork bundle accept;
+          Flock.fork accept;
           set_nonblock client;
           let bytes = Bytes.create n in
           let n = Unix.read client bytes 0 (Bytes.length bytes) in
@@ -87,7 +87,7 @@ let main () =
           let n = Unix.write client bytes 0 (n / 2) in
           Printf.printf "  Server wrote %d\n%!" n
         in
-        Bundle.fork bundle accept
+        Flock.fork accept
     | exception Unix.Unix_error (EPERM, _, _) when is_opam_ci -> raise Exit
   in
 
@@ -110,8 +110,8 @@ let main () =
   in
 
   begin
-    Bundle.join_after @@ fun bundle ->
-    Bundle.fork bundle server;
+    Flock.join_after @@ fun () ->
+    Flock.fork server;
     begin
       Mutex.protect mutex @@ fun () ->
       while !server_addr == loopback_0 do
@@ -119,7 +119,7 @@ let main () =
       done
     end;
     Run.all [ client "A"; client "B" ];
-    Bundle.terminate bundle
+    Flock.terminate ()
   end;
 
   Printf.printf "Server and Client test: OK\n%!"
diff --git a/test/test_stdio.ml b/test/test_stdio.ml
index e9840bc..6356d7d 100644
--- a/test/test_stdio.ml
+++ b/test/test_stdio.ml
@@ -46,14 +46,14 @@ let test_select_empty_timeout () =
 
 let test_select_empty_forever () =
   Test_scheduler.run ~max_domains:2 @@ fun () ->
-  Bundle.join_after @@ fun bundle ->
+  Flock.join_after @@ fun () ->
   begin
-    Bundle.fork bundle @@ fun () ->
+    Flock.fork @@ fun () ->
     let _ = Unix.select [] [] [] (-1.0) in
     Printf.printf "Impossible\n%!"
   end;
   Unix.sleepf 0.01;
-  Bundle.terminate bundle
+  Flock.terminate ()
 
 let test_select () =
   Test_scheduler.run ~max_domains:2 @@ fun () ->
@@ -79,9 +79,9 @@ let test_select () =
 
   let events = Picos_mpscq.create ~padded:true () in
 
-  Bundle.join_after @@ fun bundle ->
+  Flock.join_after @@ fun () ->
   begin
-    Bundle.fork bundle @@ fun () ->
+    Flock.fork @@ fun () ->
     while true do
       match Unix.select [ msg_inn1; msg_inn2 ] [] [] 0.2 with
       | inns, _, _ ->
@@ -111,7 +111,7 @@ let test_select () =
   assert (1 = Unix.read syn_inn (Bytes.create 1) 0 1);
   assert (1 = Unix.read syn_inn (Bytes.create 1) 0 1);
 
-  Bundle.terminate bundle;
+  Flock.terminate ();
 
   let expected = [| "Inn1"; "Inn2"; "Timeout" |] in
   let actual =
diff --git a/test/test_structured.ml b/test/test_structured.ml
index a3871e7..8033804 100644
--- a/test/test_structured.ml
+++ b/test/test_structured.ml
@@ -19,9 +19,9 @@ let check join_after scope =
 
 let test_fork_after_terminate () =
   Test_scheduler.run @@ fun () ->
-  check Bundle.join_after @@ fun bundle ->
-  Bundle.terminate bundle;
-  match Bundle.fork bundle (fun () -> Printf.printf "Hello!\n%!") with
+  check Flock.join_after @@ fun () ->
+  Flock.terminate ();
+  match Flock.fork (fun () -> Printf.printf "Hello!\n%!") with
   | () -> assert false
   | exception Control.Terminate -> ()
 
@@ -41,9 +41,9 @@ let test_exception_in_child_terminates () =
     let mutex = Mutex.create () in
     let condition = Condition.create () in
     let blocked = ref false in
-    check Bundle.join_after @@ fun bundle ->
+    check Flock.join_after @@ fun () ->
     begin
-      Bundle.fork bundle @@ fun () ->
+      Flock.fork @@ fun () ->
       begin
         Mutex.protect mutex @@ fun () ->
         while not !blocked do
@@ -69,13 +69,13 @@ let test_cancelation_awaits_children () =
   let blocked = ref false in
   let slept = ref false in
   begin
-    check Bundle.join_after @@ fun bundle ->
+    check Flock.join_after @@ fun () ->
     begin
-      Bundle.fork bundle @@ fun () ->
+      Flock.fork @@ fun () ->
       begin
-        check Bundle.join_after @@ fun bundle ->
+        check Flock.join_after @@ fun () ->
         begin
-          Bundle.fork bundle @@ fun () ->
+          Flock.fork @@ fun () ->
           try
             blocked := true;
             Control.block ()
@@ -92,7 +92,7 @@ let test_cancelation_awaits_children () =
     while not !blocked do
       Control.sleep ~seconds:0.01
     done;
-    Bundle.terminate bundle
+    Flock.terminate ()
   end;
   assert !slept
 
@@ -126,13 +126,13 @@ let test_termination_nests () =
   let condition = Condition.create () in
   let blocked = ref false in
   begin
-    check Bundle.join_after @@ fun bundle ->
+    check Flock.join_after @@ fun () ->
     begin
-      Bundle.fork bundle @@ fun () ->
+      Flock.fork @@ fun () ->
       begin
-        check Bundle.join_after @@ fun bundle ->
+        check Flock.join_after @@ fun () ->
         begin
-          Bundle.fork bundle @@ fun () ->
+          Flock.fork @@ fun () ->
           begin
             Mutex.protect mutex @@ fun () -> blocked := true
           end;
@@ -151,13 +151,13 @@ let test_termination_nests () =
       done
     end;
 
-    Bundle.terminate bundle
+    Flock.terminate ()
   end
 
 let test_promise_cancelation_does_not_terminate () =
   Test_scheduler.run ~max_domains:2 @@ fun () ->
-  Bundle.join_after @@ fun bundle ->
-  let promise = Bundle.fork_as_promise bundle @@ fun () -> Control.block () in
+  Flock.join_after @@ fun () ->
+  let promise = Flock.fork_as_promise @@ fun () -> Control.block () in
   Control.yield ();
   Promise.terminate promise;
   Control.yield ()
@@ -165,10 +165,8 @@ let test_promise_cancelation_does_not_terminate () =
 let test_error_in_promise_terminates () =
   match
     Test_scheduler.run ~max_domains:2 @@ fun () ->
-    Bundle.join_after @@ fun bundle ->
-    let promise =
-      Bundle.fork_as_promise bundle @@ fun () -> failwith "I failed"
-    in
+    Flock.join_after @@ fun () ->
+    let promise = Flock.fork_as_promise @@ fun () -> failwith "I failed" in
     Control.block () |> ignore;
     Promise.terminate promise
   with
@@ -177,9 +175,9 @@ let test_error_in_promise_terminates () =
 
 let test_can_wait_promises () =
   Test_scheduler.run ~max_domains:2 @@ fun () ->
-  Bundle.join_after @@ fun bundle ->
+  Flock.join_after @@ fun () ->
   let promise =
-    Bundle.fork_as_promise bundle @@ fun () ->
+    Flock.fork_as_promise @@ fun () ->
     Control.sleep ~seconds:0.1;
     42
   in
@@ -187,14 +185,14 @@ let test_can_wait_promises () =
 
 let test_can_select_promises () =
   Test_scheduler.run ~max_domains:2 @@ fun () ->
-  Bundle.join_after @@ fun bundle ->
+  Flock.join_after @@ fun () ->
   let a =
-    Bundle.fork_as_promise bundle @@ fun () ->
+    Flock.fork_as_promise @@ fun () ->
     Control.sleep ~seconds:0.1;
     42
-  and b = Bundle.fork_as_promise bundle @@ fun () -> Event.select [] in
+  and b = Flock.fork_as_promise @@ fun () -> Event.select [] in
   assert (Event.select [ Promise.completed a; Promise.completed b ] = 42);
-  Bundle.terminate bundle
+  Flock.terminate ()
 
 let test_any_and_all_errors () =
   [ Run.all; Run.any ]
diff --git a/test/test_sync.ml b/test/test_sync.ml
index 24f91db..7d799e0 100644
--- a/test/test_sync.ml
+++ b/test/test_sync.ml
@@ -48,15 +48,15 @@ let test_mutex_and_condition_errors () =
   let mutex = Mutex.create () in
   let condition = Condition.create () in
   Mutex.protect mutex @@ fun () ->
-  Bundle.join_after @@ fun bundle ->
+  Flock.join_after @@ fun () ->
   begin
-    Bundle.fork bundle @@ fun () ->
+    Flock.fork @@ fun () ->
     match Mutex.unlock mutex with
     | () -> assert false
     | exception Sys_error _ -> ()
   end;
   begin
-    Bundle.fork bundle @@ fun () ->
+    Flock.fork @@ fun () ->
     match Condition.wait condition mutex with
     | () -> assert false
     | exception Sys_error _ -> ()