Rewrite or refactor benchmarks using multicore-bench

Author: polytypic

Makefile

@@ -10,4 +10,4 @@ clean:
 	dune clean
 
 bench:
-	@dune exec --release -- ./bench/main.exe
+	@dune exec --release -- ./bench/main.exe -budget 1

bench/README.md

@@ -4,10 +4,4 @@ Benchmarks for Saturn
 
 Execute `make bench` from root of the repository to run the standard set of
 benchmarks. The output is in JSON, as it is intended to be consumed by
-ocaml-benchmark CI (in progress).
-
-# Specific structures
-
-Some benchmarks expose commandline interface targeting particular structures:
-
-- [mpmc_queue.exe](mpmc_queue_cmd.ml)
+[current-bench](https://bench.ci.dev/ocaml-multicore/saturn/branch/main/benchmark/default).

bench/bench_queue.ml

@@ -0,0 +1,58 @@
+open Multicore_bench
+module Queue = Saturn_lockfree.Queue
+
+let run_one ~budgetf ?(n_adders = 2) ?(n_takers = 2)
+    ?(n_msgs = 50 * Util.iter_factor) () =
+  let n_domains = n_adders + n_takers in
+
+  let t = Queue.create () in
+
+  let n_msgs_to_take = Atomic.make 0 |> Multicore_magic.copy_as_padded in
+  let n_msgs_to_add = Atomic.make 0 |> Multicore_magic.copy_as_padded in
+
+  let init _ =
+    Atomic.set n_msgs_to_take n_msgs;
+    Atomic.set n_msgs_to_add n_msgs
+  in
+  let work i () =
+    if i < n_adders then
+      let rec work () =
+        let n = Util.alloc n_msgs_to_add in
+        if 0 < n then begin
+          for i = 1 to n do
+            Queue.push t i
+          done;
+          work ()
+        end
+      in
+      work ()
+    else
+      let rec work () =
+        let n = Util.alloc n_msgs_to_take in
+        if n <> 0 then
+          let rec loop n =
+            if 0 < n then
+              loop (n - Bool.to_int (Option.is_some (Queue.pop_opt t)))
+            else work ()
+          in
+          loop n
+      in
+      work ()
+  in
+
+  let config =
+    let format role n =
+      Printf.sprintf "%d %s%s" n role (if n = 1 then "" else "s")
+    in
+    Printf.sprintf "%s, %s"
+      (format "nb adder" n_adders)
+      (format "nb taker" n_takers)
+  in
+
+  Times.record ~budgetf ~n_domains ~init ~work ()
+  |> Util.thruput_metrics ~n:n_msgs ~singular:"message" ~config
+
+let run_suite ~budgetf =
+  Util.cross [ 1; 2 ] [ 1; 2 ]
+  |> List.concat_map @@ fun (n_adders, n_takers) ->
+     run_one ~budgetf ~n_adders ~n_takers ()

bench/bench_relaxed_queue.ml

@@ -0,0 +1,94 @@
+open Multicore_bench
+module Queue = Saturn.Relaxed_queue
+module Spin = Queue.Spin
+module Not_lockfree = Queue.Not_lockfree
+module CAS_interface = Queue.Not_lockfree.CAS_interface
+
+let run_one ~budgetf ~n_adders ~n_takers ?(n_msgs = 50 * Util.iter_factor)
+    ?(api = `Spin) () =
+  let n_domains = n_adders + n_takers in
+
+  let t = Queue.create ~size_exponent:10 () in
+
+  let n_msgs_to_take = Atomic.make 0 |> Multicore_magic.copy_as_padded in
+  let n_msgs_to_add = Atomic.make 0 |> Multicore_magic.copy_as_padded in
+
+  let init _ =
+    Atomic.set n_msgs_to_take n_msgs;
+    Atomic.set n_msgs_to_add n_msgs
+  in
+  let work i () =
+    if i < n_adders then
+      let rec work () =
+        let n = Util.alloc n_msgs_to_add in
+        if n <> 0 then begin
+          match api with
+          | `Spin ->
+              for i = 1 to n do
+                Spin.push t i
+              done;
+              work ()
+          | `Not_lockfree ->
+              let rec loop n =
+                if 0 < n then loop (n - Bool.to_int (Not_lockfree.push t i))
+                else work ()
+              in
+              loop n
+          | `CAS_interface ->
+              let rec loop n =
+                if 0 < n then loop (n - Bool.to_int (CAS_interface.push t i))
+                else work ()
+              in
+              loop n
+        end
+      in
+      work ()
+    else
+      let rec work () =
+        let n = Util.alloc n_msgs_to_take in
+        if n <> 0 then
+          match api with
+          | `Spin ->
+              for _ = 1 to n do
+                Spin.pop t |> ignore
+              done;
+              work ()
+          | `Not_lockfree ->
+              let rec loop n =
+                if 0 < n then
+                  loop (n - Bool.to_int (Option.is_some (Not_lockfree.pop t)))
+                else work ()
+              in
+              loop n
+          | `CAS_interface ->
+              let rec loop n =
+                if 0 < n then
+                  loop (n - Bool.to_int (Option.is_some (CAS_interface.pop t)))
+                else work ()
+              in
+              loop n
+      in
+      work ()
+  in
+
+  let config =
+    let format role n =
+      Printf.sprintf "%d %s%s" n role (if n = 1 then "" else "s")
+    in
+    Printf.sprintf "%s, %s (%s)" (format "adder" n_adders)
+      (format "taker" n_takers)
+      (match api with
+      | `Spin -> "spin"
+      | `Not_lockfree -> "not lf"
+      | `CAS_interface -> "cas")
+  in
+
+  Times.record ~budgetf ~n_domains ~init ~work ()
+  |> Util.thruput_metrics ~n:n_msgs ~singular:"message" ~config
+
+let run_suite ~budgetf =
+  Util.cross
+    [ `Spin; `Not_lockfree; `CAS_interface ]
+    (Util.cross [ 1; 2 ] [ 1; 2 ])
+  |> List.concat_map @@ fun (api, (n_adders, n_takers)) ->
+     run_one ~budgetf ~n_adders ~n_takers ~api ()

bench/bench_size.ml

@@ -0,0 +1,38 @@
+open Multicore_bench
+module Size = Saturn_lockfree.Size
+
+let run_one ~budgetf ~n_domains ?(n_ops = 250 * n_domains * Util.iter_factor) ()
+    =
+  let t = Size.create () in
+
+  let n_ops_todo = Atomic.make 0 |> Multicore_magic.copy_as_padded in
+
+  let init _ = Atomic.set n_ops_todo n_ops in
+  let work _ () =
+    let rec work () =
+      let n = Util.alloc n_ops_todo in
+      if n <> 0 then
+        let rec loop n =
+          if 0 < n then begin
+            let incr = Size.new_once t Size.incr in
+            Size.update_once t incr;
+            let decr = Size.new_once t Size.decr in
+            Size.update_once t decr;
+            loop (n - 2)
+          end
+          else work ()
+        in
+        loop n
+    in
+    work ()
+  in
+
+  let config =
+    Printf.sprintf "%d worker%s" n_domains (if n_domains = 1 then "" else "s")
+  in
+  Times.record ~budgetf ~n_domains ~init ~work ()
+  |> Util.thruput_metrics ~n:n_ops ~config ~singular:"operation"
+
+let run_suite ~budgetf =
+  [ 1; 2; 4 ]
+  |> List.concat_map @@ fun n_domains -> run_one ~n_domains ~budgetf ()

bench/bench_skiplist.ml

@@ -1,57 +1,70 @@
-open Saturn
-
-let workload num_elems num_threads add remove =
-  let sl = Skiplist.create ~compare:Int.compare () in
-  let elems = Array.init num_elems (fun _ -> Random.int 10000) in
-  let push () =
-    Domain.spawn (fun () ->
-        let start_time = Unix.gettimeofday () in
-        for i = 0 to (num_elems - 1) / num_threads do
-          Domain.cpu_relax ();
-          let prob = Random.float 1.0 in
-          if prob < add then Skiplist.try_add sl (Random.int 10000) () |> ignore
-          else if prob >= add && prob < add +. remove then
-            Skiplist.try_remove sl (Random.int 10000) |> ignore
-          else Skiplist.mem sl elems.(i) |> ignore
-        done;
-        start_time)
+open Multicore_bench
+module Skiplist = Saturn.Skiplist
+
+let run_one ~budgetf ~n_domains ?(n_ops = 20 * Util.iter_factor)
+    ?(n_keys = 10000) ~percent_mem ?(percent_add = (100 - percent_mem + 1) / 2)
+    ?(prepopulate = true) () =
+  let percent_rem = 100 - (percent_mem + percent_add) in
+
+  let limit_mem = percent_mem in
+  let limit_add = percent_mem + percent_add in
+
+  assert (0 <= limit_mem && limit_mem <= 100);
+  assert (limit_mem <= limit_add && limit_add <= 100);
+
+  let t = Skiplist.create ~compare:Int.compare () in
+  if prepopulate then
+    for _ = 1 to n_keys do
+      let value = Random.bits () in
+      let key = value mod n_keys in
+      Skiplist.try_add t key value |> ignore
+    done;
+
+  let n_ops = (100 + percent_mem) * n_ops / 100 in
+  let n_ops = n_ops * n_domains in
+
+  let n_ops_todo = Atomic.make 0 |> Multicore_magic.copy_as_padded in
+
+  let init _ =
+    Atomic.set n_ops_todo n_ops;
+    Random.State.make_self_init ()
   in
-  let threads = List.init num_threads (fun _ -> push ()) in
-  let start_time_threads =
-    List.map (fun domain -> Domain.join domain) threads
+  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 < limit_mem then begin
+              Skiplist.mem t key |> ignore;
+              loop (n - 1)
+            end
+            else if op < limit_add then begin
+              Skiplist.try_add t key value |> ignore;
+              loop (n - 1)
+            end
+            else begin
+              Skiplist.try_remove t key |> ignore;
+              loop (n - 1)
+            end
+          else work ()
+        in
+        loop n
+    in
+    work ()
   in
-  let end_time = Unix.gettimeofday () in
-  let time_diff = end_time -. List.nth start_time_threads 0 in
-  time_diff
-
-(* A write heavy workload with threads with 50% adds and 50% removes. *)
-let write_heavy_workload num_elems num_threads =
-  workload num_elems num_threads 0.5 0.5
-
-(* A regular workload with 90% reads, 9% adds and 1% removes. *)
-let read_heavy_workload num_elems num_threads =
-  workload num_elems num_threads 0.09 0.01
-
-let moderate_heavy_workload num_elems num_threads =
-  workload num_elems num_threads 0.2 0.1
-
-let balanced_heavy_workload num_elems num_threads =
-  workload num_elems num_threads 0.3 0.2
-
-let bench ~workload_type ~num_elems ~num_threads () =
-  let workload =
-    if workload_type = "read_heavy" then read_heavy_workload
-    else if workload_type = "moderate_heavy" then moderate_heavy_workload
-    else if workload_type = "balanced_heavy" then balanced_heavy_workload
-    else write_heavy_workload
+
+  let config =
+    Printf.sprintf "%d workers, %d%% mem %d%% add %d%% rem" n_domains
+      percent_mem percent_add percent_rem
   in
-  let results = ref [] in
-  for i = 1 to 10 do
-    let time = workload num_elems num_threads in
-    if i > 1 then results := time :: !results
-  done;
-  let results = List.sort Float.compare !results in
-  let median_time = List.nth results 4 in
-  let median_throughput = Float.of_int num_elems /. median_time in
-  Benchmark_result.create_generic ~median_time ~median_throughput
-    ("atomic_skiplist_" ^ workload_type)
+  Times.record ~n_domains ~budgetf ~init ~work ()
+  |> Util.thruput_metrics ~n:n_ops ~singular:"operation" ~config
+
+let run_suite ~budgetf =
+  Util.cross [ 10; 50; 90 ] [ 1; 2; 4 ]
+  |> List.concat_map @@ fun (percent_mem, n_domains) ->
+     run_one ~budgetf ~n_domains ~percent_mem ()

bench/bench_spsc_queue.ml

@@ -1,37 +1,30 @@
-module Spsc_queue = Saturn.Single_prod_single_cons_queue
+open Multicore_bench
+module Queue = Saturn.Single_prod_single_cons_queue
 
-let item_count = 2_000_000
+let run_one ~budgetf ?(size_exponent = 3) ?(n_msgs = 80 * Util.iter_factor) () =
+  let queue = Queue.create ~size_exponent in
 
-let rec try_until_success f =
-  try f () with Spsc_queue.Full -> try_until_success f
-
-let run () =
-  let queue = Spsc_queue.create ~size_exponent:3 in
-  let pusher =
-    Domain.spawn (fun () ->
-        let start_time = Unix.gettimeofday () in
-        for i = 1 to item_count do
-          try_until_success (fun () -> Spsc_queue.push queue i)
-        done;
-        start_time)
+  let init _ = () in
+  let work i () =
+    if i = 0 then
+      let rec loop n =
+        if 0 < n then loop (n - Bool.to_int (Queue.try_push queue n))
+      in
+      loop n_msgs
+    else
+      let rec loop n =
+        if 0 < n then
+          match Queue.pop_opt queue with
+          | Some _ -> loop (n - 1)
+          | None -> loop n
+      in
+      loop n_msgs
   in
-  for _ = 1 to item_count do
-    while Option.is_none (Spsc_queue.pop_opt queue) do
-      ()
-    done
-  done;
-  let end_time = Unix.gettimeofday () in
-  let start_time = Domain.join pusher in
-  let time_diff = end_time -. start_time in
-  time_diff
 
-let bench () =
-  let results = ref [] in
-  for i = 1 to 10 do
-    let time = run () in
-    if i > 1 then results := time :: !results
-  done;
-  let results = List.sort Float.compare !results in
-  let median_time = List.nth results 4 in
-  let median_throughput = Float.of_int item_count /. median_time in
-  Benchmark_result.create_generic ~median_time ~median_throughput "spsc-queue"
+  let config = Printf.sprintf "2 workers, capacity %d" (1 lsl size_exponent) in
+  Times.record ~n_domains:2 ~budgetf ~init ~work ()
+  |> Util.thruput_metrics ~n:n_msgs ~singular:"message" ~config
+
+let run_suite ~budgetf =
+  [ 0; 3; 6; 9 ]
+  |> List.concat_map @@ fun size_exponent -> run_one ~budgetf ~size_exponent ()