Merge pull request #21 from bartoszmodelski/mpmc-queue

MPMC relaxed queue

Author: lyrm

Makefile

@@ -10,4 +10,4 @@ clean:
 	dune clean
 
 bench:
-	@dune exec -- ./bench/main.exe --json
+	@dune exec -- ./bench/main.exe

README.md

@@ -7,6 +7,8 @@ A collection of Concurrent Lockfree Data Structures for OCaml 5. It contains:
 
 * [SPSC Queue](src/spsc_queue.mli) Simple single-producer single-consumer fixed-size queue. Thread-safe as long as at most one thread acts as producer and at most one as consumer at any single point in time.
 
+* [MPMC Relaxed Queue](src/mpmc_relaxed_queue.mli) Multi-producer, multi-consumer, fixed-size relaxed queue. Optimised for high number of threads. Not strictly FIFO. Note, it exposes two interfaces: a lockfree and a non-lockfree (albeit more practical) one. See the mli for details. 
+
 * [MPSC Queue](src/mpsc_queue.mli) A multi-producer, single-consumer, thread-safe queue without support for cancellation. This makes a good data structure for a scheduler's run queue. It is used in [Eio](https://github.com/ocaml-multicore/eio). It is a single consumer version of the queue described in [Implementing lock-free queues](https://people.cs.pitt.edu/~jacklange/teaching/cs2510-f12/papers/implementing_lock_free.pdf).
 
 
@@ -25,6 +27,10 @@ let () = Ws_deque.push q 100
 let () = assert (Ws_deque.pop q = 100)
 ```
 
+## Benchmarks
+
+There is a number of benchmarks in `bench/` directory. You can run them with `make bench`. See [bench/README.md](bench/README.md) for more details.
+
 ## Contributing
 
 Contributions of more lockfree data structures appreciated! Please create

bench/README.md

@@ -0,0 +1,11 @@
+Benchmarks for lockfree
+
+# General usage 
+
+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)

bench/dune

@@ -1,3 +1,3 @@
-(executable
- (name main)
+(executables
+ (names main mpmc_queue_cmd)
  (libraries lockfree unix yojson))

bench/main.ml

@@ -1,13 +1,23 @@
-let benchmark_list = [ Bench_spsc_queue.bench ]
+let benchmark_list =
+  [
+    Bench_spsc_queue.bench;
+    Mpmc_queue.bench ~takers:4 ~pushers:4;
+    Mpmc_queue.bench ~takers:1 ~pushers:8;
+    Mpmc_queue.bench ~takers:8 ~pushers:1;
+    Mpmc_queue.bench ~use_cas:true ~takers:4 ~pushers:4;
+    Mpmc_queue.bench ~use_cas:true ~takers:1 ~pushers:8;
+    Mpmc_queue.bench ~use_cas:true ~takers:8 ~pushers:1;
+  ]
 
 let () =
   let results =
+    (* todo: should assert no stranded domains between tests. *)
     List.map (fun f -> f ()) benchmark_list
     |> List.map Benchmark_result.to_json
     |> String.concat ", "
   in
   let output =
-    Printf.sprintf {| {"name": "lockfree", "results": %s}|} results
+    Printf.sprintf {| {"name": "lockfree", "results": [%s]}|} results
     (* Cannot use Yojson rewriters as of today none works on OCaml 5.1.0.
        This at least verifies that the manually crafted JSON is well-formed.
 

bench/mpmc_queue.ml

@@ -0,0 +1,99 @@
+open Lockfree.Mpmc_relaxed_queue
+
+let num_of_elements = ref 500_000
+let num_of_pushers = ref 4
+let num_of_takers = ref 4
+let num_of_iterations = ref 10
+let use_cas_intf = ref false
+let pop = ref Not_lockfree.pop
+let push = ref Not_lockfree.push
+
+let taker queue num_of_elements () =
+  let i = ref 0 in
+  while !i < num_of_elements do
+    if Option.is_some (!pop queue) then i := !i + 1
+  done
+
+let pusher queue num_of_elements () =
+  let i = ref 0 in
+  while !i < num_of_elements do
+    if !push queue !i then i := !i + 1
+  done
+
+let create_output ~time_median ~throughput_median ~throughput_stddev =
+  let time =
+    ({
+       name = "time";
+       value = `Numeric time_median;
+       units = "s";
+       description = "median time";
+     }
+      : Benchmark_result.Metric.t)
+  in
+  let throughput =
+    ({
+       name = "throughput";
+       value = `Numeric throughput_median;
+       units = "item/s";
+       description = "median throughput";
+     }
+      : Benchmark_result.Metric.t)
+  in
+  let throughput_stddev =
+    ({
+       name = "throughput-stddev";
+       value = `Numeric throughput_stddev;
+       units = "item/s";
+       description = "stddev throughput";
+     }
+      : Benchmark_result.Metric.t)
+  in
+  let metrics = [ time; throughput; throughput_stddev ] in
+  let name =
+    Printf.sprintf "mpmc-queue-pushers:%d,takers:%d,use-cas:%b" !num_of_pushers
+      !num_of_takers !use_cas_intf
+  in
+  ({ name; metrics } : Benchmark_result.t)
+
+let run_bench () =
+  if !use_cas_intf then (
+    push := Lockfree.Mpmc_relaxed_queue.Not_lockfree.CAS_interface.push;
+    pop := Lockfree.Mpmc_relaxed_queue.Not_lockfree.CAS_interface.pop);
+  let queue = create ~size_exponent:10 () in
+  let orchestrator =
+    Orchestrator.init
+      ~total_domains:(!num_of_takers + !num_of_pushers)
+      ~rounds:!num_of_iterations
+  in
+  (* define function to start domains *)
+  let start_n_domains n f =
+    assert (!num_of_elements mod n == 0);
+    let items_per_pusher = !num_of_elements / n in
+    List.init n (fun _ ->
+        Domain.spawn (fun () ->
+            Orchestrator.worker orchestrator (f queue items_per_pusher)))
+  in
+  (* start domains *)
+  let domains =
+    let takers = start_n_domains !num_of_takers taker in
+    let pushers = start_n_domains !num_of_pushers pusher in
+    Sys.opaque_identity (pushers @ takers)
+  in
+  (* run test *)
+  let times = Orchestrator.run orchestrator in
+  List.iter Domain.join domains;
+  let time_median = Stats.median times in
+  let throughputs =
+    List.map (fun time -> Int.to_float !num_of_elements /. time) times
+  in
+  let throughput_median = Stats.median throughputs in
+  let throughput_stddev = Stats.stddev throughputs in
+  create_output ~time_median ~throughput_median ~throughput_stddev
+
+let bench ?takers ?pushers ?use_cas ?iterations ?elements () =
+  num_of_takers := Option.value takers ~default:!num_of_takers;
+  num_of_pushers := Option.value pushers ~default:!num_of_pushers;
+  use_cas_intf := Option.value use_cas ~default:!use_cas_intf;
+  num_of_iterations := Option.value iterations ~default:!num_of_iterations;
+  num_of_elements := Option.value elements ~default:!num_of_elements;
+  run_bench ()

bench/mpmc_queue.mli

@@ -0,0 +1,8 @@
+val bench :
+  ?takers:int ->
+  ?pushers:int ->
+  ?use_cas:bool ->
+  ?iterations:int ->
+  ?elements:int ->
+  unit ->
+  Benchmark_result.t

bench/mpmc_queue_cmd.ml

@@ -0,0 +1,25 @@
+let elements = ref 100_000
+let pushers = ref 4
+let takers = ref 4
+let iterations = ref 10
+let use_cas = ref false
+
+let speclist =
+  [
+    ("-items", Arg.Set_int elements, "number of items to insert and remove");
+    ("-pushers", Arg.Set_int pushers, "number of domains pushing items");
+    ("-takers", Arg.Set_int takers, "number of domains taking times");
+    ("-iterations", Arg.Set_int iterations, "run the benchmark this many times");
+    ("-use-cas", Arg.Set use_cas, "use CAS instead of FAD");
+  ]
+
+let _f () =
+  Arg.parse speclist
+    (fun _ -> ())
+    "mpmc_queue.exe [-items INT] [-pushers INT] [-takers INT] [-iterations \
+     INT] [-use-cas]";
+  let result =
+    Mpmc_queue.bench ~takers:!takers ~pushers:!pushers ~use_cas:!use_cas
+      ~iterations:!iterations ~elements:!elements ()
+  in
+  Benchmark_result.to_json result |> Yojson.Basic.prettify |> print_string

bench/orchestrator.ml

@@ -0,0 +1,43 @@
+type t = {
+  ready : int Atomic.t;
+  total_domains : int;
+  round : int Atomic.t;
+  rounds : int;
+}
+
+let init ~total_domains ~rounds =
+  { ready = Atomic.make 0; total_domains; round = Atomic.make 0; rounds }
+
+let wait_until_all_ready ?(round = 0) { ready; total_domains; _ } =
+  while Atomic.get ready < total_domains * (round + 1) do
+    ()
+  done
+
+let worker ({ ready; round; rounds; _ } as t) f =
+  Atomic.incr ready;
+  wait_until_all_ready t;
+  (* all domains are up at this point *)
+  for i = 1 to rounds do
+    (* wait for signal to start work *)
+    while Atomic.get round < i do
+      ()
+    done;
+    f ();
+    (* signal that we're done *)
+    Atomic.incr ready
+  done
+
+let run ?(drop_first = true) ({ round; rounds; _ } as t) =
+  wait_until_all_ready t;
+  (* all domains are up, can start benchmarks *)
+  let results = ref [] in
+  for i = 1 to rounds do
+    let start_time = Unix.gettimeofday () in
+    Atomic.incr round;
+    wait_until_all_ready ~round:i t;
+    let end_time = Unix.gettimeofday () in
+
+    let diff = end_time -. start_time in
+    if drop_first && i == 1 then () else results := diff :: !results
+  done;
+  !results

bench/orchestrator.mli

@@ -0,0 +1,8 @@
+(** Helper library that ensures all workers have started before any 
+  starts making progress on the benchmark. *)
+
+type t
+
+val init : total_domains:int -> rounds:int -> t
+val worker : t -> (unit -> unit) -> unit
+val run : ?drop_first:bool -> t -> float List.t

bench/run_all.sh

@@ -0,0 +1,22 @@
+#!/bin/bash
+set -x
+
+date=$(date)
+
+echo "$date" > benchmarks_output.txt
+
+with_cas=("" "-use-cas")
+rw_balance=("-takers 4 -pushers 4" "-takers 1 pushers 8" "-takers 8 pushers 1")
+path="./../_build/default/bench/mpmc_queue.exe "
+
+for i in "${with_cas[@]}" 
+do
+  for j in "${rw_balance[@]}" 
+  do
+	  cmd="$path $j $i"
+    output=$($cmd)
+
+    echo "$cmd" >> benchmarks_output.txt
+    echo "$output" >> benchmarks_output.txt
+  done 
+done

bench/stats.ml

@@ -0,0 +1,36 @@
+let median data =
+  let data = List.sort Float.compare data in
+  let len = List.length data in
+  if len mod 2 == 1 then List.nth data (List.length data / 2)
+  else
+    let a = List.nth data ((len / 2) - 1) in
+    let b = List.nth data (len / 2) in
+    (a +. b) /. 2.
+
+let mean data =
+  let sum = List.fold_left (fun curr_sum b -> curr_sum +. b) 0. data in
+  let n = Int.to_float (List.length data) in
+  sum /. n
+
+let stddev data =
+  let mean = mean data in
+  let sum =
+    List.fold_left
+      (fun curr_sum datapoint ->
+        let squared_diff = Float.pow (datapoint -. mean) 2. in
+        curr_sum +. squared_diff)
+      0. data
+  in
+  let n = Int.to_float (List.length data) in
+  Float.sqrt (sum /. n)
+
+let cmp ?(epsilon = 0.01) a b = Float.abs (a -. b) < epsilon
+
+let sanity_checks () =
+  assert (cmp (mean [ 1.; 5. ]) 3.);
+  assert (cmp (mean [ 1.; 3.; 7. ]) 3.6666);
+  assert (cmp (stddev [ 1.; 5. ]) 2.);
+  assert (cmp (stddev [ 1.; 3.; 7. ]) 2.4944)
+;;
+
+sanity_checks ()

bench/stats.mli

@@ -0,0 +1,3 @@
+val mean : float List.t -> float
+val stddev : float List.t -> float
+val median : float List.t -> float

src/lockfree.ml

@@ -29,3 +29,4 @@ Copyright (c) 2017, Nicolas ASSOUAD <[email protected]>
 module Ws_deque = Ws_deque
 module Spsc_queue = Spsc_queue
 module Mpsc_queue = Mpsc_queue
+module Mpmc_relaxed_queue = Mpmc_relaxed_queue

src/lockfree.mli

@@ -33,3 +33,4 @@ Copyright (c) 2017, Nicolas ASSOUAD <[email protected]>
 module Ws_deque = Ws_deque
 module Spsc_queue = Spsc_queue
 module Mpsc_queue = Mpsc_queue
+module Mpmc_relaxed_queue = Mpmc_relaxed_queue