simon/moonpool
1
0
Fork 0
mirror of https://github.com/c-cube/moonpool.git synced 2026-05-05 17:04:33 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions

refactor

Browse source
This commit is contained in:
Simon Cruanes 2026-03-24 19:48:37 -04:00
parent b8467ea97e
commit 863f57859d
No known key found for this signature in database
GPG key ID: EBFFF6F283F3A2B4
1 changed files with 7 additions and 19 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

26
src/core/ws_pool.ml
View file

@ -6,18 +6,7 @@ include Runner
let ( let@ ) = ( @@ ) let ( let@ ) = ( @@ )
module Id = struct
type t = unit ref
(** Unique identifier for a pool *)
let create () : t = Sys.opaque_identity (ref ())
let equal : t -> t -> bool = ( == )
end
type state = { type state = {
id_: Id.t;
(** Unique to this pool. Used to make sure tasks stay within the same
pool. *)
active: bool A.t; (** Becomes [false] when the pool is shutdown. *) active: bool A.t; (** Becomes [false] when the pool is shutdown. *)
mutable workers: worker_state array; (** Fixed set of workers. *) mutable workers: worker_state array; (** Fixed set of workers. *)
main_q: WL.task_full Queue.t; main_q: WL.task_full Queue.t;
@ -99,12 +88,15 @@ let schedule_in_main_queue (self : state) task : unit =
longer permitted *) longer permitted *)
raise Shutdown raise Shutdown
let schedule_from_w (self : worker_state) (task : WL.task_full) : unit = let schedule_from_anywhere_ (st : state) (task : WL.task_full) : unit =
match get_current_worker_ () with match get_current_worker_ () with
| Some w when Id.equal self.st.id_ w.st.id_ -> | Some w when st == w.st ->
(* use worker from the same pool *) (* use worker from the same pool *)
schedule_on_current_worker w task schedule_on_current_worker w task
| _ -> schedule_in_main_queue self.st task | _ -> schedule_in_main_queue st task
let schedule_from_w (w : worker_state) task : unit =
schedule_from_anywhere_ w.st task
exception Got_task of WL.task_full exception Got_task of WL.task_full
@ -224,9 +216,7 @@ let as_runner_ (self : state) : t =
~shutdown:(fun ~wait () -> shutdown_ self ~wait) ~shutdown:(fun ~wait () -> shutdown_ self ~wait)
~run_async:(fun ~fiber f -> ~run_async:(fun ~fiber f ->
let task = WL.T_start { fiber; f } in let task = WL.T_start { fiber; f } in
match get_current_worker_ () with schedule_from_anywhere_ self task)
| Some wst when wst.st == self -> schedule_from_w wst task
| _ -> schedule_in_main_queue self task)
~size:(fun () -> size_ self) ~size:(fun () -> size_ self)
~num_tasks:(fun () -> num_tasks_ self) ~num_tasks:(fun () -> num_tasks_ self)
() ()
@ -243,7 +233,6 @@ type ('a, 'b) create_args =
let create ?(on_init_thread = default_thread_init_exit_) let create ?(on_init_thread = default_thread_init_exit_)
?(on_exit_thread = default_thread_init_exit_) ?(on_exn = fun _ _ -> ()) ?(on_exit_thread = default_thread_init_exit_) ?(on_exn = fun _ _ -> ())
?num_threads ?name () : t = ?num_threads ?name () : t =
let pool_id_ = Id.create () in
let num_domains = Domain_pool_.max_number_of_domains () in let num_domains = Domain_pool_.max_number_of_domains () in
let num_threads = Util_pool_.num_threads ?num_threads () in let num_threads = Util_pool_.num_threads ?num_threads () in
@ -252,7 +241,6 @@ let create ?(on_init_thread = default_thread_init_exit_)
let pool = let pool =
{ {
id_ = pool_id_;
active = A.make true; active = A.make true;
workers = [||]; workers = [||];
main_q = Queue.create (); main_q = Queue.create ();