Change to use Picos instead of DLT and DLA

This adds support for cancelation through Picos and removes explicit support for
timeouts, which simplifies the library.  This basically also means that one can
no longer use Kcas without a scheduler.

Author: polytypic

bench/bench_accumulator.ml

@@ -9,7 +9,7 @@ let run_one ~budgetf ~n_domains ?(n_ops = 180 * Util.iter_factor) () =
   let n_ops_todo = Atomic.make n_ops |> Multicore_magic.copy_as_padded in
 
   let init _ = () in
-
+  let wrap _ _ action = Scheduler.run action in
   let work _ () =
     let rec work () =
       let n = Util.alloc n_ops_todo in
@@ -34,7 +34,7 @@ let run_one ~budgetf ~n_domains ?(n_ops = 180 * Util.iter_factor) () =
       (if n_domains = 1 then "" else "s")
   in
 
-  Times.record ~budgetf ~n_domains ~init ~work ~after ()
+  Times.record ~budgetf ~n_domains ~init ~wrap ~work ~after ()
   |> Times.to_thruput_metrics ~n:n_ops ~config ~singular:"operation"
 
 let run_suite ~budgetf =

bench/bench_dllist.ml

@@ -12,9 +12,10 @@ let run_single ~budgetf ?(n_msgs = 15 * Util.iter_factor) () =
     assert (Dllist.is_empty t);
     Util.generate_push_and_pop_sequence n_msgs
   in
+  let wrap _ _ action = Scheduler.run action in
   let work _ bits = Util.Bits.iter op bits in
 
-  Times.record ~budgetf ~n_domains:1 ~init ~work ()
+  Times.record ~budgetf ~n_domains:1 ~init ~wrap ~work ()
   |> Times.to_thruput_metrics ~n:n_msgs ~singular:"message" ~config:"one domain"
 
 let run_one ~budgetf ?(n_adders = 2) ?(n_takers = 2) ?(factor = 1)
@@ -31,6 +32,7 @@ let run_one ~budgetf ?(n_adders = 2) ?(n_takers = 2) ?(factor = 1)
     Atomic.set n_msgs_to_take n_msgs;
     Atomic.set n_msgs_to_add n_msgs
   in
+  let wrap _ _ action = Scheduler.run action in
   let work i () =
     if i < n_adders then
       let rec work () =
@@ -70,7 +72,7 @@ let run_one ~budgetf ?(n_adders = 2) ?(n_takers = 2) ?(factor = 1)
       (format "taker" false n_takers)
   in
 
-  Times.record ~budgetf ~n_domains ~init ~work ()
+  Times.record ~budgetf ~n_domains ~init ~wrap ~work ()
   |> Times.to_thruput_metrics ~n:n_msgs ~singular:"message" ~config
 
 let run_suite ~budgetf =

bench/bench_hashtbl.ml

@@ -24,7 +24,7 @@ let run_one ~budgetf ~n_domains ?(n_ops = 40 * Util.iter_factor)
     Atomic.set n_ops_todo n_ops;
     Random.State.make_self_init ()
   in
-
+  let wrap _ _ action = Scheduler.run action in
   let work _ state =
     let rec work () =
       let n = Util.alloc n_ops_todo in
@@ -56,7 +56,7 @@ let run_one ~budgetf ~n_domains ?(n_ops = 40 * Util.iter_factor)
       percent_read
   in
 
-  Times.record ~budgetf ~n_domains ~init ~work ()
+  Times.record ~budgetf ~n_domains ~init ~wrap ~work ()
   |> Times.to_thruput_metrics ~n:n_ops ~singular:"operation" ~config
 
 let run_suite ~budgetf =

bench/bench_mvar.ml

@@ -11,6 +11,7 @@ let run_one ~budgetf ?(n_adders = 2) ?(blocking_add = false) ?(n_takers = 2)
   let n_msgs_to_add = Atomic.make n_msgs |> Multicore_magic.copy_as_padded in
 
   let init _ = () in
+  let wrap _ _ action = Scheduler.run action in
   let work i () =
     if i < n_adders then
       if blocking_add then
@@ -79,7 +80,7 @@ let run_one ~budgetf ?(n_adders = 2) ?(blocking_add = false) ?(n_takers = 2)
       (format "taker" blocking_take n_takers)
   in
 
-  Times.record ~budgetf ~n_domains ~init ~work ~after ()
+  Times.record ~budgetf ~n_domains ~init ~wrap ~work ~after ()
   |> Times.to_thruput_metrics ~n:n_msgs ~singular:"message" ~config
 
 let run_suite ~budgetf =

bench/bench_parallel_cmp.ml

@@ -11,7 +11,7 @@ let run_one ~budgetf ~n_domains ?(n_ops = 50 * Util.iter_factor) () =
   let n_ops_todo = Atomic.make n_ops |> Multicore_magic.copy_as_padded in
 
   let init i = Array.unsafe_get xs i in
-
+  let wrap _ _ action = Scheduler.run action in
   let work _ x =
     let tx1 ~xt =
       let a = Xt.get ~xt a in
@@ -41,7 +41,7 @@ let run_one ~budgetf ~n_domains ?(n_ops = 50 * Util.iter_factor) () =
     Printf.sprintf "%d worker%s" n_domains (if n_domains = 1 then "" else "s")
   in
 
-  Times.record ~budgetf ~n_domains ~init ~work ~after ()
+  Times.record ~budgetf ~n_domains ~init ~wrap ~work ~after ()
   |> Times.to_thruput_metrics ~n:n_ops ~singular:"transaction" ~config
 
 let run_suite ~budgetf =

bench/bench_queue.ml

@@ -10,9 +10,10 @@ let run_one_domain ~budgetf ?(n_msgs = 50 * Util.iter_factor) () =
     assert (Queue.is_empty t);
     Util.generate_push_and_pop_sequence n_msgs
   in
+  let wrap _ _ action = Scheduler.run action in
   let work _ bits = Util.Bits.iter op bits in
 
-  Times.record ~budgetf ~n_domains:1 ~init ~work ()
+  Times.record ~budgetf ~n_domains:1 ~init ~wrap ~work ()
   |> Times.to_thruput_metrics ~n:n_msgs ~singular:"message" ~config:"one domain"
 
 let run_one ~budgetf ?(n_adders = 2) ?(blocking_add = false) ?(n_takers = 2)
@@ -29,6 +30,7 @@ let run_one ~budgetf ?(n_adders = 2) ?(blocking_add = false) ?(n_takers = 2)
     Atomic.set n_msgs_to_take n_msgs;
     Atomic.set n_msgs_to_add n_msgs
   in
+  let wrap _ _ action = Scheduler.run action in
   let work i () =
     if i < n_adders then
       let rec work () =
@@ -79,7 +81,7 @@ let run_one ~budgetf ?(n_adders = 2) ?(blocking_add = false) ?(n_takers = 2)
       (format "taker" blocking_take n_takers)
   in
 
-  Times.record ~budgetf ~n_domains ~init ~work ()
+  Times.record ~budgetf ~n_domains ~init ~wrap ~work ()
   |> Times.to_thruput_metrics ~n:n_msgs ~singular:"message" ~config
 
 let run_suite ~budgetf =

bench/bench_stack.ml

@@ -10,9 +10,10 @@ let run_one_domain ~budgetf ?(n_msgs = 50 * Util.iter_factor) () =
     assert (Stack.is_empty t);
     Util.generate_push_and_pop_sequence n_msgs
   in
+  let wrap _ _ action = Scheduler.run action in
   let work _ bits = Util.Bits.iter op bits in
 
-  Times.record ~budgetf ~n_domains:1 ~init ~work ()
+  Times.record ~budgetf ~n_domains:1 ~init ~wrap ~work ()
   |> Times.to_thruput_metrics ~n:n_msgs ~singular:"message" ~config:"one domain"
 
 let run_one ~budgetf ?(n_adders = 2) ?(blocking_add = false) ?(n_takers = 2)
@@ -29,6 +30,7 @@ let run_one ~budgetf ?(n_adders = 2) ?(blocking_add = false) ?(n_takers = 2)
     Atomic.set n_msgs_to_take n_msgs;
     Atomic.set n_msgs_to_add n_msgs
   in
+  let wrap _ _ action = Scheduler.run action in
   let work i () =
     if i < n_adders then
       let rec work () =
@@ -79,7 +81,7 @@ let run_one ~budgetf ?(n_adders = 2) ?(blocking_add = false) ?(n_takers = 2)
       (format "taker" blocking_take n_takers)
   in
 
-  Times.record ~budgetf ~n_domains ~init ~work ()
+  Times.record ~budgetf ~n_domains ~init ~wrap ~work ()
   |> Times.to_thruput_metrics ~n:n_msgs ~singular:"message" ~config
 
 let run_suite ~budgetf =

bench/bench_xt.ml

@@ -15,6 +15,7 @@ let run_one ~budgetf ?(n_locs = 2)
   in
 
   let init _ = () in
+  let wrap _ _ action = Scheduler.run action in
   let work _ () =
     let rec loop i =
       if i > 0 then begin
@@ -27,7 +28,7 @@ let run_one ~budgetf ?(n_locs = 2)
 
   let config = Printf.sprintf "%d loc tx" n_locs in
 
-  Times.record ~budgetf ~n_domains:1 ~init ~work ()
+  Times.record ~budgetf ~n_domains:1 ~init ~wrap ~work ()
   |> Times.to_thruput_metrics ~n:n_iter ~singular:"transaction" ~config
 
 let run_suite ~budgetf =

bench/bench_xt_ro.ml

@@ -15,6 +15,7 @@ let run_one ~budgetf ?(n_locs = 2)
   in
 
   let init _ = () in
+  let wrap _ _ action = Scheduler.run action in
   let work _ () =
     let rec loop i =
       if i > 0 then begin
@@ -27,7 +28,7 @@ let run_one ~budgetf ?(n_locs = 2)
 
   let config = Printf.sprintf "%d loc tx" n_locs in
 
-  Times.record ~budgetf ~n_domains:1 ~init ~work ()
+  Times.record ~budgetf ~n_domains:1 ~init ~wrap ~work ()
   |> Times.to_thruput_metrics ~n:n_iter ~singular:"transaction" ~config
 
 let run_suite ~budgetf =

bench/dune

@@ -14,6 +14,7 @@ let () =
  (action
   (run %{test} -brief))
  (libraries
+  scheduler
   kcas_data
   multicore-bench
   backoff