Add Picos_stdio benchmark

polytypic · polytypic · commit b083335bc11a · 2024-04-08T04:27:20.000+03:00
diff --git a/bench/bench_cancel_after.ml b/bench/bench_cancel_after.ml
@@ -32,7 +32,8 @@ let run_round_trip ~budgetf ~n_domains () =
   let config =
     Printf.sprintf "%d worker%s" n_domains (if n_domains = 1 then "" else "s")
   in
-  Times.record ~budgetf ~n_domains ~init ~wrap ~work ()
+  Times.record ~budgetf ~n_domains ~init ~n_warmups:1 ~n_runs_min:1 ~wrap ~work
+    ()
   |> Times.to_thruput_metrics ~n:n_ops ~singular:"round-trip" ~config
 
 let run_async ~budgetf ~n_domains () =
@@ -81,7 +82,8 @@ let run_async ~budgetf ~n_domains () =
   let config =
     Printf.sprintf "%d worker%s" n_domains (if n_domains = 1 then "" else "s")
   in
-  Times.record ~budgetf ~n_domains ~init ~wrap ~work ()
+  Times.record ~budgetf ~n_domains ~n_warmups:1 ~n_runs_min:1 ~init ~wrap ~work
+    ()
   |> Times.to_thruput_metrics ~n:n_ops ~singular:"async round-trip" ~config
 
 let run_suite ~budgetf =
diff --git a/bench/bench_mutex.ml b/bench/bench_mutex.ml
@@ -65,7 +65,8 @@ let run_one ~budgetf ~n_fibers ~use_domains () =
       (if use_domains then "domain" else "fiber")
       (if n_fibers = 1 then "" else "s")
   in
-  Times.record ~budgetf ~n_domains ~init ~wrap ~work ()
+  Times.record ~budgetf ~n_domains ~n_warmups:1 ~n_runs_min:1 ~init ~wrap ~work
+    ()
   |> Times.to_thruput_metrics ~n:n_ops ~singular:"locked yield" ~config
 
 let run_suite ~budgetf =
diff --git a/bench/bench_spawn.ml b/bench/bench_spawn.ml
@@ -26,7 +26,8 @@ let run_one ~budgetf ~at_a_time () =
   in
 
   let config = Printf.sprintf "%d at a time" at_a_time in
-  Times.record ~budgetf ~n_domains:1 ~init ~wrap ~work ()
+  Times.record ~budgetf ~n_domains:1 ~n_warmups:1 ~n_runs_min:1 ~init ~wrap
+    ~work ()
   |> Times.to_thruput_metrics ~n:n_spawns ~singular:"spawn" ~config
 
 let run_suite ~budgetf =
diff --git a/bench/bench_stdio.ml b/bench/bench_stdio.ml
@@ -0,0 +1,53 @@
+open Multicore_bench
+open Picos_stdio
+
+let run_one ~budgetf ~block_or_nonblock ~n_domains () =
+  let n_bytes =
+    match block_or_nonblock with `Block -> 4096 | `Nonblock -> 65536
+  in
+
+  let init _ =
+    let inn, out = Unix.pipe ~cloexec:true () in
+    (inn, out, Bytes.create 1)
+  in
+  let wrap _ _ = Scheduler.run in
+  let work _ (inn, out, byte) =
+    begin
+      match block_or_nonblock with
+      | `Block -> ()
+      | `Nonblock -> Unix.set_nonblock inn
+    end;
+    let n = Unix.write out (Bytes.create n_bytes) 0 n_bytes in
+    assert (n = n_bytes);
+    for _ = 1 to n_bytes do
+      let n : int = Unix.read inn byte 0 1 in
+      assert (n = 1)
+    done;
+    Unix.close inn;
+    Unix.close out
+  in
+
+  let times =
+    Times.record ~budgetf ~n_domains ~n_warmups:1 ~n_runs_min:1 ~init ~wrap
+      ~work ()
+  in
+
+  let config =
+    Printf.sprintf "%d worker%s" n_domains (if n_domains = 1 then "" else "s")
+  and singular =
+    match block_or_nonblock with
+    | `Block -> "blocking read"
+    | `Nonblock -> "non-blocking read"
+  in
+  Times.to_thruput_metrics ~n:(n_bytes * n_domains) ~singular ~config times
+
+let run_suite ~budgetf =
+  Util.cross [ `Nonblock; `Block ] [ 1; 2; 4 ]
+  |> List.concat_map @@ fun (block_or_nonblock, n_domains) ->
+     if
+       Sys.win32
+       || Picos_domain.recommended_domain_count () < n_domains
+       || String.starts_with ~prefix:"5.0." Sys.ocaml_version
+          && block_or_nonblock == `Block
+     then []
+     else run_one ~budgetf ~block_or_nonblock ~n_domains ()
diff --git a/bench/bench_yield.ml b/bench/bench_yield.ml
@@ -27,7 +27,8 @@ let run_one ~budgetf ~n_fibers () =
   let config =
     Printf.sprintf "%d fiber%s" n_fibers (if n_fibers = 1 then "" else "s")
   in
-  Times.record ~budgetf ~n_domains:1 ~init ~wrap ~work ()
+  Times.record ~budgetf ~n_domains:1 ~n_warmups:1 ~n_runs_min:1 ~init ~wrap
+    ~work ()
   |> Times.to_thruput_metrics ~n:n_yields ~singular:"yield" ~config
 
 let run_suite ~budgetf =
diff --git a/bench/dune b/bench/dune
@@ -10,6 +10,7 @@
   picos.sync
   picos.domain
   picos.tls
+  picos.stdio
   (select
    scheduler.ml
    from
diff --git a/bench/main.ml b/bench/main.ml
@@ -12,6 +12,7 @@ let benchmarks =
     ("Ref with Picos_sync.Mutex", Bench_ref_mutex.run_suite);
     ("Foundation Mpsc_queue", Bench_mpsc_queue.run_suite);
     ("Picos_htbl", Bench_htbl.run_suite);
+    ("Picos_stdio", Bench_stdio.run_suite);
   ]
 
 let () = Multicore_bench.Cmd.run ~benchmarks ()
diff --git a/lib/picos_select/dune b/lib/picos_select/dune
@@ -10,6 +10,7 @@
   picos.domain
   picos.tls
   picos_thread_atomic
+  backoff
   threads.posix
   psq
   mtime
diff --git a/lib/picos_select/picos_select.ml b/lib/picos_select/picos_select.ml
@@ -3,7 +3,6 @@ open Picos
 let intr_sig = Sys.sigusr2
 let is_intr_sig s = s == intr_sig
 let intr_sigs = [ intr_sig ]
-let intr_pending = Atomic.make 0
 
 type cancel_at =
   | Cancel_at : {
@@ -49,6 +48,21 @@ type state = {
   new_ex : return_on list ref;
 }
 
+type intr_status = Cleared | Signaled
+
+type _ tdt =
+  | Nothing : [> `Nothing ] tdt
+  | Req : {
+      state : state;
+      mutable unused : bool;
+      computation : intr_status Computation.t;
+    }
+      -> [> `Req ] tdt
+
+type req = R : [< `Nothing | `Req ] tdt -> req [@@unboxed]
+type counter_state = { value : int; req : req }
+
+let intr_pending = Atomic.make { value = 0; req = R Nothing }
 let exit_exn_bt = Exn_bt.get_callstack 0 Exit
 
 let key =
@@ -180,9 +194,10 @@ and select_thread_continue s rd wr ex (rd_fds, wr_fds, ex_fds) =
   let ex = process_fds ex_fds ex (Picos_thread_atomic.exchange s.new_ex []) in
   let tos = process_timeouts s in
   let tos =
-    let n = Atomic.get intr_pending in
-    if n = 0 then tos
+    let state = Atomic.get intr_pending in
+    if state.value = 0 then tos
     else begin
+      assert (0 < state.value);
       Unix.kill (Unix.getpid ()) intr_sig;
       let idle = 0.000_001 (* 1μs *) in
       if tos < 0.0 || idle <= tos then idle else tos
@@ -302,17 +317,7 @@ let await_on file_descr op =
 (* *)
 
 module Intr = struct
-  type intr_status = Cleared | Signaled
-
-  type _ tdt =
-    | Nothing : [> `Nothing ] tdt
-    | Req : {
-        state : state;
-        mutable computation : intr_status Computation.t;
-      }
-        -> [> `Req ] tdt
-
-  type t = T : [< `Nothing | `Req ] tdt -> t [@@unboxed]
+  type t = req
 
   let cleared =
     let computation = Computation.create () in
@@ -321,29 +326,42 @@ module Intr = struct
 
   let intr_key =
     Picos_tls.new_key @@ fun () : [ `Req ] tdt ->
-    Req { state = get (); computation = cleared }
+    Req { state = get (); unused = true; computation = cleared }
+
+  let[@inline] use = function R Nothing -> () | R (Req r) -> r.unused <- false
 
   let handle _ =
     let (Req r) = Picos_tls.get intr_key in
     Computation.return r.computation Signaled
 
-  let intr_action trigger (Req r : [ `Req ] tdt) id =
+  let rec finish (Req r as req : [ `Req ] tdt) backoff =
+    let before = Atomic.get intr_pending in
+    r.unused && before.req != R req
+    && begin
+         use before.req;
+         let after = { value = before.value + 1; req = R req } in
+         if Atomic.compare_and_set intr_pending before after then
+           after.value = 1
+         else finish req (Backoff.once backoff)
+       end
+
+  let intr_action trigger (Req r as req : [ `Req ] tdt) id =
     match Computation.await r.computation with
     | Cleared ->
         (* No signal needs to be delivered. *)
         remove_action trigger r.state id
     | Signaled ->
         (* Signal was delivered before timeout. *)
         remove_action trigger r.state id;
-        if Atomic.fetch_and_add intr_pending 1 = 0 then begin
+        if finish req Backoff.default then
           (* We need to make sure at least one select thread will keep on
              triggering interrupts. *)
           wakeup r.state `Alive
-        end
     | exception Exit ->
         (* The timeout was triggered.  This must have been called from the
            select thread, which will soon trigger an interrupt. *)
-        Atomic.incr intr_pending
+        let _ : bool = finish req Backoff.default in
+        ()
 
   let () =
     if not Sys.win32 then begin
@@ -353,37 +371,47 @@ module Intr = struct
       assert (old_behavior == Signal_default)
     end
 
-  let nothing = T Nothing
+  let nothing = R Nothing
 
   let[@alert "-handler"] req ~seconds =
     if Sys.win32 then
       invalid_arg "Picos_select.Intr is not supported on Windows"
     else begin
       let time = to_deadline ~seconds in
-      let (Req r as req) = Picos_tls.get intr_key in
-      assert (not (Computation.is_running r.computation));
-      let id = next_id r.state in
-      let computation = Computation.with_action req id intr_action in
-      r.computation <- computation;
+      (* assert (not (Computation.is_running r.computation)); *)
+      let state = get () in
+      let id = next_id state in
+      let computation = Computation.create () in
+      let (Req _ as req : [ `Req ] tdt) =
+        Req { state; unused = true; computation }
+      in
+      let _ : bool =
+        Computation.try_attach computation
+          (Trigger.from_action req id intr_action)
+      in
+      Picos_tls.set intr_key req;
       let entry = Cancel_at { time; exn_bt = exit_exn_bt; computation } in
-      add_timeout r.state id entry;
-      let was_blocked : int list = Thread.sigmask SIG_UNBLOCK intr_sigs in
-      assert (List.exists is_intr_sig was_blocked);
-      T req
+      add_timeout state id entry;
+      let _was_blocked : int list = Thread.sigmask SIG_UNBLOCK intr_sigs in
+      (* assert (List.exists is_intr_sig was_blocked); *)
+      R req
     end
 
+  let rec decr backoff =
+    let before = Atomic.get intr_pending in
+    use before.req;
+    let after = { value = before.value - 1; req = R Nothing } in
+    assert (0 <= after.value);
+    if not (Atomic.compare_and_set intr_pending before after) then
+      decr (Backoff.once backoff)
+
   let clr = function
-    | T Nothing -> ()
-    | T (Req r) ->
-        let was_blocked : int list = Thread.sigmask SIG_BLOCK intr_sigs in
-        assert (not (List.exists is_intr_sig was_blocked));
+    | R Nothing -> ()
+    | R (Req r as req) ->
+        let _was_blocked : int list = Thread.sigmask SIG_BLOCK intr_sigs in
+        (* assert (not (List.exists is_intr_sig was_blocked)); *)
         if not (Computation.try_return r.computation Cleared) then begin
-          while
-            let count = Atomic.get intr_pending in
-            count <= 0
-            || not (Atomic.compare_and_set intr_pending count (count - 1))
-          do
-            Thread.yield ()
-          done
+          let _ : bool = finish req Backoff.default in
+          decr Backoff.default
         end
 end
diff --git a/lib/picos_stdio/picos_stdio.ml b/lib/picos_stdio/picos_stdio.ml
diff --git a/test/test_select.ml b/test/test_select.ml

Original file line number	Diff line number	Diff line change
`@@ -12,6 +12,7 @@ let benchmarks =`
`12`	`12`	`("Ref with Picos_sync.Mutex", Bench_ref_mutex.run_suite);`
`13`	`13`	`("Foundation Mpsc_queue", Bench_mpsc_queue.run_suite);`
`14`	`14`	`("Picos_htbl", Bench_htbl.run_suite);`
	`15`	`+ ("Picos_stdio", Bench_stdio.run_suite);`
`15`	`16`	`]`
`16`	`17`
`17`	`18`	`let () = Multicore_bench.Cmd.run ~benchmarks ()`