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

breaking: remove around_task from schedulers

Browse source
This commit is contained in:
Simon Cruanes 2025-11-11 23:59:36 -05:00
parent 2ce3fa7d3e
commit ee7972910f
No known key found for this signature in database
GPG key ID: EBFFF6F283F3A2B4
9 changed files with 17 additions and 84 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

13
src/core/background_thread.ml
View file

@ -6,18 +6,15 @@ type ('a, 'b) create_args =
?on_init_thread:(dom_id:int -> t_id:int -> unit -> unit) ->
?on_exit_thread:(dom_id:int -> t_id:int -> unit -> unit) ->
?on_exn:(exn -> Printexc.raw_backtrace -> unit) ->
?around_task:(t -> 'b) * (t -> 'b -> unit) ->
?name:string ->
'a
(** Arguments used in {!create}. See {!create} for explanations. *)
let create ?on_init_thread ?on_exit_thread ?on_exn ?around_task ?name () : t =
Fifo_pool.create ?on_init_thread ?on_exit_thread ?on_exn ?around_task ?name
~num_threads:1 ()
let create ?on_init_thread ?on_exit_thread ?on_exn ?name () : t =
Fifo_pool.create ?on_init_thread ?on_exit_thread ?on_exn ?name ~num_threads:1
()
let with_ ?on_init_thread ?on_exit_thread ?on_exn ?around_task ?name () f =
let pool =
create ?on_init_thread ?on_exit_thread ?on_exn ?around_task ?name ()
in
let with_ ?on_init_thread ?on_exit_thread ?on_exn ?name () f =
let pool = create ?on_init_thread ?on_exit_thread ?on_exn ?name () in
let@ () = Fun.protect ~finally:(fun () -> shutdown pool) in
f pool

1
src/core/background_thread.mli
View file

@ -13,7 +13,6 @@ type ('a, 'b) create_args =
?on_init_thread:(dom_id:int -> t_id:int -> unit -> unit) ->
?on_exit_thread:(dom_id:int -> t_id:int -> unit -> unit) ->
?on_exn:(exn -> Printexc.raw_backtrace -> unit) ->
?around_task:(t -> 'b) * (t -> 'b -> unit) ->
?name:string ->
'a
(** Arguments used in {!create}. See {!create} for explanations. *)

28
src/core/fifo_pool.ml
View file

@ -10,7 +10,6 @@ let ( let@ ) = ( @@ )
type state = {
threads: Thread.t array;
q: task_full Bb_queue.t; (** Queue for tasks. *)
around_task: WL.around_task;
mutable as_runner: t;
(* init options *)
name: string option;
@ -43,13 +42,10 @@ type ('a, 'b) create_args =
?on_init_thread:(dom_id:int -> t_id:int -> unit -> unit) ->
?on_exit_thread:(dom_id:int -> t_id:int -> unit -> unit) ->
?on_exn:(exn -> Printexc.raw_backtrace -> unit) ->
?around_task:(t -> 'b) * (t -> 'b -> unit) ->
?num_threads:int ->
?name:string ->
'a
let default_around_task_ : WL.around_task = AT_pair (ignore, fun _ _ -> ())
(** Run [task] as is, on the pool. *)
let schedule_ (self : state) (task : task_full) : unit =
try Bb_queue.push self.q task with Bb_queue.Closed -> raise Shutdown
@ -88,7 +84,6 @@ let cleanup (self : worker_state) : unit =
let worker_ops : worker_state WL.ops =
let runner (st : worker_state) = st.st.as_runner in
let around_task st = st.st.around_task in
let on_exn (st : worker_state) (ebt : Exn_bt.t) =
st.st.on_exn (Exn_bt.exn ebt) (Exn_bt.bt ebt)
in
@ -96,7 +91,6 @@ let worker_ops : worker_state WL.ops =
WL.schedule = schedule_w;
runner;
get_next_task;
around_task;
on_exn;
before_start;
cleanup;
@ -104,19 +98,11 @@ let worker_ops : worker_state WL.ops =
let create_ ?(on_init_thread = default_thread_init_exit_)
?(on_exit_thread = default_thread_init_exit_) ?(on_exn = fun _ _ -> ())
?around_task ~threads ?name () : state =
(* wrapper *)
let around_task =
match around_task with
| Some (f, g) -> WL.AT_pair (f, g)
| None -> default_around_task_
in
~threads ?name () : state =
let self =
{
threads;
q = Bb_queue.create ();
around_task;
as_runner = Runner.dummy;
name;
on_init_thread;
@ -127,8 +113,7 @@ let create_ ?(on_init_thread = default_thread_init_exit_)
self.as_runner <- runner_of_state self;
self
let create ?on_init_thread ?on_exit_thread ?on_exn ?around_task ?num_threads
?name () : t =
let create ?on_init_thread ?on_exit_thread ?on_exn ?num_threads ?name () : t =
let num_domains = Domain_pool_.max_number_of_domains () in
(* number of threads to run *)
@ -140,8 +125,7 @@ let create ?on_init_thread ?on_exit_thread ?on_exn ?around_task ?num_threads
let pool =
let dummy_thread = Thread.self () in
let threads = Array.make num_threads dummy_thread in
create_ ?on_init_thread ?on_exit_thread ?on_exn ?around_task ~threads ?name
()
create_ ?on_init_thread ?on_exit_thread ?on_exn ~threads ?name ()
in
let runner = runner_of_state pool in
@ -181,11 +165,9 @@ let create ?on_init_thread ?on_exit_thread ?on_exn ?around_task ?num_threads
runner
let with_ ?on_init_thread ?on_exit_thread ?on_exn ?around_task ?num_threads
?name () f =
let with_ ?on_init_thread ?on_exit_thread ?on_exn ?num_threads ?name () f =
let pool =
create ?on_init_thread ?on_exit_thread ?on_exn ?around_task ?num_threads
?name ()
create ?on_init_thread ?on_exit_thread ?on_exn ?num_threads ?name ()
in
let@ () = Fun.protect ~finally:(fun () -> shutdown pool) in
f pool

4
src/core/fifo_pool.mli
View file

@ -20,7 +20,6 @@ type ('a, 'b) create_args =
?on_init_thread:(dom_id:int -> t_id:int -> unit -> unit) ->
?on_exit_thread:(dom_id:int -> t_id:int -> unit -> unit) ->
?on_exn:(exn -> Printexc.raw_backtrace -> unit) ->
?around_task:(t -> 'b) * (t -> 'b -> unit) ->
?num_threads:int ->
?name:string ->
'a
@ -35,9 +34,6 @@ val create : (unit -> t, _) create_args
[Domain.recommended_domain_count()], ie one worker per CPU core. On OCaml
4 the default is [4] (since there is only one domain).
@param on_exit_thread called at the end of each worker thread in the pool.
@param around_task
a pair of [before, after] functions ran around each task. See
{!Pool.create_args}.
@param name name for the pool, used in tracing (since 0.6) *)
val with_ : (unit -> (t -> 'a) -> 'a, _) create_args

11
src/core/worker_loop_.ml
View file

@ -13,15 +13,11 @@ type task_full =
}
-> task_full
type around_task =
| AT_pair : (Runner.t -> 'a) * (Runner.t -> 'a -> unit) -> around_task
exception No_more_tasks
type 'st ops = {
schedule: 'st -> task_full -> unit;
get_next_task: 'st -> task_full; (** @raise No_more_tasks *)
around_task: 'st -> around_task;
on_exn: 'st -> Exn_bt.t -> unit;
runner: 'st -> Runner.t;
before_start: 'st -> unit;
@ -117,7 +113,6 @@ module Fine_grained (Args : FINE_GRAINED_ARGS) () = struct
let state = ref New
let run_task (task : task_full) : unit =
let (AT_pair (before_task, after_task)) = ops.around_task st in
let fiber =
match task with
| T_start { fiber; _ } | T_resume { fiber; _ } -> fiber
@ -125,7 +120,8 @@ module Fine_grained (Args : FINE_GRAINED_ARGS) () = struct
cur_fiber := fiber;
TLS.set k_cur_fiber fiber;
let _ctx = before_task runner in
(* let _ctx = before_task runner in *)
(* run the task now, catching errors, handling effects *)
assert (task != _dummy_task);
@ -140,8 +136,7 @@ module Fine_grained (Args : FINE_GRAINED_ARGS) () = struct
let ebt = Exn_bt.make e bt in
ops.on_exn st ebt);
after_task runner _ctx;
(* after_task runner _ctx; *)
cur_fiber := _dummy_fiber;
TLS.set k_cur_fiber _dummy_fiber

4
src/core/worker_loop_.mli
View file

@ -18,15 +18,11 @@ type task_full =
val _dummy_task : task_full
type around_task =
| AT_pair : (Runner.t -> 'a) * (Runner.t -> 'a -> unit) -> around_task
exception No_more_tasks
type 'st ops = {
schedule: 'st -> task_full -> unit;
get_next_task: 'st -> task_full;
around_task: 'st -> around_task;
on_exn: 'st -> Exn_bt.t -> unit;
runner: 'st -> Runner.t;
before_start: 'st -> unit;

20
src/core/ws_pool.ml
View file

@ -28,7 +28,6 @@ type state = {
cond: Condition.t;
mutable as_runner: t;
(* init options *)
around_task: WL.around_task;
name: string option;
on_init_thread: dom_id:int -> t_id:int -> unit -> unit;
on_exit_thread: dom_id:int -> t_id:int -> unit -> unit;
@ -198,7 +197,6 @@ let cleanup (self : worker_state) : unit =
let worker_ops : worker_state WL.ops =
let runner (st : worker_state) = st.st.as_runner in
let around_task st = st.st.around_task in
let on_exn (st : worker_state) (ebt : Exn_bt.t) =
st.st.on_exn (Exn_bt.exn ebt) (Exn_bt.bt ebt)
in
@ -206,7 +204,6 @@ let worker_ops : worker_state WL.ops =
WL.schedule = schedule_from_w;
runner;
get_next_task;
around_task;
on_exn;
before_start;
cleanup;
@ -235,7 +232,6 @@ type ('a, 'b) create_args =
?on_init_thread:(dom_id:int -> t_id:int -> unit -> unit) ->
?on_exit_thread:(dom_id:int -> t_id:int -> unit -> unit) ->
?on_exn:(exn -> Printexc.raw_backtrace -> unit) ->
?around_task:(t -> 'b) * (t -> 'b -> unit) ->
?num_threads:int ->
?name:string ->
'a
@ -243,15 +239,8 @@ type ('a, 'b) create_args =
let create ?(on_init_thread = default_thread_init_exit_)
?(on_exit_thread = default_thread_init_exit_) ?(on_exn = fun _ _ -> ())
?around_task ?num_threads ?name () : t =
?num_threads ?name () : t =
let pool_id_ = Id.create () in
(* wrapper *)
let around_task =
match around_task with
| Some (f, g) -> WL.AT_pair (f, g)
| None -> WL.AT_pair (ignore, fun _ _ -> ())
in
let num_domains = Domain_pool_.max_number_of_domains () in
let num_threads = Util_pool_.num_threads ?num_threads () in
@ -268,7 +257,6 @@ let create ?(on_init_thread = default_thread_init_exit_)
n_waiting_nonzero = true;
mutex = Mutex.create ();
cond = Condition.create ();
around_task;
on_exn;
on_init_thread;
on_exit_thread;
@ -324,11 +312,9 @@ let create ?(on_init_thread = default_thread_init_exit_)
pool.as_runner
let with_ ?on_init_thread ?on_exit_thread ?on_exn ?around_task ?num_threads
?name () f =
let with_ ?on_init_thread ?on_exit_thread ?on_exn ?num_threads ?name () f =
let pool =
create ?on_init_thread ?on_exit_thread ?on_exn ?around_task ?num_threads
?name ()
create ?on_init_thread ?on_exit_thread ?on_exn ?num_threads ?name ()
in
let@ () = Fun.protect ~finally:(fun () -> shutdown pool) in
f pool

6
src/core/ws_pool.mli
View file

@ -24,7 +24,6 @@ type ('a, 'b) create_args =
?on_init_thread:(dom_id:int -> t_id:int -> unit -> unit) ->
?on_exit_thread:(dom_id:int -> t_id:int -> unit -> unit) ->
?on_exn:(exn -> Printexc.raw_backtrace -> unit) ->
?around_task:(t -> 'b) * (t -> 'b -> unit) ->
?num_threads:int ->
?name:string ->
'a
@ -40,11 +39,6 @@ val create : (unit -> t, _) create_args
[Domain.recommended_domain_count()], ie one worker thread per CPU core. On
OCaml 4 the default is [4] (since there is only one domain).
@param on_exit_thread called at the end of each thread in the pool
@param around_task
a pair of [before, after], where [before pool] is called before a task is
processed, on the worker thread about to run it, and returns [x]; and
[after pool x] is called by the same thread after the task is over. (since
0.2)
@param name
a name for this thread pool, used if tracing is enabled (since 0.6) *)

14
src/lwt/moonpool_lwt.ml
View file

@ -7,8 +7,6 @@ end
module Fut = Moonpool.Fut
let default_around_task_ : WL.around_task = AT_pair (ignore, fun _ _ -> ())
let on_uncaught_exn : (Moonpool.Exn_bt.t -> unit) ref =
ref (fun ebt ->
Printf.eprintf "uncaught exception in moonpool-lwt:\n%s" (Exn_bt.show ebt))
@ -90,8 +88,6 @@ end
module Ops = struct
type st = Scheduler_state.st
let around_task _ = default_around_task_
let schedule (self : st) t =
if Atomic.get self.closed then
failwith "moonpool-lwt.schedule: scheduler is closed";
@ -122,15 +118,7 @@ module Ops = struct
()
let ops : st WL.ops =
{
schedule;
around_task;
get_next_task;
on_exn;
runner;
before_start;
cleanup;
}
{ schedule; get_next_task; on_exn; runner; before_start; cleanup }
let setup st =
if Atomic.compare_and_set Scheduler_state.cur_st None (Some st) then