test: add dep on trace-tef; add new test for scheduling issues

trying to expose that sometimes, some workers might be asleep while
others do several tasks, because they're sleeping on the "wrong" queue

Makefile

@@ -28,7 +28,7 @@ BENCH_CUTOFF?=20
 bench-fib:
 	@echo running for N=$(N)
 	dune build $(DUNE_OPTS_BENCH) benchs/fib_rec.exe
-	hyperfine -L psize $(BENCH_PSIZE) \
+	hyperfine -L psize $(BENCH_PSIZE) --warmup=1 \
 		'./_build/default/benchs/fib_rec.exe -cutoff $(BENCH_CUTOFF) -niter $(NITER) -psize={psize} -n $(N)'
 
 PI_NSTEPS?=100_000_000
@@ -36,7 +36,7 @@ PI_MODES?=seq,par1,forkjoin
 bench-pi:
 	@echo running for N=$(PI_NSTEPS)
 	dune build $(DUNE_OPTS_BENCH) benchs/pi.exe
-	hyperfine -L mode $(PI_MODES) \
+	hyperfine -L mode $(PI_MODES) --warmup=1 \
 		'./_build/default/benchs/pi.exe -mode={mode} -n $(PI_NSTEPS)'
 
 .PHONY: test clean bench-fib bench-pi

dune-project

@@ -20,6 +20,7 @@
   dune
   (either (>= 1.0))
   (trace :with-test)
+  (trace-tef :with-test)
   (qcheck-core (and :with-test (>= 0.19)))
   (odoc :with-doc)
   (mdx

moonpool.opam

@@ -13,6 +13,7 @@ depends: [
   "dune" {>= "3.0"}
   "either" {>= "1.0"}
   "trace" {with-test}
+  "trace-tef" {with-test}
   "qcheck-core" {with-test & >= "0.19"}
   "odoc" {with-doc}
   "mdx" {>= "1.9.0" & with-test}

test/dune

@@ -8,10 +8,13 @@
   t_props
   t_chan_train
   t_resource
+  t_unfair
   t_bounded_queue)
  (libraries
   moonpool
   qcheck-core
   qcheck-core.runner
   ;tracy-client.trace
+  unix
+  trace-tef
   trace))

test/t_unfair.ml

@@ -0,0 +1,44 @@
+(* exhibits unfairness *)
+
+open Moonpool
+
+let ( let@ ) = ( @@ )
+
+let sleep_for f () =
+  let@ _sp = Trace.with_span ~__FILE__ ~__LINE__ "sleep" in
+  Thread.delay f
+
+let () =
+  let@ () = Trace_tef.with_setup () in
+  let@ _sp = Trace.with_span ~__FILE__ ~__LINE__ "main" in
+
+  let pool =
+    Pool.create ~min:3
+      ~on_init_thread:(fun ~dom_id:_ ~t_id () ->
+        Trace.set_thread_name (Printf.sprintf "pool worker %d" t_id))
+      ~around_task:
+        ( (fun self -> Trace.counter_int "n_tasks" (Pool.num_tasks self)),
+          fun self () -> Trace.counter_int "n_tasks" (Pool.num_tasks self) )
+      ()
+  in
+
+  (* make all threads busy *)
+  Pool.run_async pool (sleep_for 0.01);
+  Pool.run_async pool (sleep_for 0.01);
+  Pool.run_async pool (sleep_for 0.05);
+
+  let t = Unix.gettimeofday () in
+  for _i = 1 to 100 do
+    let@ _sp = Trace.with_span ~__FILE__ ~__LINE__ "schedule step" in
+    Pool.run_async pool (sleep_for 0.001);
+    Pool.run_async pool (sleep_for 0.001);
+    Pool.run_async pool (sleep_for 0.01)
+  done;
+
+  Printf.printf "pool size: %d\n%!" (Pool.num_tasks pool);
+  (let@ _sp = Trace.with_span ~__FILE__ ~__LINE__ "shutdown" in
+   Pool.shutdown pool);
+  Printf.printf "pool size after shutdown: %d\n%!" (Pool.num_tasks pool);
+
+  let elapsed = Unix.gettimeofday () -. t in
+  Printf.printf "elapsed: %.4fs\n%!" elapsed