Add Queue, Stack, and Mvar benchmarks

Makefile

@@ -1,10 +1,15 @@
-all: build test
+.PHONY: build test bench
+
+all: build test bench
 
 build:
 	@dune build @install
 
 test:
 	@dune runtest --force
 
+bench:
+	@dune exec --release -- bench/main.exe 10
+
 clean:
 	@dune clean

bench/barrier.ml

@@ -0,0 +1,11 @@
+type t = { counter : int Atomic.t; total : int }
+
+let make total = { counter = Atomic.make 0; total }
+
+let await { counter; total } =
+  if Atomic.get counter = total then
+    Atomic.compare_and_set counter total 0 |> ignore;
+  Atomic.incr counter;
+  while Atomic.get counter < total do
+    Domain.cpu_relax ()
+  done

bench/barrier.mli

@@ -0,0 +1,4 @@
+type t
+
+val make : int -> t
+val await : t -> unit

bench/bench.ml

@@ -0,0 +1,126 @@
+module Times = struct
+  type t = { inverted : bool; times : float array }
+
+  let record ~n_domains ?(n_warmups = 3) ?(n_runs = 17) ?(before = Fun.id) ~init
+      ~work ?(after = Fun.id) () =
+    let barrier_before = Barrier.make n_domains in
+    let barrier_after = Barrier.make n_domains in
+    let results = Array.init n_domains @@ fun _ -> Stack.create () in
+    let main domain_i =
+      for _ = 1 to n_warmups do
+        if domain_i = 0 then before ();
+        let state = init domain_i in
+        Barrier.await barrier_before;
+        work domain_i state;
+        Barrier.await barrier_after;
+        if domain_i = 0 then after ()
+      done;
+      for _run_i = 0 to n_runs - 1 do
+        if domain_i = 0 then before ();
+        let state = init domain_i in
+        Barrier.await barrier_before;
+        let start = Mtime_clock.elapsed () in
+        work domain_i state;
+        let stop = Mtime_clock.elapsed () in
+        Barrier.await barrier_after;
+        if domain_i = 0 then after ();
+        Stack.push
+          (Mtime.Span.to_float_ns (Mtime.Span.abs_diff stop start)
+          *. (1. /. 1_000_000_000.0))
+          results.(domain_i)
+      done
+    in
+    let prepare_for_await () =
+      let open struct
+        type state = Init | Released | Awaiting of { mutable released : bool }
+      end in
+      let state = Atomic.make Init in
+      let release () =
+        if Multicore_magic.fenceless_get state != Released then
+          match Atomic.exchange state Released with
+          | Awaiting r -> r.released <- true
+          | _ -> ()
+      in
+      let await () =
+        if Multicore_magic.fenceless_get state != Released then
+          let awaiting = Awaiting { released = false } in
+          if Atomic.compare_and_set state Init awaiting then
+            match awaiting with
+            | Awaiting r ->
+                (* Avoid sleeping *)
+                while not r.released do
+                  Domain.cpu_relax ()
+                done
+            | _ -> ()
+      in
+      Domain_local_await.{ release; await }
+    in
+    Domain_local_await.using ~prepare_for_await ~while_running:(fun () ->
+        let domains =
+          Array.init n_domains @@ fun domain_i ->
+          Domain.spawn @@ fun () -> main domain_i
+        in
+        Array.iter Domain.join domains);
+    let n = Stack.length results.(0) in
+    let times = Array.create_float n in
+    for run_i = 0 to n - 1 do
+      times.(run_i) <- 0.0;
+      for domain_i = 0 to n_domains - 1 do
+        times.(run_i) <- times.(run_i) +. Stack.pop results.(domain_i)
+      done
+    done;
+    { inverted = false; times }
+
+  let invert t =
+    { inverted = not t.inverted; times = Array.map (fun v -> 1.0 /. v) t.times }
+end
+
+module Stats = struct
+  type t = { mean : float; median : float; sd : float; inverted : bool }
+
+  let scale factor t =
+    {
+      mean = t.mean *. factor;
+      median = t.median *. factor;
+      sd = t.sd *. factor;
+      inverted = t.inverted;
+    }
+
+  let mean_of times =
+    Array.fold_left ( +. ) 0.0 times /. Float.of_int (Array.length times)
+
+  let sd_of times mean =
+    Float.sqrt
+      (mean_of (Array.map (fun v -> Float.abs (v -. mean) ** 2.) times))
+
+  let median_of times =
+    Array.sort Float.compare times;
+    let n = Array.length times in
+    if n land 1 = 0 then (times.((n asr 1) - 1) +. times.(n asr 1)) /. 2.0
+    else times.(n asr 1)
+
+  let of_times (t : Times.t) =
+    let mean = mean_of t.times in
+    let sd = sd_of t.times mean in
+    let median = median_of t.times in
+    { mean; sd; median; inverted = t.inverted }
+
+  let to_nonbreaking s =
+    s |> String.split_on_char ' '
+    |> String.concat " " (* a non-breaking space *)
+
+  let to_json ~name ~description ~units t =
+    let metric value =
+      `Assoc
+        [
+          ("name", `String (to_nonbreaking name));
+          ("value", `Float value);
+          ("units", `String units);
+          ( "trend",
+            `String
+              (if t.inverted then "higher-is-better" else "lower-is-better") );
+          ("description", `String description);
+        ]
+    in
+    [ metric t.median ]
+end

bench/bench.mli

@@ -0,0 +1,26 @@
+module Times : sig
+  type t
+
+  val record :
+    n_domains:int ->
+    ?n_warmups:int ->
+    ?n_runs:int ->
+    ?before:(unit -> unit) ->
+    init:(int -> 's) ->
+    work:(int -> 's -> unit) ->
+    ?after:(unit -> unit) ->
+    unit ->
+    t
+
+  val invert : t -> t
+end
+
+module Stats : sig
+  type t
+
+  val of_times : Times.t -> t
+  val scale : float -> t -> t
+
+  val to_json :
+    name:string -> description:string -> units:string -> t -> Yojson.Safe.t list
+end

bench/bench_hashtbl.ml

@@ -0,0 +1,79 @@
+open Kcas_data
+open Bench
+
+module Int = struct
+  include Int
+
+  let hash = Fun.id
+end
+
+let run_one ~n_domains ?(factor = 1) ?(n_ops = 50 * factor * Util.iter_factor)
+    ?(n_keys = 1000) ~percent_read () =
+  let t =
+    Hashtbl.create ~hashed_type:(module Int) ()
+    |> Multicore_magic.copy_as_padded
+  in
+
+  for i = 0 to n_keys - 1 do
+    Hashtbl.replace t i i
+  done;
+
+  let n_ops_todo = Atomic.make n_ops |> Multicore_magic.copy_as_padded in
+
+  let init _ = Random.State.make_self_init () in
+
+  let work _ state =
+    let rec work () =
+      let n = Util.alloc n_ops_todo in
+      if n <> 0 then
+        let rec loop n =
+          if 0 < n then
+            let value = Random.State.bits state in
+            let op = (value asr 20) mod 100 in
+            let key = value mod n_keys in
+            if op < percent_read then begin
+              Hashtbl.find_opt t key |> ignore;
+              loop (n - 1)
+            end
+            else begin
+              Hashtbl.remove t key;
+              Hashtbl.add t key value;
+              loop (n - 2)
+            end
+          else work ()
+        in
+        loop n
+    in
+    work ()
+  in
+  let after () = Atomic.set n_ops_todo n_ops in
+
+  let times = Times.record ~n_domains ~init ~work ~after () in
+
+  let name metric =
+    Printf.sprintf "%s/%d worker%s, %d%% reads" metric n_domains
+      (if n_domains = 1 then "" else "s")
+      percent_read
+  in
+
+  List.concat
+    [
+      Stats.of_times times
+      |> Stats.scale (1_000_000_000.0 /. Float.of_int n_ops)
+      |> Stats.to_json
+           ~name:(name "time per operation")
+           ~description:"Average time to find, remove, or add a binding"
+           ~units:"ns";
+      Times.invert times |> Stats.of_times
+      |> Stats.scale (Float.of_int (n_ops * n_domains) /. 1_000_000.0)
+      |> Stats.to_json
+           ~name:(name "operations over time")
+           ~description:
+             "Number of operations performed over time using all domains"
+           ~units:"M/s";
+    ]
+
+let run_suite ~factor =
+  Util.cross [ 90; 50; 10 ] [ 1; 2; 4 ]
+  |> List.concat_map @@ fun (percent_read, n_domains) ->
+     run_one ~n_domains ~percent_read ~factor ()

bench/bench_mvar.ml

@@ -0,0 +1,108 @@
+open Kcas_data
+open Bench
+
+let run_one ?(n_adders = 2) ?(blocking_add = false) ?(n_takers = 2)
+    ?(blocking_take = false) ?(factor = 1)
+    ?(n_msgs = 2 * factor * Util.iter_factor) () =
+  let n_domains = n_adders + n_takers in
+
+  let t = Mvar.create None in
+
+  let n_msgs_to_take = Atomic.make n_msgs |> Multicore_magic.copy_as_padded in
+  let n_msgs_to_add = Atomic.make n_msgs |> Multicore_magic.copy_as_padded in
+
+  let init _ = () in
+  let work i () =
+    if i < n_adders then
+      if blocking_add then
+        let rec work () =
+          let n = Util.alloc n_msgs_to_add in
+          if 0 < n then begin
+            for i = 1 to n do
+              Mvar.put t i
+            done;
+            work ()
+          end
+        in
+        work ()
+      else
+        let rec work () =
+          let n = Util.alloc n_msgs_to_add in
+          if 0 < n then begin
+            for i = 1 to n do
+              while not (Mvar.try_put t i) do
+                Domain.cpu_relax ()
+              done
+            done;
+            work ()
+          end
+        in
+        work ()
+    else if blocking_take then
+      let rec work () =
+        let n = Util.alloc n_msgs_to_take in
+        if n <> 0 then begin
+          for _ = 1 to n do
+            ignore (Mvar.take t)
+          done;
+          work ()
+        end
+      in
+      work ()
+    else
+      let rec work () =
+        let n = Util.alloc n_msgs_to_take in
+        if n <> 0 then begin
+          for _ = 1 to n do
+            while Option.is_none (Mvar.take_opt t) do
+              Domain.cpu_relax ()
+            done
+          done;
+          work ()
+        end
+      in
+      work ()
+  in
+  let after () =
+    Atomic.set n_msgs_to_take n_msgs;
+    Atomic.set n_msgs_to_add n_msgs
+  in
+
+  let times = Times.record ~n_domains ~init ~work ~after () in
+
+  let name metric =
+    let format role blocking n =
+      Printf.sprintf "%d %s%s%s" n
+        (if blocking then "" else "nb ")
+        role
+        (if n = 1 then "" else "s")
+    in
+    Printf.sprintf "%s/%s, %s" metric
+      (format "adder" blocking_add n_adders)
+      (format "taker" blocking_take n_takers)
+  in
+
+  List.concat
+    [
+      Stats.of_times times
+      |> Stats.scale (1_000_000_000.0 /. Float.of_int n_msgs)
+      |> Stats.to_json ~name:(name "time per message")
+           ~description:
+             "Time to transmit one message from one domain to another"
+           ~units:"ns";
+      Times.invert times |> Stats.of_times
+      |> Stats.scale (Float.of_int (n_msgs * n_domains) /. 1_000_000.0)
+      |> Stats.to_json
+           ~name:(name "messages over time")
+           ~description:
+             "Number of messages transmitted over time using all domains"
+           ~units:"M/s";
+    ]
+
+let run_suite ~factor =
+  Util.cross
+    (Util.cross [ 1; 2 ] [ false; true ])
+    (Util.cross [ 1; 2 ] [ false; true ])
+  |> List.concat_map
+     @@ fun ((n_adders, blocking_add), (n_takers, blocking_take)) ->
+     run_one ~n_adders ~blocking_add ~n_takers ~blocking_take ~factor ()