[scheduler] delete unused tasks continuations

Summary: This was dead code.

Reviewed By: mbouaziz

Differential Revision: D8477026

fbshipit-source-id: 5496e7c
master
Jules Villard 7 years ago committed by Facebook Github Bot
parent 25627fd4d9
commit ea734a3e77

@ -16,7 +16,6 @@ let create_exe_env_tasks source_file exe_env : Tasks.t =
Summary.clear_cache () ; Summary.clear_cache () ;
Typ.Procname.SQLite.clear_cache () ; Typ.Procname.SQLite.clear_cache () ;
Random.self_init () ; Random.self_init () ;
Tasks.create
[ (fun () -> [ (fun () ->
Callbacks.iterate_callbacks exe_env ; Callbacks.iterate_callbacks exe_env ;
if Config.write_html then Printer.write_all_html_files source_file ) ] if Config.write_html then Printer.write_all_html_files source_file ) ]

@ -8,32 +8,18 @@
open! IStd open! IStd
module L = Logging module L = Logging
type closure = unit -> unit type task = unit -> unit
type t = {closures: closure list; continuations: closure Queue.t} type t = task list
let create ?(continuation= None) closures = (** Aggregate closures into groups of the given size *)
let continuations =
match continuation with None -> Queue.create () | Some closure -> Queue.singleton closure
in
{closures; continuations}
(* Aggregate closures into groups of the given size *)
let aggregate ~size t = let aggregate ~size t =
let group_to_closure group () = List.iter ~f:(fun closure -> closure ()) group in let group_to_closure group () = List.iter ~f:(fun closure -> closure ()) group in
let group_size = if size > 0 then size else List.length t.closures / Config.jobs in let group_size = if size > 0 then size else List.length t / Config.jobs in
if group_size > 1 then if group_size > 1 then List.chunks_of t ~length:group_size |> List.map ~f:group_to_closure else t
let groups = List.groupi ~break:(fun n _ _ -> Int.equal (n mod group_size) 0) t.closures in
let closures = List.map ~f:group_to_closure groups in
{t with closures}
else t
let run t = let run t = List.iter ~f:(fun f -> f ()) t
List.iter ~f:(fun f -> f ()) t.closures ;
Queue.iter ~f:(fun closure -> closure ()) t.continuations
let fork_protect ~f x = let fork_protect ~f x =
EventLogger.prepare () ; EventLogger.prepare () ;
@ -43,21 +29,18 @@ let fork_protect ~f x =
module Runner = struct module Runner = struct
type runner = {pool: ProcessPool.t; all_continuations: closure Queue.t} type tasks = t
type t = {pool: ProcessPool.t}
let create ~jobs = {pool= ProcessPool.create ~jobs; all_continuations= Queue.create ()} let create ~jobs = {pool= ProcessPool.create ~jobs}
let start runner ~tasks = let start runner ~tasks =
let pool = runner.pool in let pool = runner.pool in
Queue.enqueue_all runner.all_continuations (Queue.to_list tasks.continuations) ;
(* Flush here all buffers to avoid passing unflushed data to forked processes, leading to duplication *) (* Flush here all buffers to avoid passing unflushed data to forked processes, leading to duplication *)
Pervasives.flush_all () ; Pervasives.flush_all () ;
List.iter List.iter ~f:(fun x -> ProcessPool.start_child ~f:(fun f -> fork_protect ~f ()) ~pool x) tasks
~f:(fun x -> ProcessPool.start_child ~f:(fun f -> fork_protect ~f ()) ~pool x)
tasks.closures
let complete runner = let complete runner = ProcessPool.wait_all runner.pool
ProcessPool.wait_all runner.pool ;
Queue.iter ~f:(fun f -> f ()) runner.all_continuations
end end

@ -7,37 +7,33 @@
open! IStd open! IStd
(** A sequence of tasks that can be executed in parallel, (** Each task executes a closure *)
with a continuation to be executed at the end *) type task = unit -> unit
type t
(** Each task/continuation executes a closure *) (** A sequence of tasks that can be executed in parallel *)
type closure = unit -> unit type t = task list
(* Aggregate closures into groups of the given size *)
val aggregate : size:int -> t -> t val aggregate : size:int -> t -> t
(** Aggregate closures into groups of the given size *)
val create : ?continuation:closure option -> closure list -> t
(** Create tasks with a list of closures to be executed in parallel,
and an optional continuation to be executed afterwards *)
val run : t -> unit val run : t -> unit
(** Run the closures and continuation *) (** Run the closures *)
val fork_protect : f:('a -> 'b) -> 'a -> 'b val fork_protect : f:('a -> 'b) -> 'a -> 'b
(** does the bookkeeping necessary to safely execute an infer function [f] after a call to fork(2) *) (** does the bookkeeping necessary to safely execute an infer function [f] after a call to fork(2) *)
(** A runner accepts new tasks repeatedly for parallel execution *)
module Runner : sig module Runner : sig
(** A runner accepts new tasks repeatedly for parallel execution *) type tasks = t
type runner
type t
val create : jobs:int -> runner val create : jobs:int -> t
(** Create a runner *) (** Create a runner running [jobs] jobs in parallel *)
val start : runner -> tasks:t -> unit val start : t -> tasks:tasks -> unit
(** Start the given tasks with the runner *) (** Start the given tasks with the runner *)
val complete : runner -> unit val complete : t -> unit
(** Complete all the outstanding tasks *) (** Wait until all the outstanding tasks are completed *)
end end

Loading…
Cancel
Save