Repository: ocaml-multicore/picos
Pull request #62 (2024-03-20 10:47:54)

Publish the sample FIFO scheduler
This PR:
* Moves the FIFO scheduler into a public library `picos_fifos` of the `picos` package.
* Changes FIFO scheduler to directly use `Picos_select` to handle the `Cancel_after` effect.
* Adds a non-allocating `is_canceled` operation for computations and fibers.
* Enhances the MPSC queue used in the FIFO scheduler to allow producers to push elements to both ends of the queue and changes the FIFO scheduler to give priority to canceled fibers.
* Moves the MPSC queue to a public library `picos_mpsc_queue` of the `picos` package.
* Changes the interface of the `Picos_threaded` scheduler to match the interface of the `Picos_fifos` schedulers.


Patch:
diff --git a/TODO.md b/TODO.md
index a00f5bf..7bf3aad 100644
--- a/TODO.md
+++ b/TODO.md
@@ -10,8 +10,6 @@
 
 - JavaScript scheduler
 
-- Change example scheduler to give priority to canceled fibers
-
 - Should cancelation exception be raised only once?
 
 - Test what performance of `Fiber.current` could be with TLS
diff --git a/bench/bench_mpsc_queue.ml b/bench/bench_mpsc_queue.ml
index 60f6ffc..55904ea 100644
--- a/bench/bench_mpsc_queue.ml
+++ b/bench/bench_mpsc_queue.ml
@@ -1,17 +1,17 @@
 open Multicore_bench
-module Queue = Foundation.Mpsc_queue
+module Queue = Picos_mpsc_queue
 
 let run_one_domain ~budgetf ?(n_msgs = 50 * Util.iter_factor) () =
   let t = Queue.create () in
 
   let op push =
-    if push then Queue.enqueue t 101
-    else match Queue.dequeue t with _ -> () | exception Queue.Empty -> ()
+    if push then Queue.push t 101
+    else match Queue.pop_exn t with _ -> () | exception Queue.Empty -> ()
   in
 
   let init _ =
     assert (
-      match Queue.dequeue t with _ -> false | exception Queue.Empty -> true);
+      match Queue.pop_exn t with _ -> false | exception Queue.Empty -> true);
     Util.generate_push_and_pop_sequence n_msgs
   in
   let work _ bits = Util.Bits.iter op bits in
@@ -31,7 +31,7 @@ let run_one ~budgetf ~n_adders () =
 
   let init _ =
     assert (
-      match Queue.dequeue t with _ -> false | exception Queue.Empty -> true);
+      match Queue.pop_exn t with _ -> false | exception Queue.Empty -> true);
     Atomic.set n_msgs_to_add n_msgs
   in
   let work i () =
@@ -40,7 +40,7 @@ let run_one ~budgetf ~n_adders () =
         let n = Util.alloc n_msgs_to_add in
         if 0 < n then begin
           for i = 1 to n do
-            Queue.enqueue t i
+            Queue.push t i
           done;
           work ()
         end
@@ -49,7 +49,7 @@ let run_one ~budgetf ~n_adders () =
     else
       let rec loop n =
         if 0 < n then
-          match Queue.dequeue t with
+          match Queue.pop_exn t with
           | _ -> loop (n - 1)
           | exception Queue.Empty ->
               Domain.cpu_relax ();
@@ -64,7 +64,6 @@ let run_one ~budgetf ~n_adders () =
     in
     Printf.sprintf "%s, 1 nb taker" (format "nb adder" n_adders)
   in
-
   Times.record ~budgetf ~n_domains ~init ~work ()
   |> Times.to_thruput_metrics ~n:n_msgs ~singular:"message" ~config
 
diff --git a/bench/dune b/bench/dune
index 6fe82a4..554af3e 100644
--- a/bench/dune
+++ b/bench/dune
@@ -1,12 +1,3 @@
-(* -*- tuareg -*- *)
-
-let maybe_domain_shims =
-  if Jbuild_plugin.V1.ocaml_version < "5" then "domain_shims" else ""
-
-let () =
-  Jbuild_plugin.V1.send
-  @@ {|
-
 (test
  (name main)
  (action
@@ -14,11 +5,12 @@ let () =
  (package picos)
  (libraries
   picos
-  foundation
-  (select scheduler.ml from
-   (schedulers -> scheduler.ocaml5.ml)
+  picos_mpsc_queue
+  (select
+   scheduler.ml
+   from
+   (picos.fifos -> scheduler.ocaml5.ml)
    (picos.threaded -> scheduler.ocaml4.ml))
   multicore-bench
-  multicore-magic |}
-  ^ maybe_domain_shims ^ {| ))
-|}
+  multicore-magic
+  domain_shims))
diff --git a/bench/scheduler.ocaml4.ml b/bench/scheduler.ocaml4.ml
index 03a9af7..f741a3e 100644
--- a/bench/scheduler.ocaml4.ml
+++ b/bench/scheduler.ocaml4.ml
@@ -1,6 +1,1 @@
-open Picos
-
-let run action =
-  Handler.using Picos_threaded.handler
-    (Picos_threaded.create ~forbid:false (Computation.create ()))
-    action
+let run main = Picos_threaded.run ~forbid:false main
diff --git a/bench/scheduler.ocaml5.ml b/bench/scheduler.ocaml5.ml
index 7ce512d..bed2960 100644
--- a/bench/scheduler.ocaml5.ml
+++ b/bench/scheduler.ocaml5.ml
@@ -1,3 +1,1 @@
-open Schedulers
-
-let run action = Fifos.run ~forbid:false action
+let run main = Picos_fifos.run ~forbid:false main
diff --git a/dune-project b/dune-project
index f306d96..bd2988a 100644
--- a/dune-project
+++ b/dune-project
@@ -32,11 +32,10 @@
    (>= 2.0.0))
   (psq
    (>= 0.2.1))
-  ;; Test dependencies
+  ;; For mpsc queue
   (multicore-magic
-   (and
-    (>= 2.1.0)
-    :with-test))
+   (>= 2.1.0))
+  ;; Test dependencies
   (multicore-bench
    (and
     (>= 0.1.2)
@@ -53,6 +52,10 @@
    (and
     (>= 0.3)
     :with-test))
+  (qcheck-multicoretests-util
+   (and
+    (>= 0.3)
+    :with-test))
   (mdx
    (and
     (>= 2.4.0)
diff --git a/lib/picos/bootstrap/picos_bootstrap.ml b/lib/picos/bootstrap/picos_bootstrap.ml
index 047b171..3a51c87 100644
--- a/lib/picos/bootstrap/picos_bootstrap.ml
+++ b/lib/picos/bootstrap/picos_bootstrap.ml
@@ -63,6 +63,11 @@ module Computation = struct
     let balance_and_mode = Bool.to_int (mode == `FIFO) in
     Atomic.make (Continue { balance_and_mode; triggers = [] })
 
+  let is_canceled t =
+    match Atomic.get t with
+    | Canceled _ -> true
+    | Returned _ | Continue _ -> false
+
   let canceled t =
     match Atomic.get t with
     | Canceled exn_bt -> Some exn_bt
@@ -205,6 +210,9 @@ module Fiber = struct
   let create ~forbid computation = Fiber { computation; forbid; fls = [||] }
   let has_forbidden (Fiber r) = r.forbid
 
+  let is_canceled (Fiber r) =
+    (not r.forbid) && Computation.is_canceled r.computation
+
   let canceled (Fiber r) =
     if r.forbid then None else Computation.canceled r.computation
 
diff --git a/lib/picos/dune b/lib/picos/dune
index 3e9f55e..69d9ce3 100644
--- a/lib/picos/dune
+++ b/lib/picos/dune
@@ -16,8 +16,19 @@
    (picos_ocaml5 -> intf.ocaml5.ml))))
 
 (mdx
+ (enabled_if
+  (< %{ocaml_version} 5.0.0))
  (package picos)
  (deps
   (package picos))
  (libraries picos picos.exn_bt picos.threaded domain_shims)
  (files picos.mli))
+
+(mdx
+ (enabled_if
+  (>= %{ocaml_version} 5.0.0))
+ (package picos)
+ (deps
+  (package picos))
+ (libraries picos picos.exn_bt picos.fifos)
+ (files picos.mli))
diff --git a/lib/picos/intf.ocaml4.ml b/lib/picos/intf.ocaml4.ml
index 2cc4c99..e8139e4 100644
--- a/lib/picos/intf.ocaml4.ml
+++ b/lib/picos/intf.ocaml4.ml
@@ -11,4 +11,5 @@ end
 module type Fiber = sig
   type t
   type _ computation
+  type exn_bt
 end
diff --git a/lib/picos/intf.ocaml5.ml b/lib/picos/intf.ocaml5.ml
index 853ccf7..8974920 100644
--- a/lib/picos/intf.ocaml5.ml
+++ b/lib/picos/intf.ocaml5.ml
@@ -48,6 +48,19 @@ end
 module type Fiber = sig
   type t
   type _ computation
+  type exn_bt
+
+  val resume : t -> (exn_bt option, 'r) Effect.Deep.continuation -> 'r
+  (** [resume fiber k] is equivalent to
+      [Effect.Deep.continue k (Fiber.canceled t)]. *)
+
+  val resume_with :
+    t ->
+    (exn_bt option, 'b) Effect.Shallow.continuation ->
+    ('b, 'r) Effect.Shallow.handler ->
+    'r
+  (** [resume_with fiber k h] is equivalent to
+      [Effect.Shallow.continue_with k (Fiber.canceled t) h]. *)
 
   val continue : t -> ('v, 'r) Effect.Deep.continuation -> 'v -> 'r
   (** [continue fiber k v] is equivalent to:
diff --git a/lib/picos/ocaml5/picos_ocaml.ml b/lib/picos/ocaml5/picos_ocaml.ml
index 65b4c9a..8e84513 100644
--- a/lib/picos/ocaml5/picos_ocaml.ml
+++ b/lib/picos/ocaml5/picos_ocaml.ml
@@ -7,6 +7,9 @@ module Trigger = struct
 end
 
 module Fiber = struct
+  let resume t k = Effect.Deep.continue k (Fiber.canceled t)
+  let resume_with t k h = Effect.Shallow.continue_with k (Fiber.canceled t) h
+
   let continue t k v =
     match Fiber.canceled t with
     | None -> Effect.Deep.continue k v
diff --git a/lib/picos/picos.mli b/lib/picos/picos.mli
index 357fb5a..914a65f 100644
--- a/lib/picos/picos.mli
+++ b/lib/picos/picos.mli
@@ -223,19 +223,24 @@
 
     {[open Picos]}
 
-    and define a simple scheduler
+    and define a simple scheduler for running the examples in this document on
+    OCaml 4
 
-    {[
+    {@ocaml version<5.0.0[
       let run main =
-        Handler.using
-          Picos_threaded.handler
-          (Picos_threaded.create ~forbid:false (Computation.create ()))
-          main
+        Picos_threaded.run ~forbid:false main
     ]}
 
-    using {{!Picos_threaded.handler} the basic thread based implementation} that
-    comes with Picos for running the example code snippets in this
-    documentation.
+    using {{!Picos_threaded} the basic thread based scheduler}
+    and on OCaml 5
+
+    {@ocaml version>=5.0.0[
+      let run main =
+        Picos_fifos.run ~forbid:false main
+    ]}
+
+    using {{!Picos_fifos} the basic effects based scheduler} that come with
+    Picos as samples.
 
     {3 Auxiliary modules} *)
 
@@ -689,6 +694,10 @@ module Computation : sig
   (** [is_running computation] determines whether the computation is in the
       running state meaning that it has not yet been completed. *)
 
+  val is_canceled : 'a t -> bool
+  (** [is_canceled computation] determines whether the computation is in the
+      canceled state. *)
+
   val canceled : 'a t -> Exn_bt.t option
   (** [canceled computation] returns the exception that the computation has been
       canceled with or returns [None] in case the computation has not been
@@ -899,6 +908,20 @@ module Fiber : sig
 
       ⚠️ It is only safe to call [permit] from the fiber itself. *)
 
+  val is_canceled : t -> bool
+  (** [is_canceled fiber] is equivalent to
+      {@ocaml skip[
+        not (Fiber.has_forbidden fiber) &&
+        let (Packed computation) =
+          Fiber.computation fiber
+        in
+        Computation.is_canceled computation
+      ]}
+
+      ℹ️ This is mostly useful in the effect handlers of schedulers.
+
+      ⚠️ It is only safe to call [is_canceled] from the fiber itself. *)
+
   val canceled : t -> Exn_bt.t option
   (** [canceled fiber] is equivalent to:
       {@ocaml skip[
@@ -1016,7 +1039,10 @@ module Fiber : sig
   (** [create ~forbid computation] creates a new fiber. *)
 
   include
-    Intf.Fiber with type t := t with type 'a computation := 'a Computation.t
+    Intf.Fiber
+      with type t := t
+      with type 'a computation := 'a Computation.t
+      with type exn_bt := Exn_bt.t
 
   (** {2 Design rationale}
 
diff --git a/lib/picos_fifos/dune b/lib/picos_fifos/dune
new file mode 100644
index 0000000..aaa7b52
--- /dev/null
+++ b/lib/picos_fifos/dune
@@ -0,0 +1,6 @@
+(library
+ (name picos_fifos)
+ (public_name picos.fifos)
+ (enabled_if
+  (>= %{ocaml_version} 5.0.0))
+ (libraries picos picos.ptmc picos.select picos_mpsc_queue))
diff --git a/lib/picos_fifos/picos_fifos.ml b/lib/picos_fifos/picos_fifos.ml
new file mode 100644
index 0000000..fd96100
--- /dev/null
+++ b/lib/picos_fifos/picos_fifos.ml
@@ -0,0 +1,208 @@
+open Picos
+module Queue = Picos_mpsc_queue
+
+(* As a minor optimization, we avoid allocating closures, which take slightly
+   more memory than values of this type. *)
+type ready =
+  | Spawn of Fiber.t * (unit -> unit)
+  | Continue of Fiber.t * (unit, unit) Effect.Deep.continuation
+  | Resume of Fiber.t * (Exn_bt.t option, unit) Effect.Deep.continuation
+  | Resume_forbidden of (Exn_bt.t option, unit) Effect.Deep.continuation
+
+type t = {
+  ready : ready Queue.t;
+  needs_wakeup : bool Atomic.t;
+  num_alive_fibers : int Atomic.t;
+  mc : Picos_ptmc.t;
+      (** We must store this for [resume], which may come from any thread. *)
+  resume :
+    Trigger.t ->
+    Fiber.t ->
+    (Exn_bt.t option, unit) Effect.Deep.continuation ->
+    unit;
+}
+
+let rec next t =
+  match Queue.pop_exn t.ready with
+  | Spawn (fiber, main) ->
+      let current = Some (fun k -> Fiber.continue fiber k fiber)
+      and yield =
+        Some
+          (fun k ->
+            Queue.push t.ready (Continue (fiber, k));
+            next t)
+      in
+      let[@alert "-handler"] effc (type a) :
+          a Effect.t -> ((a, _) Effect.Deep.continuation -> _) option = function
+        | Fiber.Current ->
+            (* We handle [Current] first as it is perhaps the most latency
+               sensitive effect. *)
+            current
+        | Fiber.Spawn r ->
+            Some
+              (fun k ->
+                (* We check cancelation status once and then either perform the
+                   whole operation or discontinue the fiber. *)
+                match Fiber.canceled fiber with
+                | None ->
+                    r.mains
+                    |> List.iter (fun main ->
+                           let fiber =
+                             Fiber.create ~forbid:r.forbid r.computation
+                           in
+                           Queue.push t.ready (Spawn (fiber, main));
+                           Atomic.incr t.num_alive_fibers);
+                    (* We intentionally do not check cancelation status
+                       again. *)
+                    Effect.Deep.continue k ()
+                | Some exn_bt -> Exn_bt.discontinue k exn_bt)
+        | Fiber.Yield -> yield
+        | Computation.Cancel_after r ->
+            Some
+              (fun k ->
+                (* We check cancelation status once and then either perform the
+                   whole operation or discontinue the fiber. *)
+                match Fiber.canceled fiber with
+                | None ->
+                    Picos_select.cancel_after r.computation ~seconds:r.seconds
+                      r.exn_bt;
+                    (* We intentionally do not check cancelation status
+                       again. *)
+                    Effect.Deep.continue k ()
+                | Some exn_bt -> Exn_bt.discontinue k exn_bt)
+        | Trigger.Await trigger ->
+            (* We handle [Await] last as it is probably the least latency
+               sensitive effect.  It could also be that another fiber running in
+               parallel is just about to signal the trigger, so checking the
+               trigger last gives a tiny bit of time for that to happen and
+               potentially allows us to make better/different decisions here. *)
+            Some
+              (fun k ->
+                (* The non-blocking logic below for suspending a fiber with
+                   support for parallelism safe cancelation is somewhat
+                   intricate.  Hopefully the comments help to understand it. *)
+                if Fiber.has_forbidden fiber then begin
+                  (* Fiber has forbidden propagation of cancelation.  This is
+                     the easy case to handle. *)
+                  if Trigger.on_signal trigger fiber k t.resume then begin
+                    (* Fiber is now suspended and can be resumed through the
+                       trigger.  We just continue the next ready fiber. *)
+                    next t
+                  end
+                  else begin
+                    (* The trigger was already signaled.  We could now freely
+                       choose which fiber to continue here, but in this
+                       scheduler we choose to continue the current fiber. *)
+                    Effect.Deep.continue k None
+                  end
+                end
+                else begin
+                  (* Fiber permits propagation of cancelation.  We support
+                     cancelation and so first try to attach the trigger to the
+                     computation of the fiber. *)
+                  if Fiber.try_attach fiber trigger then begin
+                    (* The trigger was successfully attached, which means the
+                       computation has not been canceled. *)
+                    if Trigger.on_signal trigger fiber k t.resume then begin
+                      (* Fiber is now suspended and can be resumed through the
+                         trigger.  That can now happen by signaling the trigger
+                         directly or by canceling the computation of the fiber,
+                         which will also signal the trigger.  We just continue
+                         the next ready fiber. *)
+                      next t
+                    end
+                    else begin
+                      (* The trigger was already signaled.  We first need to
+                         ensure that the trigger is detached from the
+                         computation of the fiber. *)
+                      Fiber.detach fiber trigger;
+                      (* We could now freely decide which fiber to continue, but
+                         in this scheduler we choose to continue the current
+                         fiber. *)
+                      Fiber.resume fiber k
+                    end
+                  end
+                  else begin
+                    (* We could not attach the trigger to the computation of the
+                       fiber, which means that either the computation has been
+                       canceled or the trigger has been signaled.  We still need
+                       to ensure that the trigger really is put into the
+                       signaled state before the fiber is continued. *)
+                    Trigger.dispose trigger;
+                    (* We could now freely decide which fiber to continue, but
+                       in this scheduler we choose to continue the current
+                       fiber. *)
+                    Fiber.resume fiber k
+                  end
+                end)
+        | _ -> None
+      and retc () =
+        Atomic.decr t.num_alive_fibers;
+        next t
+      in
+      Effect.Deep.match_with main () { retc; exnc = raise; effc }
+  | Continue (fiber, k) -> Fiber.continue fiber k ()
+  | Resume (fiber, k) -> Fiber.resume fiber k
+  | Resume_forbidden k -> Effect.Deep.continue k None
+  | exception Queue.Empty ->
+      if Atomic.get t.num_alive_fibers <> 0 then begin
+        if Atomic.get t.needs_wakeup then begin
+          Picos_ptmc.lock t.mc;
+          match if Atomic.get t.needs_wakeup then Picos_ptmc.wait t.mc with
+          | () -> Picos_ptmc.unlock t.mc
+          | exception exn ->
+              Picos_ptmc.unlock t.mc;
+              raise exn
+        end
+        else Atomic.set t.needs_wakeup true;
+        next t
+      end
+
+let run ~forbid main =
+  let ready = Queue.create ()
+  and needs_wakeup = Atomic.make false
+  and num_alive_fibers = Atomic.make 1
+  and mc = Picos_ptmc.get () in
+  let rec t = { ready; needs_wakeup; num_alive_fibers; mc; resume }
+  and resume trigger fiber k =
+    begin
+      if Fiber.has_forbidden fiber then
+        (* Fiber has forbidden propagation of cancelation.  This is the easy
+           case. *)
+        Queue.push t.ready (Resume_forbidden k)
+      else
+        let resume = Resume (fiber, k) in
+        if Fiber.is_canceled fiber then begin
+          (* The fiber has been canceled so we give priority to it in this
+             scheduler.
+
+             Assuming fibers are written to cooperate and perform cleanup
+             promptly, this can be advantageous as it allows resources to be
+             released more quickly.  However, malicious or buggy fibers could
+             use this to prevent other fibers from running. *)
+          Queue.push_head t.ready resume
+        end
+        else begin
+          (* The fiber hasn't yet been canceled.
+
+             As propagation of cancelation was not forbidden, and we have
+             attached a trigger, we need to ensure that the trigger will not be
+             leaked. *)
+          Fiber.detach fiber trigger;
+          Queue.push t.ready resume
+        end
+    end;
+    (* As the trigger might have been signaled from another domain or systhread
+       outside of the scheduler, we check whether the scheduler needs to be
+       woken up and take care of it if necessary. *)
+    if
+      Atomic.get t.needs_wakeup
+      && Atomic.compare_and_set t.needs_wakeup true false
+    then Picos_ptmc.broadcast t.mc
+  in
+  let computation = Computation.create () in
+  let fiber = Fiber.create ~forbid computation in
+  let main = Computation.capture computation main in
+  Queue.push t.ready (Spawn (fiber, main));
+  next t;
+  Computation.await computation
diff --git a/lib/picos_fifos/picos_fifos.mli b/lib/picos_fifos/picos_fifos.mli
new file mode 100644
index 0000000..248e73e
--- /dev/null
+++ b/lib/picos_fifos/picos_fifos.mli
@@ -0,0 +1,23 @@
+(** A basic single-threaded effects based {!Picos} compatible scheduler for
+    OCaml 5.
+
+    ℹ️ This scheduler implementation is mostly meant as an example and for use in
+    testing libraries implemented in {!Picos}.
+
+    Briefly:
+
+    - [current] returns the current fiber.
+    - [spawn] forks a new deep effect handler for each fiber.
+    - [yield] pushes the current fiber to the back of the internal queue and
+      takes the next fiber to run from the head of the internal queue.
+    - [cancel_after] uses a per-domain {{!Picos_select} background thread} that
+      runs a [Unix.select] loop to cancel computations.
+    - [await] stores the current fiber to be resumed through the trigger and
+      takes next fiber to run from the head of the internal queue.
+
+    This scheduler also gives priority to fibers woken up from [await] due to
+    being canceled. *)
+
+val run : forbid:bool -> (unit -> 'a) -> 'a
+(** [run ~forbid main] runs the [main] thunk with the scheduler.  Returns after
+    [main] and all of the fibers spawned by [main] have returned. *)
diff --git a/lib/picos_mpsc_queue/dune b/lib/picos_mpsc_queue/dune
new file mode 100644
index 0000000..36f41b5
--- /dev/null
+++ b/lib/picos_mpsc_queue/dune
@@ -0,0 +1,4 @@
+(library
+ (name picos_mpsc_queue)
+ (public_name picos.mpsc_queue)
+ (libraries backoff multicore-magic))
diff --git a/lib/picos_mpsc_queue/picos_mpsc_queue.ml b/lib/picos_mpsc_queue/picos_mpsc_queue.ml
new file mode 100644
index 0000000..1794e07
--- /dev/null
+++ b/lib/picos_mpsc_queue/picos_mpsc_queue.ml
@@ -0,0 +1,104 @@
+type 'a t = { tail : 'a tail Atomic.t; head : 'a head Atomic.t }
+
+and ('a, _) tdt =
+  | Head : ('a, [> `Head ]) tdt
+  | Cons : { value : 'a; mutable next : 'a head } -> ('a, [> `Cons ]) tdt
+  | Tail : ('a, [> `Tail ]) tdt
+  | Snoc : { mutable prev : 'a tail; value : 'a } -> ('a, [> `Snoc ]) tdt
+
+and 'a head = H : ('a, [< `Head | `Cons ]) tdt -> 'a head [@@unboxed]
+and 'a tail = T : ('a, [< `Tail | `Snoc ]) tdt -> 'a tail [@@unboxed]
+
+exception Empty
+
+let[@inline never] impossible () =
+  invalid_arg "Picos_mpsc_queue: multiple consumers not allowed"
+
+let create () =
+  let tail = Multicore_magic.copy_as_padded @@ Atomic.make (T Tail) in
+  let head = Multicore_magic.copy_as_padded @@ Atomic.make (H Head) in
+  Multicore_magic.copy_as_padded { tail; head }
+
+let rec push_head head (Cons r as after : (_, [< `Cons ]) tdt) backoff =
+  let before = Atomic.get head in
+  r.next <- before;
+  if not (Atomic.compare_and_set head before (H after)) then
+    push_head head after (Backoff.once backoff)
+
+let push_head t value =
+  let head = t.head in
+  let before = Atomic.get head in
+  let after = Cons { value; next = before } in
+  if not (Atomic.compare_and_set head before (H after)) then
+    push_head head after Backoff.default
+
+let rec append_to (Cons cons_r : (_, [< `Cons ]) tdt) tail =
+  match cons_r.next with
+  | H Head -> cons_r.next <- tail
+  | H (Cons _ as head) -> append_to head tail
+
+let rec push tail (Snoc r as after : (_, [< `Snoc ]) tdt) backoff =
+  let before = Atomic.get tail in
+  r.prev <- before;
+  if not (Atomic.compare_and_set tail before (T after)) then
+    push tail after (Backoff.once backoff)
+
+let push t value =
+  let tail = t.tail in
+  let before = Atomic.get tail in
+  let after = Snoc { prev = before; value } in
+  if not (Atomic.compare_and_set tail before (T after)) then
+    push tail after Backoff.default
+
+let rec rev_to head (Snoc r : (_, [< `Snoc ]) tdt) =
+  let head = Cons { value = r.value; next = H head } in
+  match r.prev with T Tail -> head | T (Snoc _ as prev) -> rev_to head prev
+
+let rec pop_exn t backoff = function
+  | H (Cons head_r as head) ->
+      if Atomic.compare_and_set t.head (H head) head_r.next then head_r.value
+      else
+        let backoff = Backoff.once backoff in
+        pop_exn t backoff (Atomic.get t.head)
+  | H Head ->
+      if Atomic.get t.tail != T Tail then
+        match Atomic.exchange t.tail (T Tail) with
+        | T Tail -> impossible ()
+        | T (Snoc snoc_r) -> begin
+            match snoc_r.prev with
+            | T Tail -> begin
+                match Atomic.get t.head with
+                | H Head -> snoc_r.value
+                | H (Cons _ as head) ->
+                    let next = Cons { value = snoc_r.value; next = H Head } in
+                    append_to head (H next);
+                    pop_head_exn t backoff head
+              end
+            | T (Snoc _ as prev) ->
+                let next = Cons { value = snoc_r.value; next = H Head } in
+                let (Cons cons_r as next : (_, [< `Cons ]) tdt) =
+                  rev_to next prev
+                in
+                if Atomic.compare_and_set t.head (H Head) cons_r.next then
+                  cons_r.value
+                else begin
+                  match Atomic.get t.head with
+                  | H Head -> impossible ()
+                  | H (Cons _ as head) ->
+                      append_to head (H next);
+                      pop_head_exn t backoff head
+                end
+          end
+      else begin
+        match Atomic.get t.head with
+        | H Head -> raise_notrace Empty
+        | H (Cons _ as head) -> pop_head_exn t backoff head
+      end
+
+and pop_head_exn t backoff (Cons head_r as head : (_, [< `Cons ]) tdt) =
+  if Atomic.compare_and_set t.head (H head) head_r.next then head_r.value
+  else
+    let backoff = Backoff.once backoff in
+    pop_exn t backoff (Atomic.get t.head)
+
+let[@inline] pop_exn t = pop_exn t Backoff.default (Atomic.get t.head)
diff --git a/lib/picos_mpsc_queue/picos_mpsc_queue.mli b/lib/picos_mpsc_queue/picos_mpsc_queue.mli
new file mode 100644
index 0000000..42ac636
--- /dev/null
+++ b/lib/picos_mpsc_queue/picos_mpsc_queue.mli
@@ -0,0 +1,30 @@
+(** A multi-producer, single-consumer queue.
+
+    🏎️ This data structure is optimized for use as a scheduler's ready queue. *)
+
+type !'a t
+(** A multi-producer, single-consumer queue. *)
+
+val create : unit -> 'a t
+(** [create ()] returns a new empty multi-producer, single-consumer queue. *)
+
+(** {2 Interface for producers} *)
+
+val push : 'a t -> 'a -> unit
+(** [push queue value] adds the [value] to the tail of the [queue]. *)
+
+val push_head : 'a t -> 'a -> unit
+(** [push_head queue value] adds the [value] to the head of the [queue]. *)
+
+(** {2 Interface for the owner / consumer} *)
+
+exception Empty
+(** Raised by {!pop_exn} in case it finds the queue empty. *)
+
+val pop_exn : 'a t -> 'a
+(** [pop_exn queue] tries to remove the value at the head of the [queue].
+    Returns the removed value or raises {!Empty} in case the queue was empty.
+
+    ⚠️ This should only be called by the owner / consumer of the queue.
+
+    @raise Empty in case the queue was empty. *)
diff --git a/lib/picos_threaded/picos_threaded.ml b/lib/picos_threaded/picos_threaded.ml
index f27739c..46bee28 100644
--- a/lib/picos_threaded/picos_threaded.ml
+++ b/lib/picos_threaded/picos_threaded.ml
@@ -1,9 +1,5 @@
 open Picos
 
-type t = Fiber.t
-
-let create = Fiber.create
-
 let block trigger ptmc =
   (* We block fibers (or threads) on a per thread mutex and condition. *)
   Picos_ptmc.lock ptmc;
@@ -84,14 +80,14 @@ and yield fiber =
   Fiber.check fiber;
   Systhreads.yield ()
 
-and cancel_after : type a. t -> a Computation.t -> _ =
+and cancel_after : type a. _ -> a Computation.t -> _ =
  (* We need an explicit type signature to allow OCaml to generalize the tyoe as
     all of the handlers are in a single recursive definition. *)
  fun fiber computation ~seconds exn_bt ->
   Fiber.check fiber;
   Select.cancel_after computation ~seconds exn_bt
 
-and spawn : type a. t -> forbid:bool -> a Computation.t -> _ =
+and spawn : type a. _ -> forbid:bool -> a Computation.t -> _ =
  fun fiber ~forbid computation mains ->
   Fiber.check fiber;
   mains
@@ -100,8 +96,11 @@ and spawn : type a. t -> forbid:bool -> a Computation.t -> _ =
        (fun () ->
          (* We need to (recursively) install the handler on each new thread
             that we create. *)
-         Handler.using handler (create ~forbid computation) main)
+         Handler.using handler (Fiber.create ~forbid computation) main)
        ()
      |> ignore
 
 and handler = Handler.{ current; spawn; yield; cancel_after; await }
+
+let run ~forbid main =
+  Handler.using handler (Fiber.create ~forbid (Computation.create ())) main
diff --git a/lib/picos_threaded/picos_threaded.mli b/lib/picos_threaded/picos_threaded.mli
index 6cc8503..79e180e 100644
--- a/lib/picos_threaded/picos_threaded.mli
+++ b/lib/picos_threaded/picos_threaded.mli
@@ -1,37 +1,27 @@
-(** A basic [Thread] based implementation of the effects based operations of
-    {!Picos} for OCaml 4.
+(** A basic [Thread] based {!Picos} compatible scheduler for OCaml 4.
+
+    ℹ️ This scheduler implementation is mostly meant as an example and for use in
+    testing libraries implemented in {!Picos}.
 
     Briefly:
 
     - [current] returns the current fiber from the per thread state.
     - [spawn] creates a new [Thread] for each fiber.
     - [yield] just calls [Thread.yield].
-    - [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.
+    - [cancel_after] uses a per-domain {{!Picos_select} background thread} that
+      runs a [Unix.select] loop to cancel computations.
     - [await] uses {{!Picos_ptmc} a per thread mutex and condition} to suspend
       the thread.
 
-    ⚠️ This scheduler implementation is mostly meant as an example and is
-    probably suitable for simple applications that do not spawn a lot of fibers.
-    If an application uses a lot of short lived fibers, then a more
-    sophisticated scheduler implementation using some sort of thread pool will
-    likely perform significantly better.
+    ⚠️ This scheduler is probably suitable for simple applications that do not
+    spawn a lot of fibers.  If an application uses a lot of short lived fibers,
+    then a more sophisticated scheduler implementation using some sort of thread
+    pool will likely perform significantly better.
 
     ⚠️ This scheduler implementation also works on OCaml 5, of course, but on
     OCaml 5 a scheduler that implements an effect handler directly is likely to
     perform better. *)
 
-open Picos
-
-type t
-(** The per thread state for a {!handler}. *)
-
-val create : forbid:bool -> 'a Computation.t -> t
-(** [create ~forbid computation] creates a new per thread state for a fiber with
-    the given [computation] and propagation of cancelation [forbid]den as
-    given. *)
-
-val handler : t Handler.t
-(** A basic handler of the effects based operations of {!Picos} using [Thread]s
-    and a [Unix.select] based IO event loop. *)
+val run : forbid:bool -> (unit -> 'a) -> 'a
+(** [run ~forbid main] runs the [main] thunk with the scheduler.  Returns after
+    [main] and all of the fibers spawned by [main] have returned. *)
diff --git a/picos.opam b/picos.opam
index 59bcd91..11a6b2c 100644
--- a/picos.opam
+++ b/picos.opam
@@ -12,11 +12,12 @@ depends: [
   "thread-local-storage" {>= "0.1"}
   "mtime" {>= "2.0.0"}
   "psq" {>= "0.2.1"}
-  "multicore-magic" {>= "2.1.0" & with-test}
+  "multicore-magic" {>= "2.1.0"}
   "multicore-bench" {>= "0.1.2" & with-test}
   "alcotest" {>= "1.7.0" & with-test}
   "qcheck-core" {>= "0.21.2" & with-test}
   "qcheck-stm" {>= "0.3" & with-test}
+  "qcheck-multicoretests-util" {>= "0.3" & with-test}
   "mdx" {>= "2.4.0" & with-test}
   "ocaml-version" {>= "3.6.4" & with-test}
   "domain_shims" {>= "0.1.0" & with-test}


Test patch:
diff --git a/test/dune b/test/dune
index a1b8830..146546a 100644
--- a/test/dune
+++ b/test/dune
@@ -6,7 +6,7 @@
   (select
    test_scheduler.ml
    from
-   (picos.select schedulers -> test_scheduler.ocaml5.ml)
+   (picos.fifos -> test_scheduler.ocaml5.ml)
    (picos.threaded -> test_scheduler.ocaml4.ml))))
 
 ;;
@@ -53,13 +53,11 @@
  (package picos)
  (modules test_mpsc_queue)
  (libraries
-  foundation
+  picos_mpsc_queue
   qcheck-core
-  qcheck-core.runner
+  qcheck-multicoretests-util
   qcheck-stm.stm
-  qcheck-stm.sequential)
- (action
-  (run %{test} --verbose)))
+  stm_run))
 
 ;;
 
diff --git a/test/lib/foundation/mpsc_queue.ml b/test/lib/foundation/mpsc_queue.ml
deleted file mode 100644
index bbf1e74..0000000
--- a/test/lib/foundation/mpsc_queue.ml
+++ /dev/null
@@ -1,58 +0,0 @@
-type 'a t = { tail : 'a tail Atomic.t; head : 'a head ref }
-
-and ('a, _) tdt =
-  | Head : ('a, [> `Head ]) tdt
-  | Cons : { value : 'a; next : 'a head } -> ('a, [> `Cons ]) tdt
-  | Tail : ('a, [> `Tail ]) tdt
-  | Snoc : { mutable prev : 'a tail; value : 'a } -> ('a, [> `Snoc ]) tdt
-
-and 'a head = H : ('a, [< `Head | `Cons ]) tdt -> 'a head [@@unboxed]
-and 'a tail = T : ('a, [< `Tail | `Snoc ]) tdt -> 'a tail [@@unboxed]
-
-let create () =
-  let tail = Multicore_magic.copy_as_padded @@ Atomic.make (T Tail) in
-  let head = Multicore_magic.copy_as_padded @@ ref (H Head) in
-  Multicore_magic.copy_as_padded { tail; head }
-
-let rec enqueue tail (Snoc r as after : (_, [< `Snoc ]) tdt) backoff =
-  let before = Atomic.get tail in
-  r.prev <- before;
-  if not (Atomic.compare_and_set tail before (T after)) then
-    enqueue tail after (Backoff.once backoff)
-
-let enqueue t value =
-  let tail = t.tail in
-  let before = Atomic.get tail in
-  let after = Snoc { prev = before; value } in
-  if not (Atomic.compare_and_set tail before (T after)) then
-    enqueue tail after Backoff.default
-
-exception Empty
-
-let rec rev_to head (Snoc r : (_, [< `Snoc ]) tdt) =
-  let head = Cons { value = r.value; next = H head } in
-  match r.prev with T Tail -> head | T (Snoc _ as prev) -> rev_to head prev
-
-let dequeue t =
-  match !(t.head) with
-  | H (Cons r) ->
-      t.head := r.next;
-      r.value
-  | H Head ->
-      if Atomic.get t.tail == T Tail then raise_notrace Empty
-      else begin
-        match Atomic.exchange t.tail (T Tail) with
-        | T Tail ->
-            (* Impossible *)
-            raise_notrace Empty
-        | T (Snoc r) -> begin
-            match r.prev with
-            | T Tail -> r.value
-            | T (Snoc _ as prev) ->
-                let (Cons r : (_, [< `Cons ]) tdt) =
-                  rev_to (Cons { value = r.value; next = H Head }) prev
-                in
-                t.head := r.next;
-                r.value
-          end
-      end
diff --git a/test/lib/foundation/mpsc_queue.mli b/test/lib/foundation/mpsc_queue.mli
deleted file mode 100644
index 0ae7310..0000000
--- a/test/lib/foundation/mpsc_queue.mli
+++ /dev/null
@@ -1,16 +0,0 @@
-(** A multi-producer, single-consumer queue *)
-
-type !'a t
-(** *)
-
-val create : unit -> 'a t
-(** *)
-
-val enqueue : 'a t -> 'a -> unit
-(** *)
-
-exception Empty
-(** *)
-
-val dequeue : 'a t -> 'a
-(** *)
diff --git a/test/lib/schedulers/dune b/test/lib/schedulers/dune
deleted file mode 100644
index c8a2448..0000000
--- a/test/lib/schedulers/dune
+++ /dev/null
@@ -1,5 +0,0 @@
-(library
- (name schedulers)
- (enabled_if
-  (>= %{ocaml_version} 5.0.0))
- (libraries picos picos.ptmc foundation))
diff --git a/test/lib/schedulers/fifos.ml b/test/lib/schedulers/fifos.ml
deleted file mode 100644
index b2ec83b..0000000
--- a/test/lib/schedulers/fifos.ml
+++ /dev/null
@@ -1,163 +0,0 @@
-open Picos
-open Foundation
-
-type t = {
-  ready : (unit -> unit) Mpsc_queue.t;
-  needs_wakeup : bool Atomic.t;
-  num_alive_fibers : int ref;
-  mc : Picos_ptmc.t;
-  on_unhandled : unit Try_handle.t option;
-}
-
-let rec next t =
-  match Mpsc_queue.dequeue t.ready with
-  | work -> work ()
-  | exception Mpsc_queue.Empty ->
-      if !(t.num_alive_fibers) <> 0 then begin
-        if Atomic.get t.needs_wakeup then begin
-          Picos_ptmc.lock t.mc;
-          match if Atomic.get t.needs_wakeup then Picos_ptmc.wait t.mc with
-          | () -> Picos_ptmc.unlock t.mc
-          | exception exn ->
-              Picos_ptmc.unlock t.mc;
-              raise exn
-        end
-        else Atomic.set t.needs_wakeup true;
-        next t
-      end
-
-let run ?on_unhandled ~forbid main =
-  let t =
-    let ready = Mpsc_queue.create () in
-    let needs_wakeup = Atomic.make false in
-    let num_alive_fibers = ref 1 in
-    let mc = Picos_ptmc.get () in
-    { ready; needs_wakeup; num_alive_fibers; mc; on_unhandled }
-  in
-  let resume trigger fiber k =
-    if not (Fiber.has_forbidden fiber) then begin
-      (* As propagation of cancelation was not forbidden, and we have attached a
-         trigger, we need to ensure that the trigger will not be leaked. *)
-      Fiber.detach fiber trigger
-    end;
-    let work () =
-      (* In this scheduler we determine the cancelation status just before we
-         continue the resumed fiber after dequeuing it from the ready queue. *)
-      let exn_bt_opt = Fiber.canceled fiber in
-      Effect.Deep.continue k exn_bt_opt
-    in
-    (* In this scheduler we don't schedule canceled fibers any differently, but
-       we could have, at this point, checked if the fiber actually has been
-       canceled and enqueue the fiber differently. *)
-    Mpsc_queue.enqueue t.ready work;
-    (* As the trigger might have been signaled from another domain or systhread
-       outside of the scheduler, we check whether the scheduler needs to be
-       woken up and take care of it if necessary. *)
-    if
-      Atomic.get t.needs_wakeup
-      && Atomic.compare_and_set t.needs_wakeup true false
-    then Picos_ptmc.broadcast t.mc
-  in
-  let rec fork fiber main =
-    let current = Some (fun k -> Fiber.continue fiber k fiber)
-    and yield =
-      Some
-        (fun k ->
-          Mpsc_queue.enqueue t.ready (Fiber.continue fiber k);
-          next t)
-    in
-    let[@alert "-handler"] effc (type a) :
-        a Effect.t -> ((a, _) Effect.Deep.continuation -> _) option = function
-      | Fiber.Current ->
-          (* We handle [Current] first as it is perhaps the most latency
-             sensitive effect. *)
-          current
-      | Fiber.Spawn spawn ->
-          Some
-            (fun k ->
-              match Fiber.canceled fiber with
-              | None ->
-                  spawn.mains
-                  |> List.iter (fun main ->
-                         let fiber =
-                           Fiber.create ~forbid:spawn.forbid spawn.computation
-                         in
-                         Mpsc_queue.enqueue t.ready (fun () -> fork fiber main);
-                         incr t.num_alive_fibers);
-                  Effect.Deep.continue k ()
-              | Some exn_bt -> Exn_bt.discontinue k exn_bt)
-      | Fiber.Yield -> yield
-      | Trigger.Await trigger ->
-          (* We handle [Await] last as it is probably the least latency
-             sensitive effect.  It could also be that another fiber running in
-             parallel is just about to signal the trigger, so checking the
-             trigger last gives a tiny bit of time for that to happen and
-             potentially allows us to make better/different decisions here. *)
-          Some
-            (fun k ->
-              (* The non-blocking logic below for suspending a fiber with
-                 support for parallelism safe cancelation is somewhat intricate.
-                 Hopefully the comments help to understand it. *)
-              if Fiber.has_forbidden fiber then begin
-                (* Fiber has forbidden propagation of cancelation.  This is the
-                   easy case to handle. *)
-                if Trigger.on_signal trigger fiber k resume then begin
-                  (* Fiber is now suspended and can be resumed through the
-                     trigger.  We just continue the next ready fiber. *)
-                  next t
-                end
-                else begin
-                  (* The trigger was already signaled.  We could now freely
-                     choose which fiber to continue here, but in this scheduler
-                     we choose to continue the current fiber. *)
-                  Effect.Deep.continue k None
-                end
-              end
-              else begin
-                (* Fiber permits propagation of cancelation.  We support
-                   cancelation and so first try to attach the trigger to the
-                   computation of the fiber. *)
-                if Fiber.try_attach fiber trigger then begin
-                  (* The trigger was successfully attached, which means the
-                     computation has not been canceled. *)
-                  if Trigger.on_signal trigger fiber k resume then begin
-                    (* Fiber is now suspended and can be resumed through the
-                       trigger.  That can now happen by signaling the trigger
-                       directly or by canceling the computation of the fiber,
-                       which will also signal the trigger.  We just continue the
-                       next ready fiber. *)
-                    next t
-                  end
-                  else begin
-                    (* The trigger was already signaled.  We first need to
-                       ensure that the trigger is detached from the computation
-                       of the fiber. *)
-                    Fiber.detach fiber trigger;
-                    (* We could now freely decide which fiber to continue, but
-                       in this scheduler we choose to continue the current
-                       fiber. *)
-                    Effect.Deep.continue k (Fiber.canceled fiber)
-                  end
-                end
-                else begin
-                  (* We could not attach the trigger to the computation of the
-                     fiber, which means that either the computation has been
-                     canceled or the trigger has been signaled.  We still need
-                     to ensure that the trigger really is put into the signaled
-                     state before the fiber is continued. *)
-                  Trigger.dispose trigger;
-                  (* We could now freely decide which fiber to continue, but in
-                     this scheduler we choose to continue the current fiber. *)
-                  Effect.Deep.continue k (Fiber.canceled fiber)
-                end
-              end)
-      | e -> Try_handle.call t.on_unhandled e
-    and retc () =
-      decr t.num_alive_fibers;
-      next t
-    in
-    Effect.Deep.match_with main () { retc; exnc = raise; effc }
-  in
-  let computation = Computation.create () in
-  fork (Fiber.create ~forbid computation) (Computation.capture computation main);
-  Computation.await computation
diff --git a/test/lib/schedulers/fifos.mli b/test/lib/schedulers/fifos.mli
deleted file mode 100644
index a2b3ec5..0000000
--- a/test/lib/schedulers/fifos.mli
+++ /dev/null
@@ -1,4 +0,0 @@
-(** A simple FIFO scheduler for Picos *)
-
-val run : ?on_unhandled:unit Try_handle.t -> forbid:bool -> (unit -> 'a) -> 'a
-(** *)
diff --git a/test/lib/schedulers/try_handle.ml b/test/lib/schedulers/try_handle.ml
deleted file mode 100644
index bafc7dc..0000000
--- a/test/lib/schedulers/try_handle.ml
+++ /dev/null
@@ -1,7 +0,0 @@
-type 'b t = {
-  try_handle :
-    'a. 'a Effect.t -> (('a, 'b) Effect.Deep.continuation -> 'b) option;
-}
-[@@unboxed]
-
-let call t e = match t with None -> None | Some r -> r.try_handle e
diff --git a/test/lib/schedulers/try_handle.mli b/test/lib/schedulers/try_handle.mli
deleted file mode 100644
index 2530b76..0000000
--- a/test/lib/schedulers/try_handle.mli
+++ /dev/null
@@ -1,8 +0,0 @@
-type 'b t = {
-  try_handle :
-    'a. 'a Effect.t -> (('a, 'b) Effect.Deep.continuation -> 'b) option;
-}
-[@@unboxed]
-
-val call :
-  'b t option -> 'a Effect.t -> (('a, 'b) Effect.Deep.continuation -> 'b) option
diff --git a/test/stm_run/dune b/test/stm_run/dune
new file mode 100644
index 0000000..fb6ced2
--- /dev/null
+++ b/test/stm_run/dune
@@ -0,0 +1,24 @@
+(rule
+ (enabled_if %{lib-available:qcheck-stm.domain})
+ (action
+  (copy stm_run.ocaml5.ml stm_run.ml)))
+
+(rule
+ (enabled_if
+  (not %{lib-available:qcheck-stm.domain}))
+ (action
+  (copy stm_run.ocaml4.ml stm_run.ml)))
+
+(library
+ (name stm_run)
+ (libraries
+  qcheck-core
+  qcheck-core.runner
+  qcheck-stm.stm
+  qcheck-stm.sequential
+  qcheck-stm.thread
+  (select
+   empty.ml
+   from
+   (qcheck-stm.domain -> empty.ocaml5.ml)
+   (-> empty.ocaml4.ml))))
diff --git a/test/stm_run/empty.ocaml4.ml b/test/stm_run/empty.ocaml4.ml
new file mode 100644
index 0000000..e69de29
diff --git a/test/stm_run/empty.ocaml5.ml b/test/stm_run/empty.ocaml5.ml
new file mode 100644
index 0000000..e69de29
diff --git a/test/stm_run/intf.ml b/test/stm_run/intf.ml
new file mode 100644
index 0000000..fd300e7
--- /dev/null
+++ b/test/stm_run/intf.ml
@@ -0,0 +1,45 @@
+module type STM_domain = sig
+  module Spec : STM.Spec
+
+  val check_obs :
+    (Spec.cmd * STM.res) list ->
+    (Spec.cmd * STM.res) list ->
+    (Spec.cmd * STM.res) list ->
+    Spec.state ->
+    bool
+
+  val all_interleavings_ok :
+    Spec.cmd list * Spec.cmd list * Spec.cmd list -> bool
+
+  val arb_cmds_triple :
+    int ->
+    int ->
+    (Spec.cmd list * Spec.cmd list * Spec.cmd list) QCheck.arbitrary
+
+  val arb_triple :
+    int ->
+    int ->
+    (Spec.state -> Spec.cmd QCheck.arbitrary) ->
+    (Spec.state -> Spec.cmd QCheck.arbitrary) ->
+    (Spec.state -> Spec.cmd QCheck.arbitrary) ->
+    (Spec.cmd list * Spec.cmd list * Spec.cmd list) QCheck.arbitrary
+
+  val arb_triple_asym :
+    int ->
+    int ->
+    (Spec.state -> Spec.cmd QCheck.arbitrary) ->
+    (Spec.state -> Spec.cmd QCheck.arbitrary) ->
+    (Spec.state -> Spec.cmd QCheck.arbitrary) ->
+    (Spec.cmd list * Spec.cmd list * Spec.cmd list) QCheck.arbitrary
+
+  val interp_sut_res : Spec.sut -> Spec.cmd list -> (Spec.cmd * STM.res) list
+  val agree_prop_par : Spec.cmd list * Spec.cmd list * Spec.cmd list -> bool
+
+  val agree_prop_par_asym :
+    Spec.cmd list * Spec.cmd list * Spec.cmd list -> bool
+
+  val agree_test_par : count:int -> name:string -> QCheck.Test.t
+  val neg_agree_test_par : count:int -> name:string -> QCheck.Test.t
+  val agree_test_par_asym : count:int -> name:string -> QCheck.Test.t
+  val neg_agree_test_par_asym : count:int -> name:string -> QCheck.Test.t
+end
diff --git a/test/stm_run/stm_run.ocaml4.ml b/test/stm_run/stm_run.ocaml4.ml
new file mode 100644
index 0000000..43af6ff
--- /dev/null
+++ b/test/stm_run/stm_run.ocaml4.ml
@@ -0,0 +1,13 @@
+include Intf
+
+let run ~verbose ~count ~name ?make_domain (module Spec : STM.Spec) =
+  let module Seq = STM_sequential.Make (Spec) in
+  let module Con = STM_thread.Make (Spec) [@alert "-experimental"] in
+  [
+    [ Seq.agree_test ~count ~name:(name ^ " sequential") ];
+    (match make_domain with
+    | None -> [ Con.agree_test_conc ~count ~name:(name ^ " concurrent") ]
+    | Some _ -> []);
+  ]
+  |> List.concat
+  |> QCheck_base_runner.run_tests ~verbose
diff --git a/test/stm_run/stm_run.ocaml5.ml b/test/stm_run/stm_run.ocaml5.ml
new file mode 100644
index 0000000..85b0f09
--- /dev/null
+++ b/test/stm_run/stm_run.ocaml5.ml
@@ -0,0 +1,25 @@
+include Intf
+
+let run (type cmd state sut) ~verbose ~count ~name ?make_domain
+    (module Spec : STM.Spec
+      with type cmd = cmd
+       and type state = state
+       and type sut = sut) =
+  let module Seq = STM_sequential.Make (Spec) in
+  let module Dom = struct
+    module Spec = Spec
+    include STM_domain.Make (Spec)
+  end in
+  [
+    [ Seq.agree_test ~count ~name:(name ^ " sequential") ];
+    (match make_domain with
+    | None -> [ Dom.agree_test_par ~count ~name:(name ^ " parallel") ]
+    | Some make_domain ->
+        make_domain ~count ~name
+          (module Dom : STM_domain
+            with type Spec.cmd = cmd
+             and type Spec.state = state
+             and type Spec.sut = sut));
+  ]
+  |> List.concat
+  |> QCheck_base_runner.run_tests ~verbose
diff --git a/test/test_mpsc_queue.ml b/test/test_mpsc_queue.ml
index 0fe1921..130a646 100644
--- a/test/test_mpsc_queue.ml
+++ b/test/test_mpsc_queue.ml
@@ -1,53 +1,87 @@
-open Foundation
+module Queue = Picos_mpsc_queue
 
 module Spec = struct
-  type cmd = Enq of int | Deq
+  type cmd = Push of int | Push_head of int | Pop
 
   let show_cmd c =
-    match c with Enq i -> "Enq " ^ string_of_int i | Deq -> "Deq"
+    match c with
+    | Push i -> "Push " ^ string_of_int i
+    | Push_head i -> "Push_head " ^ string_of_int i
+    | Pop -> "Pop"
 
   type state = int list
-  type sut = int Mpsc_queue.t
+  type sut = int Queue.t
+
+  let producer_cmd _s =
+    QCheck.(
+      make ~print:show_cmd
+        (Gen.oneof
+           [
+             Gen.map (fun i -> Push i) Gen.nat;
+             Gen.map (fun i -> Push_head i) Gen.nat;
+           ]))
 
   let arb_cmd _s =
     QCheck.(
       make ~print:show_cmd
-        (Gen.oneof [ Gen.map (fun i -> Enq i) Gen.nat; Gen.return Deq ]))
+        (Gen.oneof
+           [
+             Gen.map (fun i -> Push i) Gen.nat;
+             Gen.map (fun i -> Push_head i) Gen.nat;
+             Gen.return Pop;
+           ]))
 
   let init_state = []
-  let init_sut () = Mpsc_queue.create ()
+  let init_sut () = Queue.create ()
   let cleanup _ = ()
 
   let next_state c s =
     match c with
-    | Enq i -> s @ [ i ]
-    | Deq -> ( match s with _ :: s -> s | [] -> [])
+    | Push i -> s @ [ i ]
+    | Push_head i -> i :: s
+    | Pop -> ( match s with _ :: s -> s | [] -> [])
 
   let precond _ _ = true
 
   let run c d =
     let open STM in
     match c with
-    | Enq i -> Res (unit, Mpsc_queue.enqueue d i)
-    | Deq ->
+    | Push i -> Res (unit, Queue.push d i)
+    | Push_head i -> Res (unit, Queue.push_head d i)
+    | Pop ->
         Res
           ( option int,
-            match Mpsc_queue.dequeue d with
+            match Queue.pop_exn d with
             | i -> Some i
-            | exception Mpsc_queue.Empty -> None )
+            | exception Queue.Empty -> None )
 
   let postcond c (s : state) res =
     let open STM in
     match (c, res) with
-    | Enq _, Res ((Unit, _), ()) -> true
-    | Deq, Res ((Option Int, _), res) ->
+    | Push _, Res ((Unit, _), ()) -> true
+    | Push_head _, Res ((Unit, _), ()) -> true
+    | Pop, Res ((Option Int, _), res) ->
         (match s with [] -> None | x :: _ -> Some x) = res
     | _, _ -> false
 end
 
-module Seq = STM_sequential.Make (Spec)
-
 let () =
-  let count = 1000 in
-  QCheck_base_runner.run_tests_main
-    [ Seq.agree_test ~count ~name:"STM Mpsc_queue test sequential" ]
+  let make_domain ~count ~name
+      (module Dom : Stm_run.STM_domain
+        with type Spec.cmd = Spec.cmd
+         and type Spec.state = Spec.state
+         and type Spec.sut = Spec.sut) =
+    let arb_cmds_par =
+      Dom.arb_triple 20 12 Spec.arb_cmd Spec.arb_cmd Spec.producer_cmd
+    in
+    let agree_test_par_asym ~count ~name =
+      let rep_count = 50 in
+      QCheck.Test.make ~retries:10 ~count ~name arb_cmds_par @@ fun triple ->
+      QCheck.assume (Dom.all_interleavings_ok triple);
+      Util.repeat rep_count Dom.agree_prop_par_asym triple
+    in
+    [ agree_test_par_asym ~count ~name:(name ^ " parallel") ]
+  in
+  Stm_run.run ~count:1000 ~name:"Picos_mpsc_queue" ~verbose:true ~make_domain
+    (module Spec)
+  |> exit
diff --git a/test/test_scheduler.ocaml4.ml b/test/test_scheduler.ocaml4.ml
index bf43bbb..f741a3e 100644
--- a/test/test_scheduler.ocaml4.ml
+++ b/test/test_scheduler.ocaml4.ml
@@ -1,6 +1,1 @@
-open Picos
-
-let run main =
-  Handler.using Picos_threaded.handler
-    (Picos_threaded.create ~forbid:false (Computation.create ()))
-    main
+let run main = Picos_threaded.run ~forbid:false main
diff --git a/test/test_scheduler.ocaml5.ml b/test/test_scheduler.ocaml5.ml
index 3b4fa89..bed2960 100644
--- a/test/test_scheduler.ocaml5.ml
+++ b/test/test_scheduler.ocaml5.ml
@@ -1,14 +1,1 @@
-let[@alert "-handler"] on_unhandled =
-  let try_handle :
-      type a. a Effect.t -> ((a, _) Effect.Deep.continuation -> _) option =
-    function
-    | Picos.Computation.Cancel_after r ->
-        Some
-          (fun k ->
-            Picos_select.cancel_after r.computation ~seconds:r.seconds r.exn_bt;
-            Effect.Deep.continue k ())
-    | _ -> None
-  in
-  Schedulers.Try_handle.{ try_handle }
-
-let run main = Schedulers.Fifos.run ~on_unhandled ~forbid:false main
+let run main = Picos_fifos.run ~forbid:false main