Repository: ocaml-multicore/picos
Pull request #172 (2024-07-10 13:29:21)

Use `copy_as ?padded` from `multicore-magic`


Patch:
diff --git a/bench/bench_mpscq.ml b/bench/bench_mpscq.ml
index 9167d5a..158c29d 100644
--- a/bench/bench_mpscq.ml
+++ b/bench/bench_mpscq.ml
@@ -1,17 +1,21 @@
 open Multicore_bench
-module Queue = Picos_mpscq
 
 let run_one_domain ~budgetf ?(n_msgs = 50 * Util.iter_factor) () =
-  let t = Queue.create () in
+  let t = Picos_mpscq.create ~padded:true () in
 
   let op push =
-    if push then Queue.push t 101
-    else match Queue.pop_exn t with _ -> () | exception Queue.Empty -> ()
+    if push then Picos_mpscq.push t 101
+    else
+      match Picos_mpscq.pop_exn t with
+      | _ -> ()
+      | exception Picos_mpscq.Empty -> ()
   in
 
   let init _ =
     assert (
-      match Queue.pop_exn t with _ -> false | exception Queue.Empty -> true);
+      match Picos_mpscq.pop_exn t with
+      | _ -> false
+      | exception Picos_mpscq.Empty -> true);
     Util.generate_push_and_pop_sequence n_msgs
   in
   let work _ bits = Util.Bits.iter op bits in
@@ -25,13 +29,15 @@ let run_one ~budgetf ~n_adders () =
 
   let n_msgs = 200 / n_takers * Util.iter_factor in
 
-  let t = Queue.create () in
+  let t = Picos_mpscq.create ~padded:true () in
 
   let n_msgs_to_add = Atomic.make 0 |> Multicore_magic.copy_as_padded in
 
   let init _ =
     assert (
-      match Queue.pop_exn t with _ -> false | exception Queue.Empty -> true);
+      match Picos_mpscq.pop_exn t with
+      | _ -> false
+      | exception Picos_mpscq.Empty -> true);
     Atomic.set n_msgs_to_add n_msgs
   in
   let work i () =
@@ -40,7 +46,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.push t i
+            Picos_mpscq.push t i
           done;
           work ()
         end
@@ -49,9 +55,9 @@ let run_one ~budgetf ~n_adders () =
     else
       let rec loop n =
         if 0 < n then
-          match Queue.pop_exn t with
+          match Picos_mpscq.pop_exn t with
           | _ -> loop (n - 1)
-          | exception Queue.Empty ->
+          | exception Picos_mpscq.Empty ->
               Backoff.once Backoff.default |> ignore;
               loop n
       in
diff --git a/dune-project b/dune-project
index e3da720..1eb8068 100644
--- a/dune-project
+++ b/dune-project
@@ -36,7 +36,7 @@
    (>= 0.2.1))
   ;; For mpsc queue
   (multicore-magic
-   (>= 2.2.0))
+   (>= 2.3.0))
   ;; For picos.lwt
   (lwt
    (>= 5.7.0))
diff --git a/lib/picos_fifos/dune b/lib/picos_fifos/dune
index 765b2a4..b03fff0 100644
--- a/lib/picos_fifos/dune
+++ b/lib/picos_fifos/dune
@@ -10,4 +10,5 @@
    from
    (picos.select -> select.some.ml)
    (-> select.none.ml))
+  multicore-magic
   picos.mpscq))
diff --git a/lib/picos_fifos/picos_fifos.ml b/lib/picos_fifos/picos_fifos.ml
index 54e5636..eca46f4 100644
--- a/lib/picos_fifos/picos_fifos.ml
+++ b/lib/picos_fifos/picos_fifos.ml
@@ -1,5 +1,4 @@
 open Picos
-module Queue = Picos_mpscq
 
 (* As a minor optimization, we avoid allocating closures, which take slightly
    more memory than values of this type. *)
@@ -9,7 +8,7 @@ type ready =
   | Resume of Fiber.t * (Exn_bt.t option, unit) Effect.Deep.continuation
 
 type t = {
-  ready : ready Queue.t;
+  ready : ready Picos_mpscq.t;
   needs_wakeup : bool Atomic.t;
   num_alive_fibers : int Atomic.t;
   mutex : Mutex.t;
@@ -26,13 +25,13 @@ let rec spawn t n forbid packed = function
   | [] -> Atomic.fetch_and_add t.num_alive_fibers n |> ignore
   | main :: mains ->
       let fiber = Fiber.create_packed ~forbid packed in
-      Queue.push t.ready (Spawn (fiber, main));
+      Picos_mpscq.push t.ready (Spawn (fiber, main));
       spawn t (n + 1) forbid packed mains
 
 let continue = Some (fun k -> Effect.Deep.continue k ())
 
 let rec next t =
-  match Queue.pop_exn t.ready with
+  match Picos_mpscq.pop_exn t.ready with
   | Spawn (fiber, main) ->
       let current =
         (* The current handler must never propagate cancelation, but it would be
@@ -42,7 +41,7 @@ let rec next t =
       and yield =
         Some
           (fun k ->
-            Queue.push t.ready (Continue (fiber, k));
+            Picos_mpscq.push t.ready (Continue (fiber, k));
             next t)
       and discontinue = Some (fun k -> Fiber.continue fiber k ()) in
       let[@alert "-handler"] effc (type a) :
@@ -83,7 +82,7 @@ let rec next t =
       Effect.Deep.match_with main () { retc = t.retc; exnc = raise; effc }
   | Continue (fiber, k) -> Fiber.continue fiber k ()
   | Resume (fiber, k) -> Fiber.resume fiber k
-  | exception Queue.Empty ->
+  | exception Picos_mpscq.Empty ->
       if Atomic.get t.num_alive_fibers <> 0 then begin
         if Atomic.get t.needs_wakeup then begin
           Mutex.lock t.mutex;
@@ -105,9 +104,9 @@ let rec next t =
 
 let run ?(forbid = false) main =
   Select.check_configured ();
-  let ready = Queue.create ()
-  and needs_wakeup = Atomic.make false
-  and num_alive_fibers = Atomic.make 1
+  let ready = Picos_mpscq.create ~padded:true ()
+  and needs_wakeup = Atomic.make false |> Multicore_magic.copy_as_padded
+  and num_alive_fibers = Atomic.make 1 |> Multicore_magic.copy_as_padded
   and mutex = Mutex.create ()
   and condition = Condition.create () in
   let rec t =
@@ -119,11 +118,11 @@ let run ?(forbid = false) main =
     let resume = Resume (fiber, k) in
     if Fiber.unsuspend fiber trigger then
       (* The fiber has not been canceled, so we queue the fiber normally. *)
-      Queue.push t.ready resume
+      Picos_mpscq.push t.ready resume
     else
       (* The fiber has been canceled, so we give priority to it in this
          scheduler. *)
-      Queue.push_head t.ready resume;
+      Picos_mpscq.push_head t.ready resume;
     (* 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. *)
@@ -147,6 +146,6 @@ let run ?(forbid = false) main =
   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));
+  Picos_mpscq.push t.ready (Spawn (fiber, main));
   next t;
   Computation.await computation
diff --git a/lib/picos_lwt_unix/picos_lwt_unix.ml b/lib/picos_lwt_unix/picos_lwt_unix.ml
index 4bb106d..e32aa13 100644
--- a/lib/picos_lwt_unix/picos_lwt_unix.ml
+++ b/lib/picos_lwt_unix/picos_lwt_unix.ml
@@ -1,7 +1,7 @@
 let[@inline never] not_main_thread () =
   invalid_arg "not called from the main thread"
 
-let ready = Picos_mpscq.create ()
+let ready = Picos_mpscq.create ~padded:true ()
 
 type notification = { mutable ref_count : int; mutable id : int }
 
diff --git a/lib/picos_mpscq/picos_mpscq.ml b/lib/picos_mpscq/picos_mpscq.ml
index 7ec7e50..4e8ecef 100644
--- a/lib/picos_mpscq/picos_mpscq.ml
+++ b/lib/picos_mpscq/picos_mpscq.ml
@@ -13,10 +13,10 @@ exception Empty
 
 let[@inline never] impossible () = invalid_arg "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 create ?padded () =
+  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
diff --git a/lib/picos_mpscq/picos_mpscq.mli b/lib/picos_mpscq/picos_mpscq.mli
index 6039cef..abfec6c 100644
--- a/lib/picos_mpscq/picos_mpscq.mli
+++ b/lib/picos_mpscq/picos_mpscq.mli
@@ -8,7 +8,7 @@
 type !'a t
 (** A multi-producer, single-consumer queue. *)
 
-val create : unit -> 'a t
+val create : ?padded:bool -> unit -> 'a t
 (** [create ()] returns a new empty multi-producer, single-consumer queue. *)
 
 (** {2 Interface for producers}
diff --git a/lib/picos_sync/condition.ml b/lib/picos_sync/condition.ml
index 730ea26..1cfcaed 100644
--- a/lib/picos_sync/condition.ml
+++ b/lib/picos_sync/condition.ml
@@ -3,9 +3,7 @@ open Picos
 type state = Empty | Queue of { head : Trigger.t list; tail : Trigger.t list }
 type t = state Atomic.t
 
-let create ?(padded = false) () =
-  let t = Atomic.make Empty in
-  if padded then Multicore_magic.copy_as_padded t else t
+let create ?padded () = Multicore_magic.copy_as ?padded @@ Atomic.make Empty
 
 let broadcast t =
   if Atomic.get t != Empty then
diff --git a/lib/picos_sync/latch.ml b/lib/picos_sync/latch.ml
index 22ea60a..208ed47 100644
--- a/lib/picos_sync/latch.ml
+++ b/lib/picos_sync/latch.ml
@@ -7,13 +7,14 @@ type t = { count : int Atomic.t; computation : unit Computation.t }
 
 let zero = Atomic.make 0
 
-let create n =
+let create ?padded n =
   if n < 0 then negative_count ();
-  let count = if 0 < n then Atomic.make n else zero
+  let count =
+    if 0 < n then Atomic.make n |> Multicore_magic.copy_as ?padded else zero
   and computation =
     if n <= 0 then Computation.finished else Computation.create ()
   in
-  { count; computation }
+  Multicore_magic.copy_as ?padded { count; computation }
 
 type _ result =
   | Unit : unit result  (** [Unit] is physically same as [()]. *)
diff --git a/lib/picos_sync/mutex.ml b/lib/picos_sync/mutex.ml
index 345e131..14b928d 100644
--- a/lib/picos_sync/mutex.ml
+++ b/lib/picos_sync/mutex.ml
@@ -12,9 +12,7 @@ type state =
 
 type t = state Atomic.t
 
-let create ?(padded = false) () =
-  let t = Atomic.make Unlocked in
-  if padded then Multicore_magic.copy_as_padded t else t
+let create ?padded () = Multicore_magic.copy_as ?padded @@ Atomic.make Unlocked
 
 (* We try to avoid starvation of unlock by making it so that when, at the start
    of lock or unlock, the head is empty, the tail is reversed into the head.
diff --git a/lib/picos_sync/picos_sync.mli b/lib/picos_sync/picos_sync.mli
index 8bcf6bb..2898746 100644
--- a/lib/picos_sync/picos_sync.mli
+++ b/lib/picos_sync/picos_sync.mli
@@ -280,7 +280,7 @@ module Latch : sig
   type t
   (** Represents a dynamic countdown latch. *)
 
-  val create : int -> t
+  val create : ?padded:bool -> int -> t
   (** [create initial] creates a new countdown latch with the specified
       [initial] count.
 
@@ -445,9 +445,10 @@ end
 (** {1 Conventions}
 
     The optional [padded] argument taken by several constructor functions, e.g.
-    {!Mutex.create} and {!Condition.create}, defaults to [false].  When
-    explicitly specified as [~padded:true] the object is allocated in a way to
-    avoid {{:https://en.wikipedia.org/wiki/False_sharing} false sharing}.  For
-    relatively long lived objects this can improve performance and make
-    performance more stable at the cost of using more memory.  It is not
-    recommended to use [~padded:true] for short lived objects. *)
+    {!Latch.create}, {!Mutex.create}, and {!Condition.create}, defaults to
+    [false].  When explicitly specified as [~padded:true] the object is
+    allocated in a way to avoid {{:https://en.wikipedia.org/wiki/False_sharing}
+    false sharing}.  For relatively long lived objects this can improve
+    performance and make performance more stable at the cost of using more
+    memory.  It is not recommended to use [~padded:true] for short lived
+    objects. *)
diff --git a/picos.opam b/picos.opam
index c72b998..5282719 100644
--- a/picos.opam
+++ b/picos.opam
@@ -14,7 +14,7 @@ depends: [
   "thread-local-storage" {>= "0.1"}
   "mtime" {>= "2.0.0"}
   "psq" {>= "0.2.1"}
-  "multicore-magic" {>= "2.2.0"}
+  "multicore-magic" {>= "2.3.0"}
   "lwt" {>= "5.7.0"}
   "multicore-bench" {>= "0.1.4" & with-test}
   "alcotest" {>= "1.7.0" & with-test}


Test patch:
diff --git a/test/test_stdio.ml b/test/test_stdio.ml
index 3da015d..3e1e2b4 100644
--- a/test/test_stdio.ml
+++ b/test/test_stdio.ml
@@ -77,7 +77,7 @@ let test_select () =
   Unix.set_nonblock syn_inn;
   Unix.set_nonblock syn_out;
 
-  let events = Picos_mpscq.create () in
+  let events = Picos_mpscq.create ~padded:true () in
 
   Bundle.join_after @@ fun bundle ->
   begin
diff --git a/test/test_structured.ml b/test/test_structured.ml
index 322fb17..7b9c882 100644
--- a/test/test_structured.ml
+++ b/test/test_structured.ml
@@ -182,7 +182,7 @@ let test_any_and_all_errors () =
   [ Run.all; Run.any ]
   |> List.iter @@ fun run_op ->
      Test_scheduler.run ~max_domains:6 @@ fun () ->
-     let raised = Picos_mpscq.create () in
+     let raised = Picos_mpscq.create ~padded:true () in
      let raiser exn () =
        Picos_mpscq.push raised exn;
        raise exn