simon/moonpool
1
0
Fork 0
mirror of https://github.com/c-cube/moonpool.git synced 2025-12-10 13:14:05 -05:00
Code Issues Projects Releases Packages Wiki Activity Actions

wip: debug pool

Browse source
This commit is contained in:
Simon Cruanes 2023-10-25 09:57:19 -04:00
parent 3d7e272d01
commit fdc188c291
No known key found for this signature in database
GPG key ID: EBFFF6F283F3A2B4
1 changed files with 46 additions and 26 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

72
src/pool.ml
View file

@ -40,8 +40,12 @@ let find_current_worker_ (self : state) : worker_state option =
(** Run [task] as is, on the pool. *) (** Run [task] as is, on the pool. *)
let run_direct_ (self : state) (w : worker_state option) (task : task) : unit = let run_direct_ (self : state) (w : worker_state option) (task : task) : unit =
match w with match w with
| Some w -> WSQ.push w.q task | Some w ->
| None -> Bb_queue.push self.main_q task print_endline "push local";
WSQ.push w.q task
| None ->
print_endline "push blocking";
Bb_queue.push self.main_q task
let run_async_ (self : state) (task : task) : unit = let run_async_ (self : state) (task : task) : unit =
(* stay on current worker if possible *) (* stay on current worker if possible *)
@ -88,6 +92,7 @@ let worker_thread_ (self : state) (runner : t) (w : worker_state) ~on_exn
in in
let run_self_tasks_ () = let run_self_tasks_ () =
print_endline "run self tasks";
let continue = ref true in let continue = ref true in
let pop_retries = ref 0 in let pop_retries = ref 0 in
while !continue do while !continue do
@ -104,6 +109,7 @@ let worker_thread_ (self : state) (runner : t) (w : worker_state) ~on_exn
(* get a task from another worker *) (* get a task from another worker *)
let try_to_steal_work () : task option = let try_to_steal_work () : task option =
print_endline "try to steal work";
try try
for _retry = 1 to 3 do for _retry = 1 to 3 do
Array.iter Array.iter
@ -119,32 +125,46 @@ let worker_thread_ (self : state) (runner : t) (w : worker_state) ~on_exn
with Got_task task -> Some task with Got_task task -> Some task
in in
let main_loop () = (* try to steal work multiple times *)
let steal_attempts = ref 0 in let try_to_steal_work_loop () : bool =
while true do try
run_self_tasks_ (); let unsuccessful_steal_attempts = ref 0 in
while !unsuccessful_steal_attempts < steal_attempt_max_retry do
match try_to_steal_work () with match try_to_steal_work () with
| Some task -> | Some task ->
steal_attempts := 0; run_task task;
run_task task raise_notrace Exit
| None -> | None ->
incr steal_attempts; incr unsuccessful_steal_attempts;
Domain_.relax (); Domain_.relax ()
done;
if !steal_attempts > steal_attempt_max_retry then ( false
steal_attempts := 0; with Exit -> true
let task = Bb_queue.pop self.main_q in
run_task task
)
done
in in
try let main_loop () =
(* handle domain-local await *) (try
Dla_.using ~prepare_for_await:Suspend_.prepare_for_await while true do
~while_running:main_loop run_self_tasks_ ();
with Bb_queue.Closed -> ()
if not (try_to_steal_work_loop ()) then (
Array.iteri
(fun i w -> Printf.printf "w[%d].q.size=%d\n" i (WSQ.size w.q))
self.workers;
Printf.printf "bq.size=%d\n%!" (Bb_queue.size self.main_q);
print_endline "wait block";
let task = Bb_queue.pop self.main_q in
run_task task
)
done
with Bb_queue.Closed -> ());
run_self_tasks_ ()
in
(* handle domain-local await *)
Dla_.using ~prepare_for_await:Suspend_.prepare_for_await
~while_running:main_loop
let default_thread_init_exit_ ~dom_id:_ ~t_id:_ () = () let default_thread_init_exit_ ~dom_id:_ ~t_id:_ () = ()