Repository: ocaml-multicore/picos
Pull request #54 (2024-03-13 21:23:40)

Minimalistic `Unix.select` event loop for Picos
This PR introduces
* `Picos_select` &mdash; a basic `Unix.select` based IO event loop for Picos,
* `Picos.Implementation` &mdash; a signature for primitive effect based operations and a mechanism that allows setting the implementation of Picos on OCaml 4, which, like on OCaml 5, no longer has a default implementation, and
* `Picos_threaded` &mdash; a basic `Thread` based implementation for OCaml 4, which uses `Picos_select`.


Patch:
diff --git a/README.md b/README.md
index 20cf895..ec2720a 100644
--- a/README.md
+++ b/README.md
@@ -1,4 +1,4 @@
-[API reference](https://ocaml-multicore.github.io/picos/doc/picos/Picos/index.html)
+[API reference](https://ocaml-multicore.github.io/picos/doc/picos/index.html)
 &middot; [Benchmarks](https://bench.ci.dev/ocaml-multicore/picos/branch/main)
 
 # **Picos** &mdash; Interoperable effects based concurrency
diff --git a/bench/dune b/bench/dune
index 8223358..6fe82a4 100644
--- a/bench/dune
+++ b/bench/dune
@@ -17,7 +17,7 @@ let () =
   foundation
   (select scheduler.ml from
    (schedulers -> scheduler.ocaml5.ml)
-   (           -> scheduler.ocaml4.ml))
+   (picos.threaded -> scheduler.ocaml4.ml))
   multicore-bench
   multicore-magic |}
   ^ maybe_domain_shims ^ {| ))
diff --git a/bench/scheduler.ocaml4.ml b/bench/scheduler.ocaml4.ml
index b837beb..1c54aaf 100644
--- a/bench/scheduler.ocaml4.ml
+++ b/bench/scheduler.ocaml4.ml
@@ -1,1 +1,2 @@
+let () = Picos.set_picos_implementation Picos_threaded.implementation
 let run action = action ()
diff --git a/lib/picos/bootstrap/dune b/lib/picos/bootstrap/dune
index cf47882..b525fc3 100644
--- a/lib/picos/bootstrap/dune
+++ b/lib/picos/bootstrap/dune
@@ -1,4 +1,6 @@
 (library
- (name picos_private_bootstrap)
+ (name picos_bootstrap)
  (package picos)
- (libraries picos_exn_bt backoff))
+ (libraries
+  (re_export picos.exn_bt)
+  backoff))
diff --git a/lib/picos/bootstrap/picos_private_bootstrap.ml b/lib/picos/bootstrap/picos_bootstrap.ml
similarity index 89%
rename from lib/picos/bootstrap/picos_private_bootstrap.ml
rename to lib/picos/bootstrap/picos_bootstrap.ml
index 3abbd79..7edbd85 100644
--- a/lib/picos/bootstrap/picos_private_bootstrap.ml
+++ b/lib/picos/bootstrap/picos_bootstrap.ml
@@ -1,3 +1,5 @@
+module Exn_bt = Picos_exn_bt
+
 module Trigger = struct
   let[@inline never] awaiting () = invalid_arg "Trigger: already awaiting"
 
@@ -10,7 +12,12 @@ module Trigger = struct
 
   let create () = Atomic.make Initial
   let is_signaled t = Atomic.get t == Signaled
-  let is_initial t = Atomic.get t == Initial
+
+  let is_initial t =
+    match Atomic.get t with
+    | Initial -> true
+    | Awaiting _ -> awaiting ()
+    | Signaled -> false
 
   let rec finish t ~allow_awaiting =
     match Atomic.get t with
@@ -39,10 +46,11 @@ module Trigger = struct
 end
 
 module Computation = struct
-  let[@inline never] negative () = invalid_arg "Computation: negative seconds"
+  let[@inline never] negative_or_nan () =
+    invalid_arg "Computation: seconds must be non-negative"
 
   type 'a state =
-    | Canceled of Picos_exn_bt.t
+    | Canceled of Exn_bt.t
     | Returned of 'a
     | Continue of { balance_and_mode : int; triggers : Trigger.t list }
 
@@ -148,13 +156,13 @@ module Computation = struct
   let try_capture t fn x =
     match fn x with
     | y -> try_return t y
-    | exception exn -> try_cancel t (Picos_exn_bt.get exn)
+    | exception exn -> try_cancel t (Exn_bt.get exn)
 
   let capture t fn x = try_capture t fn x |> ignore
 
   let check t =
     match Atomic.get t with
-    | Canceled exn_bt -> Picos_exn_bt.raise exn_bt
+    | Canceled exn_bt -> Exn_bt.raise exn_bt
     | Returned _ | Continue _ -> ()
 
   let peek t =
@@ -172,6 +180,15 @@ module Computation = struct
   end
 
   let canceler ~from ~into = Trigger.from_action from into propagate
+
+  let check_non_negative seconds =
+    if not (0.0 <= seconds) then negative_or_nan ()
+
+  let rec get_or block t =
+    match Atomic.get t with
+    | Returned value -> value
+    | Canceled exn_bt -> Exn_bt.raise exn_bt
+    | Continue _ -> get_or block (block t)
 end
 
 module Fiber = struct
@@ -290,3 +307,19 @@ module Fiber = struct
           (Sys.opaque_identity (Obj.magic value : non_float))
   end
 end
+
+module type Implementation = sig
+  module Fiber : sig
+    val current : unit -> Fiber.t
+    val spawn : forbid:bool -> 'a Computation.t -> (unit -> unit) list -> unit
+    val yield : unit -> unit
+  end
+
+  module Computation : sig
+    val cancel_after : 'a Computation.t -> seconds:float -> Exn_bt.t -> unit
+  end
+
+  module Trigger : sig
+    val await : Trigger.t -> Exn_bt.t option
+  end
+end
diff --git a/lib/picos/default/domain.ocaml4.ml b/lib/picos/default/domain.ocaml4.ml
deleted file mode 100644
index a2facd6..0000000
--- a/lib/picos/default/domain.ocaml4.ml
+++ /dev/null
@@ -1,1 +0,0 @@
-let at_exit = Stdlib.at_exit
diff --git a/lib/picos/default/picos_default.ml b/lib/picos/default/picos_default.ml
deleted file mode 100644
index d3117f9..0000000
--- a/lib/picos/default/picos_default.ml
+++ /dev/null
@@ -1,330 +0,0 @@
-module type Default = sig
-  module Trigger : sig
-    type t
-
-    val await : t -> Picos_exn_bt.t option
-  end
-
-  module Computation : sig
-    type !'a t
-
-    val cancel_after : 'a t -> seconds:float -> Picos_exn_bt.t -> unit
-  end
-
-  module Fiber : sig
-    type t
-
-    val current : unit -> t
-    val spawn : forbid:bool -> 'a Computation.t -> (unit -> unit) list -> unit
-    val yield : unit -> unit
-  end
-end
-
-module type Bootstrap = sig
-  module Trigger : sig
-    type t
-
-    val is_signaled : t -> bool
-    val on_signal : t -> 'x -> 'y -> (t -> 'x -> 'y -> unit) -> bool
-    val from_action : 'x -> 'y -> (t -> 'x -> 'y -> unit) -> t
-    val signal : t -> unit
-    val dispose : t -> unit
-  end
-
-  module Computation : sig
-    type !'a t
-
-    val create : ?mode:[ `FIFO | `LIFO ] -> unit -> 'a t
-    val cancel : 'a t -> Picos_exn_bt.t -> unit
-    val try_attach : 'a t -> Trigger.t -> bool
-  end
-
-  module Fiber : sig
-    type t
-
-    val create : forbid:bool -> 'a Computation.t -> t
-    val has_forbidden : t -> bool
-    val canceled : t -> Picos_exn_bt.t option
-    val try_attach : t -> Trigger.t -> bool
-    val detach : t -> Trigger.t -> unit
-    val check : t -> unit
-  end
-end
-
-module Make (Bootstrap : Bootstrap) :
-  Default
-    with type Trigger.t = Bootstrap.Trigger.t
-    with type 'a Computation.t = 'a Bootstrap.Computation.t
-    with type Fiber.t = Bootstrap.Fiber.t = struct
-  module Per_thread = struct
-    type t = {
-      mutex : Mutex.t;
-      condition : Condition.t;
-      fiber : Bootstrap.Fiber.t;
-    }
-
-    open struct
-      let create fiber =
-        let mutex = Mutex.create () and condition = Condition.create () in
-        { mutex; condition; fiber }
-
-      let per_thread_key =
-        Picos_tls.new_key @@ fun () ->
-        create
-          (Bootstrap.Fiber.create ~forbid:false
-             (Bootstrap.Computation.create ()))
-    end
-
-    let get () = Picos_tls.get per_thread_key
-    let set fiber = Picos_tls.set per_thread_key (create fiber)
-  end
-
-  module Trigger = struct
-    type t = Bootstrap.Trigger.t
-
-    let release _ _ (per_thread : Per_thread.t) =
-      Mutex.lock per_thread.mutex;
-      Mutex.unlock per_thread.mutex;
-      Condition.broadcast per_thread.condition
-
-    let block trigger (per_thread : Per_thread.t) =
-      Mutex.lock per_thread.mutex;
-      match
-        while not (Bootstrap.Trigger.is_signaled trigger) do
-          Condition.wait per_thread.condition per_thread.mutex
-        done
-      with
-      | () ->
-          Mutex.unlock per_thread.mutex;
-          Bootstrap.Fiber.canceled per_thread.fiber
-      | exception exn ->
-          (* Condition.wait may be interrupted by asynchronous exceptions and we
-             must make sure to unlock even in that case. *)
-          Mutex.unlock per_thread.mutex;
-          raise exn
-
-    let await trigger =
-      let per_thread = Per_thread.get () in
-      if Bootstrap.Fiber.has_forbidden per_thread.fiber then
-        if Bootstrap.Trigger.on_signal trigger () per_thread release then
-          block trigger per_thread
-        else None
-      else if Bootstrap.Fiber.try_attach per_thread.fiber trigger then
-        if Bootstrap.Trigger.on_signal trigger () per_thread release then
-          block trigger per_thread
-        else begin
-          Bootstrap.Fiber.detach per_thread.fiber trigger;
-          Bootstrap.Fiber.canceled per_thread.fiber
-        end
-      else begin
-        Bootstrap.Trigger.dispose trigger;
-        Bootstrap.Fiber.canceled per_thread.fiber
-      end
-  end
-
-  module Computation = struct
-    type 'a t = 'a Bootstrap.Computation.t
-
-    module Entry = struct
-      type t =
-        | Entry : {
-            time : Mtime.span;
-            exn_bt : Picos_exn_bt.t;
-            computation : 'a Bootstrap.Computation.t;
-          }
-            -> t
-
-      let compare (Entry l) (Entry r) = Mtime.Span.compare l.time r.time
-    end
-
-    module Q = Psq.Make (Int) (Entry)
-
-    module Per_domain = struct
-      type t = {
-        timeouts : Q.t Atomic.t;
-            (** Must be atomic to allow removal from another domain. *)
-        mutable next_id : int;
-        mutable needs_wakeup : bool;
-        mutable state : [ `Init | `Locked | `Alive | `Dead ];
-        mutable exn_bt : Picos_exn_bt.t;
-        mutable pipe_inn : Select.file_descr list;
-        mutable pipe_out : Select.file_descr;
-        byte : Bytes.t;
-      }
-
-      let key =
-        Picos_domain.DLS.new_key @@ fun () ->
-        {
-          timeouts = Atomic.make Q.empty;
-          next_id = 0;
-          needs_wakeup = true;
-          state = `Init;
-          exn_bt = Picos_exn_bt.get_callstack 0 Exit;
-          pipe_inn = [];
-          pipe_out = Select.stdin;
-          byte = Bytes.create 1;
-        }
-
-      let get () = Picos_domain.DLS.get key
-
-      let[@poll error] [@inline never] try_lock t =
-        t.state == `Init
-        && begin
-             t.state <- `Locked;
-             true
-           end
-
-      let[@poll error] [@inline never] unlock t state = t.state <- state
-
-      let[@poll error] [@inline never] next_id t =
-        let id = t.next_id in
-        t.next_id <- id + 1;
-        id
-
-      let[@poll error] [@inline never] wakeup_needed t =
-        t.needs_wakeup
-        && begin
-             t.needs_wakeup <- false;
-             true
-           end
-
-      let wakeup t =
-        let pipe_out = t.pipe_out in
-        if wakeup_needed t then
-          let n = Select.write pipe_out t.byte 0 1 in
-          assert (n = 1)
-
-      let rec add_timeout t id entry =
-        let before = Atomic.get t.timeouts in
-        let after = Q.add id entry before in
-        if Atomic.compare_and_set t.timeouts before after then
-          match Q.min after with
-          | Some (id', _) -> if id = id' then wakeup t
-          | None -> ()
-        else add_timeout t id entry
-
-      let rec remove_timeout _trigger t id =
-        let before = Atomic.get t.timeouts in
-        let after = Q.remove id before in
-        if not (Atomic.compare_and_set t.timeouts before after) then
-          remove_timeout (Obj.magic ()) t id
-
-      let wait t =
-        while t.state == `Locked do
-          Systhreads.yield ()
-        done;
-        if t.state != `Alive then
-          invalid_arg "Computation: domain has been terminated"
-
-      let[@poll error] [@inline never] running_atomically t before =
-        (* New timeouts may only be added from the same domain, which means that
-           [@poll error] is sufficient to ensure that the timeout before next
-           remains valid. *)
-        t.state == `Alive
-        && Atomic.get t.timeouts == before
-        && begin
-             t.needs_wakeup <- true;
-             true
-           end
-
-      let rec timeout_thread_sleep t before next =
-        if running_atomically t before then begin
-          match Select.select t.pipe_inn [] [] next with
-          | [ pipe_inn ], _, _ ->
-              let n = Select.read pipe_inn t.byte 0 1 in
-              assert (n = 1)
-          | _, _, _ -> ()
-        end;
-        timeout_thread_handle t
-
-      and timeout_thread_handle t =
-        if t.state == `Alive then
-          let before = Atomic.get t.timeouts in
-          match Q.pop before with
-          | None -> timeout_thread_sleep t before (-1.0)
-          | Some ((_, Entry e), after) ->
-              let elapsed = Mtime_clock.elapsed () in
-              if Mtime.Span.compare e.time elapsed <= 0 then begin
-                if Atomic.compare_and_set t.timeouts before after then
-                  Bootstrap.Computation.cancel e.computation e.exn_bt;
-                timeout_thread_handle t
-              end
-              else
-                let next =
-                  Mtime.Span.to_float_ns (Mtime.Span.abs_diff e.time elapsed)
-                  *. (1. /. 1_000_000_000.)
-                in
-                timeout_thread_sleep t before next
-
-      let timeout_thread t =
-        begin
-          try
-            let pipe_inn, pipe_out = Select.pipe () in
-            t.pipe_inn <- [ pipe_inn ];
-            t.pipe_out <- pipe_out;
-            unlock t `Alive;
-            timeout_thread_handle t
-          with exn -> t.exn_bt <- Picos_exn_bt.get exn
-        end;
-        t.needs_wakeup <- false;
-        unlock t `Dead;
-        List.iter Select.close t.pipe_inn;
-        if t.pipe_out != Select.stdin then Select.close t.pipe_out
-
-      let start t =
-        match Thread.create timeout_thread t with
-        | thread ->
-            Picos_domain.at_exit @@ fun () ->
-            unlock t `Dead;
-            wakeup t;
-            Systhreads.join thread;
-            if t.exn_bt.exn != Exit then Picos_exn_bt.raise t.exn_bt
-        | exception exn ->
-            unlock t `Dead;
-            raise exn
-
-      let init t =
-        if try_lock t then start t;
-        wait t
-    end
-
-    let cancel_after computation ~seconds exn_bt =
-      match Mtime.Span.of_float_ns (seconds *. 1_000_000_000.) with
-      | None ->
-          invalid_arg
-            "Computation: seconds should be between 0 to pow(2, 53) nanoseconds"
-      | Some span ->
-          let per_domain = Per_domain.get () in
-          if per_domain.state != `Alive then Per_domain.init per_domain;
-          let time = Mtime.Span.add (Mtime_clock.elapsed ()) span in
-          let entry = Entry.Entry { time; exn_bt; computation } in
-          let id = Per_domain.next_id per_domain in
-          Per_domain.add_timeout per_domain id entry;
-          let remover =
-            Bootstrap.Trigger.from_action per_domain id
-              Per_domain.remove_timeout
-          in
-          if not (Bootstrap.Computation.try_attach computation remover) then
-            Bootstrap.Trigger.signal remover
-  end
-
-  module Fiber = struct
-    type t = Bootstrap.Fiber.t
-
-    let current () =
-      let fiber = (Per_thread.get ()).fiber in
-      Bootstrap.Fiber.check fiber;
-      fiber
-
-    let spawn ~forbid computation mains =
-      mains
-      |> List.iter @@ fun main ->
-         Systhreads.create
-           (fun () ->
-             main (Per_thread.set (Bootstrap.Fiber.create ~forbid computation)))
-           ()
-         |> ignore
-
-    let yield () = Systhreads.yield ()
-  end
-end
diff --git a/lib/picos/default/select.none.ml b/lib/picos/default/select.none.ml
deleted file mode 100644
index 45df6d7..0000000
--- a/lib/picos/default/select.none.ml
+++ /dev/null
@@ -1,8 +0,0 @@
-type file_descr
-
-let stdin : file_descr = Obj.magic ()
-let close _ = raise (Sys_error "Computation: cancel_after unavailable")
-let read _ _ _ _ = close ()
-let write _ _ _ _ = close ()
-let pipe ?cloexec:_ _ = close ()
-let select _ _ _ _ = close ()
diff --git a/lib/picos/default/select.unix.ml b/lib/picos/default/select.unix.ml
deleted file mode 100644
index 5dbba15..0000000
--- a/lib/picos/default/select.unix.ml
+++ /dev/null
@@ -1,8 +0,0 @@
-type file_descr = Unix.file_descr
-
-let stdin = Unix.stdin
-let close = Unix.close
-let read = Unix.read
-let write = Unix.write
-let pipe = Unix.pipe
-let select = Unix.select
diff --git a/lib/picos/domain/picos_domain.mli b/lib/picos/domain/picos_domain.mli
index def432b..36c9b67 100644
--- a/lib/picos/domain/picos_domain.mli
+++ b/lib/picos/domain/picos_domain.mli
@@ -1,3 +1,6 @@
+(** A minimalistic domain API available both on OCaml 5 and on OCaml 4 for
+    {!Picos}. *)
+
 val at_exit : (unit -> unit) -> unit
 (** [at_exit action] registers [action] to be called when the current domain
     exits.
diff --git a/lib/picos/dune b/lib/picos/dune
index 998dcca..3e9f55e 100644
--- a/lib/picos/dune
+++ b/lib/picos/dune
@@ -2,22 +2,22 @@
  (name picos)
  (public_name picos)
  (libraries
-  (re_export picos_exn_bt)
-  picos_private_bootstrap
+  (re_export picos.exn_bt)
+  picos_bootstrap
   (select
    picos.ml
    from
-   (picos_private_ocaml4 -> picos.common.ml)
-   (picos_private_ocaml5 -> picos.common.ml))
+   (picos_ocaml4 -> picos.common.ml)
+   (picos_ocaml5 -> picos.common.ml))
   (select
    intf.ml
    from
-   (picos_private_ocaml4 -> intf.ocaml4.ml)
-   (picos_private_ocaml5 -> intf.ocaml5.ml))))
+   (picos_ocaml4 -> intf.ocaml4.ml)
+   (picos_ocaml5 -> intf.ocaml5.ml))))
 
 (mdx
  (package picos)
  (deps
   (package picos))
- (libraries picos picos_exn_bt domain_shims)
+ (libraries picos picos.exn_bt picos.threaded domain_shims)
  (files picos.mli))
diff --git a/lib/picos/exn_bt/picos_exn_bt.ocaml4.mli b/lib/picos/exn_bt/picos_exn_bt.ocaml4.mli
index 6750cad..bf8b9a6 100644
--- a/lib/picos/exn_bt/picos_exn_bt.ocaml4.mli
+++ b/lib/picos/exn_bt/picos_exn_bt.ocaml4.mli
@@ -1,1 +1,3 @@
+(** A wrapper for exceptions with backtraces for {!Picos}. *)
+
 include Common.Intf
diff --git a/lib/picos/exn_bt/picos_exn_bt.ocaml5.mli b/lib/picos/exn_bt/picos_exn_bt.ocaml5.mli
index 2ca0bf1..bc8bddd 100644
--- a/lib/picos/exn_bt/picos_exn_bt.ocaml5.mli
+++ b/lib/picos/exn_bt/picos_exn_bt.ocaml5.mli
@@ -1,3 +1,5 @@
+(** A wrapper for exceptions with backtraces for {!Picos}. *)
+
 include Common.Intf
 
 val discontinue : ('a, 'b) Effect.Deep.continuation -> t -> 'b
diff --git a/lib/picos/ocaml4/dune b/lib/picos/ocaml4/dune
index 7b46872..adc2bc8 100644
--- a/lib/picos/ocaml4/dune
+++ b/lib/picos/ocaml4/dune
@@ -1,7 +1,9 @@
 (library
- (name picos_private_ocaml4)
+ (name picos_ocaml4)
  (package picos)
  (wrapped false)
  (enabled_if
   (< %{ocaml_version} 5.0.0))
- (libraries picos_exn_bt picos_private_bootstrap picos_default))
+ (libraries
+  (re_export picos.exn_bt)
+  picos_bootstrap))
diff --git a/lib/picos/ocaml4/picos_ocaml.ml b/lib/picos/ocaml4/picos_ocaml.ml
new file mode 100644
index 0000000..4cee3fe
--- /dev/null
+++ b/lib/picos/ocaml4/picos_ocaml.ml
@@ -0,0 +1,62 @@
+open Picos_bootstrap
+
+let error () =
+  raise (Sys_error "Picos.set_picos_implementation must be called exactly once")
+
+type implementation = {
+  current : unit -> Fiber.t;
+  spawn : 'a. forbid:bool -> 'a Computation.t -> (unit -> unit) list -> unit;
+  yield : unit -> unit;
+  cancel_after : 'a. 'a Computation.t -> seconds:float -> Exn_bt.t -> unit;
+  await : Trigger.t -> Exn_bt.t option;
+}
+
+let implementation =
+  let current = error
+  and spawn ~forbid:_ _ _ = error ()
+  and yield = error
+  and cancel_after _ ~seconds:_ _ = error ()
+  and await _ = error () in
+  ref { current; spawn; yield; cancel_after; await }
+
+module Trigger = struct
+  let await t = if Trigger.is_initial t then !implementation.await t else None
+end
+
+module Fiber = struct
+  let current () = !implementation.current ()
+
+  let spawn ~forbid computation mains =
+    let _ = current () in
+    !implementation.spawn ~forbid computation mains
+
+  let yield () =
+    let _ = current () in
+    !implementation.yield ()
+end
+
+module Computation = struct
+  let cancel_after computation ~seconds exn_bt =
+    Computation.check_non_negative seconds;
+    let _ = Fiber.current () in
+    !implementation.cancel_after computation ~seconds exn_bt
+end
+
+let[@poll error] [@inline never] set_picos_implementation i =
+  !implementation.current == error
+  && begin
+       implementation := i;
+       true
+     end
+
+let set_picos_implementation (module I : Implementation) =
+  let i =
+    {
+      current = I.Fiber.current;
+      spawn = I.Fiber.spawn;
+      yield = I.Fiber.yield;
+      cancel_after = I.Computation.cancel_after;
+      await = I.Trigger.await;
+    }
+  in
+  if not (set_picos_implementation i) then error ()
diff --git a/lib/picos/ocaml4/picos_private.ml b/lib/picos/ocaml4/picos_private.ml
deleted file mode 100644
index 36f2ae8..0000000
--- a/lib/picos/ocaml4/picos_private.ml
+++ /dev/null
@@ -1,36 +0,0 @@
-module Exn_bt = Picos_exn_bt
-module Default = Picos_default.Make (Picos_private_bootstrap)
-
-module Trigger = struct
-  include Picos_private_bootstrap.Trigger
-
-  let await t =
-    match Atomic.get t with
-    | Initial -> Default.Trigger.await t
-    | Signaled -> None
-    | Awaiting _ -> awaiting ()
-end
-
-module Fiber = struct
-  include Picos_private_bootstrap.Fiber
-
-  let current = Default.Fiber.current
-
-  let spawn ~forbid computation mains =
-    let _ = current () in
-    Default.Fiber.spawn ~forbid computation mains
-
-  let yield () =
-    let _ = current () in
-    Default.Fiber.yield ()
-end
-
-module Computation = struct
-  include Picos_private_bootstrap.Computation
-
-  let cancel_after computation ~seconds exn_bt =
-    if 0.0 <= seconds then
-      let _ = Fiber.current () in
-      Default.Computation.cancel_after computation ~seconds exn_bt
-    else negative ()
-end
diff --git a/lib/picos/ocaml5/dune b/lib/picos/ocaml5/dune
index 99675b2..3e21312 100644
--- a/lib/picos/ocaml5/dune
+++ b/lib/picos/ocaml5/dune
@@ -1,7 +1,9 @@
 (library
- (name picos_private_ocaml5)
+ (name picos_ocaml5)
  (package picos)
  (wrapped false)
  (enabled_if
   (>= %{ocaml_version} 5.0.0))
- (libraries picos_exn_bt picos_private_bootstrap))
+ (libraries
+  (re_export picos.exn_bt)
+  picos_bootstrap))
diff --git a/lib/picos/ocaml5/intf.ml b/lib/picos/ocaml5/intf.ml
deleted file mode 100644
index e69de29..0000000
diff --git a/lib/picos/ocaml5/picos_ocaml.ml b/lib/picos/ocaml5/picos_ocaml.ml
new file mode 100644
index 0000000..36e2d37
--- /dev/null
+++ b/lib/picos/ocaml5/picos_ocaml.ml
@@ -0,0 +1,56 @@
+open Picos_bootstrap
+
+module Trigger = struct
+  type _ Effect.t += Await : Trigger.t -> Exn_bt.t option Effect.t
+
+  let await t = if Trigger.is_initial t then Effect.perform (Await t) else None
+end
+
+module Fiber = struct
+  let continue t k v =
+    match Fiber.canceled t with
+    | None -> Effect.Deep.continue k v
+    | Some exn_bt -> Exn_bt.discontinue k exn_bt
+
+  let continue_with t k v h =
+    match Fiber.canceled t with
+    | None -> Effect.Shallow.continue_with k v h
+    | Some exn_bt -> Exn_bt.discontinue_with k exn_bt h
+
+  type _ Effect.t += Current : Fiber.t Effect.t
+
+  let current () = Effect.perform Current
+
+  type _ Effect.t +=
+    | Spawn : {
+        forbid : bool;
+        computation : 'a Computation.t;
+        mains : (unit -> unit) list;
+      }
+        -> unit Effect.t
+
+  let spawn ~forbid computation mains =
+    Effect.perform @@ Spawn { forbid; computation; mains }
+
+  type _ Effect.t += Yield : unit Effect.t
+
+  let yield () = Effect.perform Yield
+end
+
+module Computation = struct
+  type _ Effect.t +=
+    | Cancel_after : {
+        seconds : float;
+        exn_bt : Exn_bt.t;
+        computation : 'a Computation.t;
+      }
+        -> unit Effect.t
+
+  let cancel_after computation ~seconds exn_bt =
+    Computation.check_non_negative seconds;
+    Effect.perform (Cancel_after { seconds; exn_bt; computation })
+end
+
+let set_picos_implementation _ =
+  raise
+    (Sys_error "Picos.set_picos_implementation must be called only on OCaml 4")
diff --git a/lib/picos/ocaml5/picos_private.ml b/lib/picos/ocaml5/picos_private.ml
deleted file mode 100644
index bd07fd0..0000000
--- a/lib/picos/ocaml5/picos_private.ml
+++ /dev/null
@@ -1,67 +0,0 @@
-module Exn_bt = Picos_exn_bt
-
-module Trigger = struct
-  include Picos_private_bootstrap.Trigger
-
-  type _ Effect.t += Await : t -> Exn_bt.t option Effect.t
-
-  let await (t : t) =
-    match Atomic.get t with
-    | Initial -> Effect.perform (Await t)
-    | Signaled -> None
-    | Awaiting _ -> awaiting ()
-end
-
-module Fiber = struct
-  include Picos_private_bootstrap.Fiber
-
-  let continue (Fiber r : Picos_private_bootstrap.Fiber.t) k v =
-    if r.forbid then Effect.Deep.continue k v
-    else
-      match Atomic.get r.computation with
-      | Canceled exn_bt -> Exn_bt.discontinue k exn_bt
-      | Continue _ | Returned _ -> Effect.Deep.continue k v
-
-  let continue_with (Fiber r) k v h =
-    if r.forbid then Effect.Shallow.continue_with k v h
-    else
-      match Atomic.get r.computation with
-      | Canceled exn_bt -> Exn_bt.discontinue_with k exn_bt h
-      | Continue _ | Returned _ -> Effect.Shallow.continue_with k v h
-
-  type _ Effect.t += Current : t Effect.t
-
-  let current () = Effect.perform Current
-
-  type _ Effect.t +=
-    | Spawn : {
-        forbid : bool;
-        computation : 'a Picos_private_bootstrap.Computation.t;
-        mains : (unit -> unit) list;
-      }
-        -> unit Effect.t
-
-  let spawn ~forbid computation mains =
-    Effect.perform @@ Spawn { forbid; computation; mains }
-
-  type _ Effect.t += Yield : unit Effect.t
-
-  let yield () = Effect.perform Yield
-end
-
-module Computation = struct
-  include Picos_private_bootstrap.Computation
-
-  type _ Effect.t +=
-    | Cancel_after : {
-        seconds : float;
-        exn_bt : Exn_bt.t;
-        computation : 'a t;
-      }
-        -> unit Effect.t
-
-  let cancel_after computation ~seconds exn_bt =
-    if 0.0 <= seconds then
-      Effect.perform (Cancel_after { seconds; exn_bt; computation })
-    else negative ()
-end
diff --git a/lib/picos/picos.common.ml b/lib/picos/picos.common.ml
index 6ec8b5f..f9f0209 100644
--- a/lib/picos/picos.common.ml
+++ b/lib/picos/picos.common.ml
@@ -1,7 +1,13 @@
-include Picos_private
+module Exn_bt = Picos_exn_bt
+
+module Trigger = struct
+  include Picos_bootstrap.Trigger
+  include Picos_ocaml.Trigger
+end
 
 module Computation = struct
-  include Computation
+  include Picos_bootstrap.Computation
+  include Picos_ocaml.Computation
 
   let block t =
     let trigger = Trigger.create () in
@@ -14,9 +20,14 @@ module Computation = struct
     end
     else t
 
-  let rec await t =
-    match Atomic.get t with
-    | Returned value -> value
-    | Canceled exn_bt -> Exn_bt.raise exn_bt
-    | Continue _ -> await (block t)
+  let await t = get_or block t
+end
+
+module Fiber = struct
+  include Picos_bootstrap.Fiber
+  include Picos_ocaml.Fiber
 end
+
+module type Implementation = Picos_bootstrap.Implementation
+
+let set_picos_implementation = Picos_ocaml.set_picos_implementation
diff --git a/lib/picos/picos.mli b/lib/picos/picos.mli
index 96e9f8c..67c00f8 100644
--- a/lib/picos/picos.mli
+++ b/lib/picos/picos.mli
@@ -219,8 +219,20 @@
 
 (** {2 Modules reference}
 
+    We first open the {!Picos} module
+
     {[open Picos]}
 
+    and set the Picos {{!implementations} implementation} to use on OCaml 4
+
+    {[
+      if String.starts_with ~prefix:"4." Sys.ocaml_version then
+        Picos.set_picos_implementation Picos_threaded.implementation
+    ]}
+
+    to {{!Picos_threaded.implementation} the basic thread based implementation}
+    for the example code snippets in this documentation.
+
     {3 Auxiliary modules} *)
 
 module Exn_bt = Picos_exn_bt
@@ -331,8 +343,8 @@ module Trigger : sig
       ℹ️ The behavior is that, {i unless [await] can return immediately},
 
       - on OCaml 5, [await] will perform the {!Await} effect, and
-      - on OCaml 4, [await] will suspend the underlying system level thread
-        using a per thread [Mutex] and [Condition] variable.
+      - on OCaml 4, [await] will call the {{!Implementation.Trigger.await}
+        [await] operation} of the {{!set_picos_implementation} implementation}.
 
       @raise Invalid_argument if the trigger was in the awaiting state, which
         means that multiple concurrent calls of [await] are being made. *)
@@ -648,13 +660,15 @@ module Computation : sig
       backtrace.  Completion of the computation before the specified time
       effectively cancels the timeout.
 
-      ℹ️ The behaviour is that [cancel_after] first checks that [seconds] is not
+      ℹ️ The behavior is that [cancel_after] first checks that [seconds] is not
       negative, and then
 
       - on OCaml 5, [cancel_after] will perform the {!Cancel_after}
         effect, and
-      - on OCaml 4, [cancel_after] registers the timeout to a background
-        [Thread] running a [Unix.select] loop.
+      - on OCaml 4, [cancel_after] will next call {!Fiber.current} and will
+        finally call the {{!Implementation.Computation.cancel_after}
+        [cancel_after] operation} of the {{!set_picos_implementation}
+        implementation}.
 
       @raise Invalid_argument if [seconds] is negative or too large as
         determined by the scheduler. *)
@@ -783,7 +797,9 @@ module Fiber : sig
       ℹ️ The behavior is that
 
       - on OCaml 5, [yield] perform the {!Yield} effect, and
-      - on OCaml 4, [yield] calls [Thread.yield ()]. *)
+      - on OCaml 4, [yield] will next call {!Fiber.current} and will finally
+        call the {{!Implementation.Fiber.yield} [yield] operation} of the
+        {{!set_picos_implementation} implementation}. *)
 
   (** {2 Interface for spawning} *)
 
@@ -806,7 +822,9 @@ module Fiber : sig
       ℹ️ The behavior is that
 
       - on OCaml 5, [spawn] performs the {!Spawn} effect, and
-      - on OCaml 4, [spawn] creates a new [Thread] for each fiber. *)
+      - on OCaml 4, [spawn] will next call {!Fiber.current} and will finally
+        call the {{!Implementation.Fiber.spawn} [spawn] operation} of the
+        {{!set_picos_implementation} implementation}. *)
 
   (** {2 Interface for current fiber} *)
 
@@ -828,7 +846,9 @@ module Fiber : sig
       ℹ️ The behavior is that
 
       - on OCaml 5, [current] performs the {!Current} effect, and
-      - on OCaml 4, [current] obtains the fiber from [Thread]-local storage. *)
+      - on OCaml 4, [current] will call the {{!Implementation.Fiber.current}
+        [current] operation} of the {{!set_picos_implementation}
+        implementation}. *)
 
   val has_forbidden : t -> bool
   (** [has_forbidden fiber] determines whether the fiber {!forbid}s or
@@ -1015,39 +1035,47 @@ module Fiber : sig
       next. *)
 end
 
-(** {2 Advanced topics}
+(** {2 Implementations}
 
-    {3 Default behaviors}
+    OCaml 4 does not have effect handlers.  It is, however, possible to
+    implement the effects based operations of Picos in terms of threads. *)
 
-    All of the effects based operations
+module type Implementation = sig
+  (** Signature for an implementation of the primitive effects based operations of
+      Picos.
 
-    - {!Trigger.await},
-    - {!Computation.cancel_after},
-    - {!Fiber.current},
-    - {!Fiber.yield}, and
-    - {!Fiber.spawn}
+      ℹ️ On OCaml 4 most of the effects based operations will call
+      {!Fiber.current} implicitly to check for cancelation before finally
+      calling the set {{!set_picos_implementation} implementation}. *)
 
-    have default behaviors on OCaml 4.
+  (** Implements primitive effects based operations of {!Picos.Fiber}. *)
+  module Fiber : sig
+    val current : unit -> Fiber.t
+    (** See {!Picos.Fiber.current}. *)
 
-    The underlying idea behind the defaults is to make it so that a fiber
-    corresponds to a [Thread].
+    val spawn : forbid:bool -> 'a Computation.t -> (unit -> unit) list -> unit
+    (** See {!Picos.Fiber.spawn}. *)
 
-    Briefly:
+    val yield : unit -> unit
+    (** See {!Picos.Fiber.yield}. *)
+  end
+
+  (** Implements primitive effects based operations of {!Picos.Computation}. *)
+  module Computation : sig
+    val cancel_after : 'a Computation.t -> seconds:float -> Exn_bt.t -> unit
+    (** See {!Picos.Computation.cancel_after}. *)
+  end
+
+  (** Implements primitive effects based operations of {!Picos.Trigger}. *)
+  module Trigger : sig
+    val await : Trigger.t -> Exn_bt.t option
+    (** See {!Picos.Trigger.await}. *)
+  end
+end
 
-    - The default {{!Fiber.spawn} [spawn]} creates a thread for each fiber.
-    - The default {{!Fiber.current} [current]} uses {!Picos_tls} to store the current
-      fiber.
-    - The default {{!Fiber.yield} [yield]} just calls [Thread.yield].
-    - The default {{!Trigger.await} [await]} uses {!Picos_tls} to store a [Mutex] and
-      [Condition] to suspend the thread.
-    - The default {{!Computation.cancel_after} [cancel_after]} uses
-      {!Picos_domain.DLS} to store a priority queue of timeouts and a per-domain
-      background timeout thread that runs a [Unix.select] loop to cancel
-      computations.
+val set_picos_implementation : (module Implementation) -> unit
+(** [set_picos_implementation (module Implementation)] sets the implementation
+    of the primitive effects based operations of Picos.
 
-    The default behaviors initialize their resources, per-thread, and per-domain
-    state and the background timeout thread, only when actually used.  If the
-    default {{!Computation.cancel_after} [cancel_after]} is not used, no
-    background timeout thread will be created and no per-domain state will be
-    used.  If none of the defaults are used, no per-thread state will be
-    used. *)
+    ⚠️ This must be called exactly once in a program only on OCaml 4 before using
+    any effects based operations of Picos. *)
diff --git a/lib/picos/select/dune b/lib/picos/select/dune
new file mode 100644
index 0000000..2a309bf
--- /dev/null
+++ b/lib/picos/select/dune
@@ -0,0 +1,13 @@
+(library
+ (name picos_select)
+ (public_name picos.select)
+ (optional)
+ (libraries
+  (re_export picos)
+  (re_export picos.exn_bt)
+  picos.domain
+  threads.posix
+  psq
+  mtime
+  mtime.clock.os
+  unix))
diff --git a/lib/picos/select/picos_select.ml b/lib/picos/select/picos_select.ml
new file mode 100644
index 0000000..a2ff894
--- /dev/null
+++ b/lib/picos/select/picos_select.ml
@@ -0,0 +1,280 @@
+let[@poll error] [@inline never] compare_and_set x before after =
+  !x == before
+  && begin
+       x := after;
+       true
+     end
+
+let[@poll error] [@inline never] exchange x after =
+  let before = !x in
+  x := after;
+  before
+
+(* *)
+
+open Picos
+
+type cancel_at =
+  | Cancel_at : {
+      time : Mtime.span;
+      exn_bt : Exn_bt.t;
+      computation : 'a Computation.t;
+    }
+      -> cancel_at
+
+module Q =
+  Psq.Make
+    (Int)
+    (struct
+      type t = cancel_at
+
+      let compare (Cancel_at l) (Cancel_at r) = Mtime.Span.compare l.time r.time
+    end)
+
+type return_on =
+  | Return_on : {
+      file_descr : Unix.file_descr;
+      value : 'a;
+      computation : 'a Computation.t;
+    }
+      -> return_on
+
+type state = {
+  phase : [ `Select | `Continue | `Process ] Atomic.t;
+  mutable state : [ `Initial | `Starting | `Alive | `Stopping | `Stopped ];
+  mutable exn_bt : Exn_bt.t;
+  mutable pipe_inn : Unix.file_descr;
+  mutable pipe_out : Unix.file_descr;
+  byte : Bytes.t;
+  (* *)
+  timeouts : Q.t Atomic.t;
+  mutable next_id : int;
+  (* *)
+  new_rd : return_on list ref;
+  new_wr : return_on list ref;
+  new_ex : return_on list ref;
+}
+
+let exit_exn_bt = Exn_bt.get_callstack 0 Exit
+
+let key =
+  Picos_domain.DLS.new_key @@ fun () ->
+  {
+    phase = Atomic.make `Continue;
+    state = `Initial;
+    exn_bt = exit_exn_bt;
+    pipe_inn = Unix.stdin;
+    pipe_out = Unix.stdin;
+    byte = Bytes.create 1;
+    timeouts = Atomic.make Q.empty;
+    next_id = 0;
+    new_rd = ref [];
+    new_wr = ref [];
+    new_ex = ref [];
+  }
+
+let[@poll error] [@inline never] try_transition s from into =
+  s.state == from
+  && begin
+       s.state <- into;
+       true
+     end
+
+let[@poll error] [@inline never] transition s into =
+  let from = s.state in
+  s.state <- into;
+  from
+
+let rec wakeup s from =
+  match Atomic.get s.phase with
+  | `Process ->
+      (* The thread will process the fds and timeouts before next select. *)
+      ()
+  | `Continue ->
+      if Atomic.compare_and_set s.phase `Continue `Process then
+        (* We managed to signal the wakeup before the thread was ready to call
+           select and the thread will notice this without us needing to write to
+           the pipe. *)
+        ()
+      else
+        (* Either the thread called select or another wakeup won the race.  We
+           need to retry. *)
+        wakeup s from
+  | `Select ->
+      if Atomic.compare_and_set s.phase `Select `Continue then
+        if s.state == from then
+          (* We are now responsible for writing to the pipe to force the thread to
+             exit the select. *)
+          let n = Unix.write s.pipe_out s.byte 0 1 in
+          assert (n = 1)
+
+type fos = { n : int; unique_fds : Unix.file_descr list; ops : return_on list }
+
+let fos_empty = { n = 1; unique_fds = []; ops = [] }
+
+module Ht = Hashtbl.Make (struct
+  type t = Unix.file_descr
+
+  let equal : t -> t -> bool = ( == )
+
+  let hash (fd : t) =
+    if Obj.is_int (Obj.repr fd) then Obj.magic fd else Hashtbl.hash fd
+end)
+
+let rec process_fds ht unique_fds ops = function
+  | [] ->
+      if unique_fds == [] && ops == [] then fos_empty
+      else { n = Ht.length ht; unique_fds; ops }
+  | (Return_on r as op) :: ops_todo ->
+      if Computation.is_running r.computation then begin
+        match Ht.find ht r.file_descr with
+        | `Return ->
+            Computation.return r.computation r.value;
+            process_fds ht unique_fds ops ops_todo
+        | `Alive -> process_fds ht unique_fds (op :: ops) ops_todo
+        | exception Not_found ->
+            Ht.add ht r.file_descr `Alive;
+            process_fds ht (r.file_descr :: unique_fds) (op :: ops) ops_todo
+      end
+      else process_fds ht unique_fds ops ops_todo
+
+let process_fds unique_fds fos new_ops =
+  if fos.ops == [] && new_ops == [] then fos_empty
+  else
+    let ht = Ht.create fos.n in
+    unique_fds |> List.iter (fun fd -> Ht.add ht fd `Return);
+    let r = process_fds ht [] [] fos.ops in
+    if new_ops == [] then r else process_fds ht r.unique_fds r.ops new_ops
+
+let rec process_timeouts s =
+  let before = Atomic.get s.timeouts in
+  match Q.pop before with
+  | None -> -1.0
+  | Some ((_, Cancel_at e), after) ->
+      let elapsed = Mtime_clock.elapsed () in
+      if Mtime.Span.compare e.time elapsed <= 0 then begin
+        if Atomic.compare_and_set s.timeouts before after then
+          Computation.cancel e.computation e.exn_bt;
+        process_timeouts s
+      end
+      else
+        Mtime.Span.to_float_ns (Mtime.Span.abs_diff e.time elapsed)
+        *. (1. /. 1_000_000_000.)
+
+let rec select_thread s timeout rd wr ex =
+  if s.state == `Alive then
+    if Atomic.compare_and_set s.phase `Continue `Select then
+      Unix.select
+        (s.pipe_inn :: rd.unique_fds)
+        wr.unique_fds ex.unique_fds timeout
+      |> select_thread_continue s rd wr ex
+    else select_thread_continue s rd wr ex ([], [], [])
+
+and select_thread_continue s rd wr ex (rd_fds, wr_fds, ex_fds) =
+  begin
+    match Atomic.exchange s.phase `Continue with
+    | `Select | `Process -> ()
+    | `Continue ->
+        let n = Unix.read s.pipe_inn s.byte 0 1 in
+        assert (n = 1)
+  end;
+  let rd = process_fds rd_fds rd (exchange s.new_rd []) in
+  let wr = process_fds wr_fds wr (exchange s.new_wr []) in
+  let ex = process_fds ex_fds ex (exchange s.new_ex []) in
+  let tos = process_timeouts s in
+  select_thread s tos rd wr ex
+
+let select_thread s =
+  begin
+    try
+      let pipe_inn, pipe_out = Unix.pipe ~cloexec:true () in
+      s.pipe_inn <- pipe_inn;
+      s.pipe_out <- pipe_out;
+      if try_transition s `Starting `Alive then
+        select_thread s (-1.0) fos_empty fos_empty fos_empty
+    with exn -> s.exn_bt <- Exn_bt.get exn
+  end;
+  transition s `Stopped |> ignore;
+  if s.pipe_inn != Unix.stdin then Unix.close s.pipe_inn;
+  if s.pipe_out != Unix.stdin then Unix.close s.pipe_out
+
+let[@inline never] init s =
+  if try_transition s `Initial `Starting then begin
+    match Thread.create select_thread s with
+    | thread ->
+        Picos_domain.at_exit @@ fun () ->
+        if try_transition s `Alive `Stopping then wakeup s `Stopping;
+        Thread.join thread;
+        if s.exn_bt != exit_exn_bt then Exn_bt.raise s.exn_bt
+    | exception exn ->
+        transition s `Stopped |> ignore;
+        raise exn
+  end;
+  while s.state == `Starting do
+    Thread.yield ()
+  done;
+  if s.state != `Alive then
+    invalid_arg "Picos_select: domain has been terminated"
+
+let get () =
+  let s = Picos_domain.DLS.get key in
+  if s.state != `Alive then init s;
+  s
+
+(* *)
+
+let[@poll error] [@inline never] next_id t =
+  let id = t.next_id in
+  t.next_id <- id + 1;
+  id
+
+let rec add_timeout s id entry =
+  let before = Atomic.get s.timeouts in
+  let after = Q.add id entry before in
+  if Atomic.compare_and_set s.timeouts before after then
+    match Q.min after with
+    | Some (id', _) -> if id = id' then wakeup s `Alive
+    | None -> ()
+  else add_timeout s id entry
+
+let rec remove_action _trigger s id =
+  let before = Atomic.get s.timeouts in
+  let after = Q.remove id before in
+  if not (Atomic.compare_and_set s.timeouts before after) then
+    remove_action (Obj.magic ()) s id
+
+let[@alert "-handler"] cancel_after computation ~seconds exn_bt =
+  match Mtime.Span.of_float_ns (seconds *. 1_000_000_000.) with
+  | None ->
+      invalid_arg
+        "Picos_select: seconds should be between 0 to pow(2, 53) nanoseconds"
+  | Some span ->
+      let s = get () in
+      let time = Mtime.Span.add (Mtime_clock.elapsed ()) span in
+      let entry = Cancel_at { time; exn_bt; computation } in
+      let id = next_id s in
+      add_timeout s id entry;
+      let remover = Trigger.from_action s id remove_action in
+      if not (Computation.try_attach computation remover) then
+        Trigger.signal remover
+
+(* *)
+
+let wakeup_action _trigger s _s = wakeup s `Alive
+
+let[@alert "-handler"] rec insert_fd s fds (Return_on r as op) =
+  let before = !fds in
+  if Computation.is_running r.computation then
+    if compare_and_set fds before (Return_on r :: before) then
+      let _ : bool =
+        Computation.try_attach r.computation
+          (Trigger.from_action s s wakeup_action)
+      in
+      wakeup s `Alive
+    else insert_fd s fds op
+
+let return_on computation file_descr event value =
+  let s = get () in
+  insert_fd s
+    (match event with `R -> s.new_rd | `W -> s.new_wr | `E -> s.new_ex)
+    (Return_on { computation; file_descr; value })
diff --git a/lib/picos/select/picos_select.mli b/lib/picos/select/picos_select.mli
new file mode 100644
index 0000000..96874b2
--- /dev/null
+++ b/lib/picos/select/picos_select.mli
@@ -0,0 +1,27 @@
+(** A basic [Unix.select] based IO event loop for {!Picos}.
+
+    The operations in this module automatically manage a [Thread] per domain
+    that runs a [Unix.select] loop to support the operations.
+
+    ⚠️ All the usual limitations of the [Unix] module apply. *)
+
+open Picos
+
+val cancel_after : _ Computation.t -> seconds:float -> Exn_bt.t -> unit
+(** [cancel_after computation ~seconds exn_bt] arranges for [computation] to be
+    {{!Picos.Computation.cancel} canceled} with given [exn_bt] after given time
+    in [seconds].  Completion of the [computation] before the specified time
+    effectively cancels the timeout.
+
+    ℹ️ You can use [cancel_after] to implement the handler for the
+    {{!Picos.Computation.Cancel_after} [Cancel_after]} effect. *)
+
+val return_on :
+  'a Computation.t -> Unix.file_descr -> [ `R | `W | `E ] -> 'a -> unit
+(** [return_on computation file_descr op value] arranges for [computation] to be
+    {{!Picos.Computation.return} returned} with given [value] when [file_descr]
+    becomes available for [op].  Completion of the [computation] before the
+    [file_descr] becomes available for [op] effectively cancels the await.
+
+    ℹ️ Using [Unix.set_nonblock] and [return_on] you can implement direct-style
+    transparently asynchronous IO on top of the [Unix] module. *)
diff --git a/lib/picos/default/dune b/lib/picos/threaded/dune
similarity index 60%
rename from lib/picos/default/dune
rename to lib/picos/threaded/dune
index b3145a5..f4efbf4 100644
--- a/lib/picos/default/dune
+++ b/lib/picos/threaded/dune
@@ -1,33 +1,29 @@
 (library
- (name picos_default)
- (package picos)
+ (name picos_threaded)
+ (public_name picos.threaded)
  (js_of_ocaml
   (javascript_files thread.js))
  (libraries
-  picos_private_bootstrap
-  picos_tls
-  picos_domain
-  picos_exn_bt
-  psq
-  mtime
-  mtime.clock.os
+  (re_export picos)
+  (re_export picos.exn_bt)
+  picos.tls
+  picos.domain
   (select
-   systhreads.ml
+   select.ml
    from
-   (threads.posix -> systhreads.posix.ml)
-   (-> systhreads.none.ml))
+   (picos.select -> select.some.ml)
+   (-> select.none.ml))
   (select
-   select.ml
+   systhreads.ml
    from
-   (unix -> select.unix.ml)
-   (-> select.none.ml))))
+   (threads.posix -> systhreads.posix.ml)
+   (-> systhreads.none.ml))))
 
 (rule
  (enabled_if
   (< %{ocaml_version} 5.0.0))
  (action
   (progn
-   (copy domain.ocaml4.ml domain.ml)
    (copy thread.ocaml4.js thread.js))))
 
 (rule
diff --git a/lib/picos/threaded/picos_threaded.ml b/lib/picos/threaded/picos_threaded.ml
new file mode 100644
index 0000000..a8f365a
--- /dev/null
+++ b/lib/picos/threaded/picos_threaded.ml
@@ -0,0 +1,82 @@
+open Picos
+
+let implementation =
+  (module struct
+    module Per_thread = struct
+      type t = { mutex : Mutex.t; condition : Condition.t; fiber : Fiber.t }
+
+      open struct
+        let create fiber =
+          let mutex = Mutex.create () and condition = Condition.create () in
+          { mutex; condition; fiber }
+
+        let per_thread_key =
+          Picos_tls.new_key @@ fun () ->
+          create (Fiber.create ~forbid:false (Computation.create ()))
+      end
+
+      let get () = Picos_tls.get per_thread_key
+      let set fiber = Picos_tls.set per_thread_key (create fiber)
+    end
+
+    module Trigger = struct
+      let release _ _ (per_thread : Per_thread.t) =
+        Mutex.lock per_thread.mutex;
+        Mutex.unlock per_thread.mutex;
+        Condition.broadcast per_thread.condition
+
+      let block trigger (per_thread : Per_thread.t) =
+        Mutex.lock per_thread.mutex;
+        match
+          while not (Trigger.is_signaled trigger) do
+            Condition.wait per_thread.condition per_thread.mutex
+          done
+        with
+        | () ->
+            Mutex.unlock per_thread.mutex;
+            Fiber.canceled per_thread.fiber
+        | exception exn ->
+            (* Condition.wait may be interrupted by asynchronous exceptions and we
+               must make sure to unlock even in that case. *)
+            Mutex.unlock per_thread.mutex;
+            raise exn
+
+      let[@alert "-handler"] await trigger =
+        let per_thread = Per_thread.get () in
+        if Fiber.has_forbidden per_thread.fiber then
+          if Trigger.on_signal trigger () per_thread release then
+            block trigger per_thread
+          else None
+        else if Fiber.try_attach per_thread.fiber trigger then
+          if Trigger.on_signal trigger () per_thread release then
+            block trigger per_thread
+          else begin
+            Fiber.detach per_thread.fiber trigger;
+            Fiber.canceled per_thread.fiber
+          end
+        else begin
+          Trigger.dispose trigger;
+          Fiber.canceled per_thread.fiber
+        end
+    end
+
+    module Computation = Select
+
+    module Fiber = struct
+      let current () =
+        let fiber = (Per_thread.get ()).fiber in
+        Fiber.check fiber;
+        fiber
+
+      let spawn ~forbid computation mains =
+        mains
+        |> List.iter @@ fun main ->
+           Systhreads.create
+             (fun () ->
+               main (Per_thread.set (Fiber.create ~forbid computation)))
+             ()
+           |> ignore
+
+      let yield () = Systhreads.yield ()
+    end
+  end : Implementation)
diff --git a/lib/picos/threaded/picos_threaded.mli b/lib/picos/threaded/picos_threaded.mli
new file mode 100644
index 0000000..7b8149b
--- /dev/null
+++ b/lib/picos/threaded/picos_threaded.mli
@@ -0,0 +1,29 @@
+(** A basic [Thread] based implementation of the effects based operations of
+    {!Picos}.
+
+    Briefly:
+
+    - [Fiber.current] uses {{!Picos_tls} [TLS]} to store the current fiber.
+    - [Fiber.spawn] creates a new [Thread] for each fiber.
+    - [Fiber.yield] just calls [Thread.yield].
+    - [Computation.cancel_after] uses {{!Picos_domain.DLS} [DLS]} to store a
+      priority queue of timeouts and a per-domain {{!Picos_select} background
+      timeout thread} that runs a [Unix.select] loop to cancel computations.
+    - [Trigger.await] uses {{!Picos_tls} [TLS]} to store a [Mutex] and
+      [Condition] to suspend the thread.
+
+    The operations initialize their resources, per-thread, and per-domain state
+    and {{!Picos_select} background timeout thread}, only when actually used.
+    If the default {{!Computation.cancel_after} [cancel_after]} is not used, no
+    background timeout thread will be created and no per-domain state will be
+    used.  If none of the defaults are used, no per-thread state will be
+    used.
+
+    ⚠️ This implementation is probably suitable for simple applications that do
+    not spawn a lot.  If an application uses a lot of short lived fibers, then a
+    more sophisticated implementation using some sort of thread pool will likely
+    perform significantly better. *)
+
+val implementation : (module Picos.Implementation)
+(** A basic implementation of the effects based operations of {!Picos} using
+    [Thread]s and a [Unix.select] based IO event loop. *)
diff --git a/lib/picos/threaded/select.none.ml b/lib/picos/threaded/select.none.ml
new file mode 100644
index 0000000..aa8d460
--- /dev/null
+++ b/lib/picos/threaded/select.none.ml
@@ -0,0 +1,2 @@
+let cancel_after _ ~seconds:_ _ =
+  raise (Sys_error "Computation: cancel_after unavailable")
diff --git a/lib/picos/threaded/select.some.ml b/lib/picos/threaded/select.some.ml
new file mode 100644
index 0000000..a64c5dc
--- /dev/null
+++ b/lib/picos/threaded/select.some.ml
@@ -0,0 +1,1 @@
+let cancel_after = Picos_select.cancel_after
diff --git a/lib/picos/default/systhreads.none.ml b/lib/picos/threaded/systhreads.none.ml
similarity index 100%
rename from lib/picos/default/systhreads.none.ml
rename to lib/picos/threaded/systhreads.none.ml
diff --git a/lib/picos/default/systhreads.posix.ml b/lib/picos/threaded/systhreads.posix.ml
similarity index 100%
rename from lib/picos/default/systhreads.posix.ml
rename to lib/picos/threaded/systhreads.posix.ml
diff --git a/lib/picos/default/thread.ocaml4.js b/lib/picos/threaded/thread.ocaml4.js
similarity index 100%
rename from lib/picos/default/thread.ocaml4.js
rename to lib/picos/threaded/thread.ocaml4.js
diff --git a/lib/picos/default/thread.ocaml5.js b/lib/picos/threaded/thread.ocaml5.js
similarity index 100%
rename from lib/picos/default/thread.ocaml5.js
rename to lib/picos/threaded/thread.ocaml5.js
diff --git a/lib/picos/tls/picos_tls.mli b/lib/picos/tls/picos_tls.mli
index 7a0a9bf..9b5437c 100644
--- a/lib/picos/tls/picos_tls.mli
+++ b/lib/picos/tls/picos_tls.mli
@@ -1,4 +1,4 @@
-(** Thread-local storage for Picos.
+(** Thread-local storage for {!Picos}.
 
     Note that here "thread" refers to system level threads rather than fibers or
     domains.  In case a system level thread implementation, i.e. the


Test patch:
diff --git a/test/dune b/test/dune
index 13b2604..a1b8830 100644
--- a/test/dune
+++ b/test/dune
@@ -6,8 +6,8 @@
   (select
    test_scheduler.ml
    from
-   (schedulers picos_default -> test_scheduler.ocaml5.ml)
-   (-> test_scheduler.ocaml4.ml))))
+   (picos.select schedulers -> test_scheduler.ocaml5.ml)
+   (picos.threaded -> test_scheduler.ocaml4.ml))))
 
 ;;
 
@@ -29,14 +29,12 @@
 (rule
  (action
   (progn
-   (copy
-    ../lib/picos/bootstrap/picos_private_bootstrap.ml
-    picos_private_bootstrap.ml))))
+   (copy ../lib/picos/bootstrap/picos_bootstrap.ml picos_bootstrap.ml))))
 
 (test
  (name test_picos_dscheck)
  (package picos)
- (modules test_picos_dscheck picos_private_bootstrap)
+ (modules test_picos_dscheck picos_bootstrap)
  (build_if
   (>= %{ocaml_version} 5))
  (libraries picos.exn_bt backoff traced_atomic dscheck alcotest))
@@ -74,4 +72,4 @@
   (flags --enable=effects))
  (build_if
   (<> %{architecture} i386))
- (libraries picos))
+ (libraries picos picos.threaded))
diff --git a/test/lib/elements/async_unix.ml b/test/lib/elements/async_unix.ml
index 85318a4..40c61fe 100644
--- a/test/lib/elements/async_unix.ml
+++ b/test/lib/elements/async_unix.ml
@@ -1,175 +1,25 @@
 open Picos
-open Foundation
-
-module Async = struct
-  (* TODO: Use better data structures for awaiters than lists. *)
-
-  module Awaiter = struct
-    type t = { file_descr : Unix.file_descr; trigger : Trigger.t }
-
-    let file_descr_of t = t.file_descr
-
-    let rec signal aws file_descr =
-      match aws with
-      | [] -> ()
-      | aw :: aws ->
-          if aw.file_descr == file_descr then Trigger.signal aw.trigger
-          else signal aws file_descr
-
-    let signal_or_wakeup wakeup aws file_descr =
-      if file_descr == wakeup then begin
-        let n = Unix.read file_descr (Bytes.create 1) 0 1 in
-        assert (n = 1)
-      end
-      else signal aws file_descr
-
-    let reject file_descr = List.filter (fun aw -> aw.file_descr != file_descr)
-  end
-
-  type per_domain = {
-    mutable state : [ `Init | `Locked | `Alive | `Dead ];
-    mutable pipe_inn : Unix.file_descr;
-    mutable pipe_out : Unix.file_descr;
-    mutable exn_bt : Exn_bt.t;
-    mutable needs_wakeup : bool;
-    reading : Awaiter.t list ref;
-    writing : Awaiter.t list ref;
-  }
-
-  let per_domain_key =
-    let dummy_pipe =
-      (* Unfortunately we cannot safely allocate a pipe here, so we use stdin as
-         a dummy value. *)
-      Unix.stdin
-    in
-    (* In this case we really want to use DLS rather than TLS. *)
-    Picos_domain.DLS.new_key @@ fun () ->
-    {
-      state = `Init;
-      pipe_inn = dummy_pipe;
-      pipe_out = dummy_pipe;
-      exn_bt = Exn_bt.get_callstack 0 Exit;
-      needs_wakeup = true;
-      reading = ref [];
-      writing = ref [];
-    }
-
-  let[@poll error] [@inline never] try_lock s =
-    s.state == `Init
-    && begin
-         s.state <- `Locked;
-         true
-       end
-
-  let[@poll error] [@inline never] unlock s state = s.state <- state
-
-  let[@poll error] [@inline never] needs_wakeup s =
-    s.needs_wakeup
-    && begin
-         s.needs_wakeup <- false;
-         true
-       end
-
-  let wakeup s =
-    if needs_wakeup s then
-      let n = Unix.write s.pipe_out (Bytes.create 1) 0 1 in
-      assert (n = 1)
-
-  let io_thread s =
-    begin
-      try
-        (* The pipe is used to wake up the select after changing the lists of
-           reading and writing file descriptors. *)
-        let pipe_inn, pipe_out = Unix.pipe ~cloexec:true () in
-        s.pipe_inn <- pipe_inn;
-        s.pipe_out <- pipe_out;
-        unlock s `Alive;
-        (* This is the IO select loop that performs select and then wakes up
-           fibers blocked on IO. *)
-        while s.state == `Alive do
-          let rs, ws, _ =
-            Unix.select
-              (s.pipe_inn :: List.map Awaiter.file_descr_of !(s.reading))
-              (List.map Awaiter.file_descr_of !(s.writing))
-              [] (-1.0)
-          in
-          List.iter (Awaiter.signal_or_wakeup s.pipe_inn !(s.reading)) rs;
-          List.iter (Awaiter.signal !(s.writing)) ws;
-          Thread_atomic.modify s.reading (List.fold_right Awaiter.reject rs);
-          Thread_atomic.modify s.writing (List.fold_right Awaiter.reject ws);
-          s.needs_wakeup <- true
-        done
-      with exn -> s.exn_bt <- Exn_bt.get exn
-    end;
-    s.needs_wakeup <- false;
-    unlock s `Dead;
-    if s.pipe_inn != Unix.stdin then Unix.close s.pipe_inn;
-    if s.pipe_out != Unix.stdin then Unix.close s.pipe_out
-
-  let start s =
-    (* DLS initialization may be run multiple times, so we perform more involved
-       initialization here. *)
-    match Thread.create io_thread s with
-    | thread ->
-        Picos_domain.at_exit @@ fun () ->
-        unlock s `Dead;
-        wakeup s;
-        Thread.join thread;
-        if s.exn_bt.exn != Exit then Exn_bt.raise s.exn_bt
-    | exception exn ->
-        unlock s `Dead;
-        raise exn
-
-  let wait s =
-    while s.state == `Locked do
-      Thread.yield ()
-    done;
-    if s.state != `Alive then
-      invalid_arg "Async_unix: domain has been terminated"
-
-  let init s =
-    if try_lock s then start s;
-    wait s
-
-  let get () =
-    let s = Picos_domain.DLS.get per_domain_key in
-    if s.state != `Alive then init s;
-    s
-
-  let await s r file_descr =
-    let trigger = Trigger.create () in
-    let awaiter = Awaiter.{ file_descr; trigger } in
-    Thread_atomic.modify r (List.cons awaiter);
-    wakeup s;
-    match Trigger.await trigger with
-    | None -> ()
-    | Some exn_bt ->
-        Thread_atomic.modify r (List.drop_first awaiter);
-        Exn_bt.raise exn_bt
-end
-
 include Unix
 
+let await file_descr op =
+  let c = Computation.create () in
+  Picos_select.return_on c file_descr op file_descr;
+  try Computation.await c
+  with exn ->
+    Computation.cancel c (Exn_bt.get_callstack 0 exn);
+    raise exn
+
 let read file_descr bytes pos len =
-  match Unix.read file_descr bytes pos len with
-  | result -> result
-  | exception Unix.Unix_error ((EAGAIN | EWOULDBLOCK), _, _) ->
-      let s = Async.get () in
-      Async.await s s.reading file_descr;
-      Unix.read file_descr bytes pos len
+  try Unix.read file_descr bytes pos len
+  with Unix.Unix_error ((EAGAIN | EWOULDBLOCK), _, _) ->
+    Unix.read (await file_descr `R) bytes pos len
 
 let write file_descr bytes pos len =
-  match Unix.write file_descr bytes pos len with
-  | result -> result
-  | exception Unix.Unix_error ((EAGAIN | EWOULDBLOCK), _, _) ->
-      let s = Async.get () in
-      Async.await s s.writing file_descr;
-      Unix.write file_descr bytes pos len
+  try Unix.write file_descr bytes pos len
+  with Unix.Unix_error ((EAGAIN | EWOULDBLOCK), _, _) ->
+    Unix.write (await file_descr `W) bytes pos len
 
 let accept ?cloexec file_descr =
-  match Unix.accept ?cloexec file_descr with
-  | result -> result
-  | exception Unix.Unix_error ((EAGAIN | EWOULDBLOCK), _, _) ->
-      let s = Async.get () in
-      Async.await s s.reading file_descr;
-      Unix.accept ?cloexec file_descr
+  try Unix.accept ?cloexec file_descr
+  with Unix.Unix_error ((EAGAIN | EWOULDBLOCK), _, _) ->
+    Unix.accept ?cloexec (await file_descr `R)
diff --git a/test/lib/elements/async_unix.mli b/test/lib/elements/async_unix.mli
index 42b8974..3d88aeb 100644
--- a/test/lib/elements/async_unix.mli
+++ b/test/lib/elements/async_unix.mli
@@ -1,3 +1,7 @@
-(** A transparently asynchronous Unix module for Picos *)
+(** A transparently asynchronous [Unix] module for {!Picos}. *)
 
 include module type of Unix
+
+val await : file_descr -> [ `R | `W | `E ] -> file_descr
+(** [await file_descr op] awaits for [file_descr] to be ready for [op] and
+    returns [file_descr]. *)
diff --git a/test/lib/elements/dune b/test/lib/elements/dune
index d518d42..50e42df 100644
--- a/test/lib/elements/dune
+++ b/test/lib/elements/dune
@@ -1,3 +1,3 @@
 (library
  (name elements)
- (libraries picos picos.domain foundation backoff threads.posix unix))
+ (libraries picos picos.domain picos.select foundation backoff unix))
diff --git a/test/test_picos_dscheck.ml b/test/test_picos_dscheck.ml
index 3444d7e..d1716b2 100644
--- a/test/test_picos_dscheck.ml
+++ b/test/test_picos_dscheck.ml
@@ -2,7 +2,7 @@
     interleavings of [Atomic] operations performed by different [Atomic.spawn]ed
     fibers. *)
 
-open Picos_private_bootstrap
+open Picos_bootstrap
 
 let sum_as fn xs = Array.fold_left (fun sum x -> sum + fn x) 0 xs
 let ( += ) x y = x := !x + y
@@ -56,9 +56,7 @@ let test_computation_contract () =
        in
        let () =
          Atomic.spawn @@ fun () ->
-         if
-           Computation.try_cancel computation
-             (Picos_exn_bt.get_callstack 1 Exit)
+         if Computation.try_cancel computation (Exn_bt.get_callstack 1 Exit)
          then incr cancels
        in
        let triggers = Array.init 2 @@ fun _ -> Trigger.create () in
diff --git a/test/test_scheduler.ocaml4.ml b/test/test_scheduler.ocaml4.ml
index a64eaf6..12d10e7 100644
--- a/test/test_scheduler.ocaml4.ml
+++ b/test/test_scheduler.ocaml4.ml
@@ -1,1 +1,2 @@
+let () = Picos.set_picos_implementation Picos_threaded.implementation
 let run main = main ()
diff --git a/test/test_scheduler.ocaml5.ml b/test/test_scheduler.ocaml5.ml
index 502ba16..3b4fa89 100644
--- a/test/test_scheduler.ocaml5.ml
+++ b/test/test_scheduler.ocaml5.ml
@@ -1,13 +1,11 @@
 let[@alert "-handler"] on_unhandled =
-  let module Default = Picos_default.Make (Picos) in
   let try_handle :
       type a. a Effect.t -> ((a, _) Effect.Deep.continuation -> _) option =
     function
     | Picos.Computation.Cancel_after r ->
         Some
           (fun k ->
-            Default.Computation.cancel_after r.computation ~seconds:r.seconds
-              r.exn_bt;
+            Picos_select.cancel_after r.computation ~seconds:r.seconds r.exn_bt;
             Effect.Deep.continue k ())
     | _ -> None
   in