--- a/etc/options Wed May 09 22:11:02 2018 +0100
+++ b/etc/options Wed May 09 23:31:11 2018 +0100
@@ -71,6 +71,8 @@
option threads_stack_limit : real = 0.25
-- "maximum stack size for worker threads (in giga words, 0 = unlimited)"
+public option parallel_limit : int = 0
+ -- "approximative limit for parallel tasks (0 = unlimited)"
public option parallel_print : bool = true
-- "parallel and asynchronous printing of results"
public option parallel_proofs : int = 1
--- a/src/HOL/ROOT Wed May 09 22:11:02 2018 +0100
+++ b/src/HOL/ROOT Wed May 09 23:31:11 2018 +0100
@@ -15,7 +15,7 @@
description {*
HOL-Main with explicit proof terms.
*}
- options [quick_and_dirty = false, record_proofs = 2, parallel_proofs = 0]
+ options [quick_and_dirty = false, record_proofs = 2, parallel_proofs = 0, parallel_limit = 500]
sessions
"HOL-Library"
theories
--- a/src/Pure/Concurrent/future.ML Wed May 09 22:11:02 2018 +0100
+++ b/src/Pure/Concurrent/future.ML Wed May 09 23:31:11 2018 +0100
@@ -34,6 +34,10 @@
val forked_results: {name: string, deps: Task_Queue.task list} ->
(unit -> 'a) list -> 'a Exn.result list
val task_context: string -> group -> ('a -> 'b) -> 'a -> 'b
+ val enabled: unit -> bool
+ val relevant: 'a list -> bool
+ val proofs_enabled: int -> bool
+ val proofs_enabled_timing: Time.time -> bool
val value_result: 'a Exn.result -> 'a future
val value: 'a -> 'a future
val cond_forks: params -> (unit -> 'a) list -> 'a future list
@@ -171,10 +175,10 @@
fun report_status () = (*requires SYNCHRONIZED*)
if ! ML_statistics then
let
- val {ready, pending, running, passive, urgent} = Task_Queue.status (! queue);
- val total = length (! workers);
- val active = count_workers Working;
- val waiting = count_workers Waiting;
+ val {ready, pending, running, passive, urgent, total} = Task_Queue.status (! queue);
+ val workers_total = length (! workers);
+ val workers_active = count_workers Working;
+ val workers_waiting = count_workers Waiting;
val stats =
[("now", Value.print_real (Time.toReal (Time.now ()))),
("tasks_ready", Value.print_int ready),
@@ -182,9 +186,10 @@
("tasks_running", Value.print_int running),
("tasks_passive", Value.print_int passive),
("tasks_urgent", Value.print_int urgent),
- ("workers_total", Value.print_int total),
- ("workers_active", Value.print_int active),
- ("workers_waiting", Value.print_int waiting)] @
+ ("tasks_total", Value.print_int total),
+ ("workers_total", Value.print_int workers_total),
+ ("workers_active", Value.print_int workers_active),
+ ("workers_waiting", Value.print_int workers_waiting)] @
ML_Statistics.get ();
in Output.try_protocol_message (Markup.ML_statistics :: stats) [] end
else ();
@@ -576,6 +581,22 @@
end);
+(* scheduling parameters *)
+
+fun enabled () =
+ Multithreading.max_threads () > 1 andalso
+ let val lim = Options.default_int "parallel_limit"
+ in lim <= 0 orelse SYNCHRONIZED "enabled" (fn () => Task_Queue.total_jobs (! queue) < lim) end;
+
+val relevant = (fn [] => false | [_] => false | _ => enabled ());
+
+fun proofs_enabled n =
+ ! Multithreading.parallel_proofs >= n andalso is_some (worker_task ()) andalso enabled ();
+
+fun proofs_enabled_timing t =
+ proofs_enabled 1 andalso Time.toReal t >= Options.default_real "parallel_subproofs_threshold";
+
+
(* fast-path operations -- bypass task queue if possible *)
fun value_result (res: 'a Exn.result) =
@@ -589,7 +610,7 @@
fun value x = value_result (Exn.Res x);
fun cond_forks args es =
- if Multithreading.enabled () then forks args es
+ if enabled () then forks args es
else map (fn e => value_result (Exn.interruptible_capture e ())) es;
fun map_future f x =
--- a/src/Pure/Concurrent/lazy.ML Wed May 09 22:11:02 2018 +0100
+++ b/src/Pure/Concurrent/lazy.ML Wed May 09 23:31:11 2018 +0100
@@ -129,7 +129,7 @@
val pending =
xs |> map_filter (fn x => if is_pending x then SOME (fn () => force_result x) else NONE);
val _ =
- if Multithreading.relevant pending then
+ if Future.relevant pending then
ignore (Future.forked_results {name = "Lazy.consolidate", deps = []} pending)
else List.app (fn e => ignore (e ())) pending;
in xs end;
--- a/src/Pure/Concurrent/multithreading.ML Wed May 09 22:11:02 2018 +0100
+++ b/src/Pure/Concurrent/multithreading.ML Wed May 09 23:31:11 2018 +0100
@@ -8,10 +8,7 @@
sig
val max_threads: unit -> int
val max_threads_update: int -> unit
- val enabled: unit -> bool
- val relevant: 'a list -> bool
val parallel_proofs: int ref
- val parallel_proofs_enabled: int -> bool
val sync_wait: Time.time option -> ConditionVar.conditionVar -> Mutex.mutex -> bool Exn.result
val trace: int ref
val tracing: int -> (unit -> string) -> unit
@@ -42,9 +39,6 @@
fun max_threads () = ! max_threads_state;
fun max_threads_update m = max_threads_state := max_threads_result m;
-fun enabled () = max_threads () > 1;
-
-val relevant = (fn [] => false | [_] => false | _ => enabled ());
end;
@@ -53,9 +47,6 @@
val parallel_proofs = ref 1;
-fun parallel_proofs_enabled n =
- enabled () andalso ! parallel_proofs >= n;
-
(* synchronous wait *)
--- a/src/Pure/Concurrent/par_list.ML Wed May 09 22:11:02 2018 +0100
+++ b/src/Pure/Concurrent/par_list.ML Wed May 09 23:31:11 2018 +0100
@@ -29,7 +29,7 @@
struct
fun forked_results name f xs =
- if Multithreading.relevant xs
+ if Future.relevant xs
then Future.forked_results {name = name, deps = []} (map (fn x => fn () => f x) xs)
else map (Exn.capture f) xs;
--- a/src/Pure/Concurrent/task_queue.ML Wed May 09 22:11:02 2018 +0100
+++ b/src/Pure/Concurrent/task_queue.ML Wed May 09 23:31:11 2018 +0100
@@ -32,7 +32,9 @@
val group_tasks: queue -> group list -> task list
val known_task: queue -> task -> bool
val all_passive: queue -> bool
- val status: queue -> {ready: int, pending: int, running: int, passive: int, urgent: int}
+ val total_jobs: queue -> int
+ val status: queue ->
+ {ready: int, pending: int, running: int, passive: int, urgent: int, total: int}
val cancel: queue -> group -> Thread.thread list
val cancel_all: queue -> group list * Thread.thread list
val finish: task -> queue -> bool * queue
@@ -217,10 +219,12 @@
(* queue *)
-datatype queue = Queue of {groups: groups, jobs: jobs, urgent: int};
+datatype queue = Queue of {groups: groups, jobs: jobs, urgent: int, total: int};
-fun make_queue groups jobs urgent = Queue {groups = groups, jobs = jobs, urgent = urgent};
-val empty = make_queue Inttab.empty Task_Graph.empty 0;
+fun make_queue groups jobs urgent total =
+ Queue {groups = groups, jobs = jobs, urgent = urgent, total = total};
+
+val empty = make_queue Inttab.empty Task_Graph.empty 0 0;
fun group_tasks (Queue {groups, ...}) gs =
fold (fn g => fn tasks => Tasks.merge (op =) (tasks, get_tasks groups (group_id g)))
@@ -249,10 +253,12 @@
fun all_passive (Queue {jobs, ...}) = is_none (Task_Graph.get_first active_job jobs);
+fun total_jobs (Queue {total, ...}) = total;
+
(* queue status *)
-fun status (Queue {jobs, urgent, ...}) =
+fun status (Queue {jobs, urgent, total, ...}) =
let
val (x, y, z, w) =
Task_Graph.fold (fn (_, (job, (deps, _))) => fn (x, y, z, w) =>
@@ -261,7 +267,7 @@
| Running _ => (x, y, z + 1, w)
| Passive _ => (x, y, z, w + 1)))
jobs (0, 0, 0, 0);
- in {ready = x, pending = y, running = z, passive = w, urgent = urgent} end;
+ in {ready = x, pending = y, running = z, passive = w, urgent = urgent, total = total} end;
@@ -295,35 +301,38 @@
(* finish *)
-fun finish task (Queue {groups, jobs, urgent}) =
+fun finish task (Queue {groups, jobs, urgent, total}) =
let
val group = group_of_task task;
val groups' = fold_groups (fn g => del_task (group_id g, task)) group groups;
val jobs' = Task_Graph.del_node task jobs;
+ val total' = total - 1;
val maximal = Task_Graph.is_maximal jobs task;
- in (maximal, make_queue groups' jobs' urgent) end;
+ in (maximal, make_queue groups' jobs' urgent total') end;
(* enroll *)
-fun enroll thread name group (Queue {groups, jobs, urgent}) =
+fun enroll thread name group (Queue {groups, jobs, urgent, total}) =
let
val task = new_task group name NONE;
val groups' = fold_groups (fn g => add_task (group_id g, task)) group groups;
val jobs' = jobs |> Task_Graph.new_node (task, Running thread);
- in (task, make_queue groups' jobs' urgent) end;
+ val total' = total + 1;
+ in (task, make_queue groups' jobs' urgent total') end;
(* enqueue *)
-fun enqueue_passive group name abort (Queue {groups, jobs, urgent}) =
+fun enqueue_passive group name abort (Queue {groups, jobs, urgent, total}) =
let
val task = new_task group name NONE;
val groups' = fold_groups (fn g => add_task (group_id g, task)) group groups;
val jobs' = jobs |> Task_Graph.new_node (task, Passive abort);
- in (task, make_queue groups' jobs' urgent) end;
+ val total' = total + 1;
+ in (task, make_queue groups' jobs' urgent total') end;
-fun enqueue name group deps pri job (Queue {groups, jobs, urgent}) =
+fun enqueue name group deps pri job (Queue {groups, jobs, urgent, total}) =
let
val task = new_task group name (SOME pri);
val groups' = fold_groups (fn g => add_task (group_id g, task)) group groups;
@@ -331,39 +340,40 @@
|> Task_Graph.new_node (task, Job [job])
|> fold (add_job task) deps;
val urgent' = if pri >= urgent_pri then urgent + 1 else urgent;
- in (task, make_queue groups' jobs' urgent') end;
+ val total' = total + 1;
+ in (task, make_queue groups' jobs' urgent' total') end;
-fun extend task job (Queue {groups, jobs, urgent}) =
+fun extend task job (Queue {groups, jobs, urgent, total}) =
(case try (get_job jobs) task of
- SOME (Job list) => SOME (make_queue groups (set_job task (Job (job :: list)) jobs) urgent)
+ SOME (Job list) => SOME (make_queue groups (set_job task (Job (job :: list)) jobs) urgent total)
| _ => NONE);
(* dequeue *)
-fun dequeue_passive thread task (queue as Queue {groups, jobs, urgent}) =
+fun dequeue_passive thread task (queue as Queue {groups, jobs, urgent, total}) =
(case try (get_job jobs) task of
SOME (Passive _) =>
let val jobs' = set_job task (Running thread) jobs
- in (SOME true, make_queue groups jobs' urgent) end
+ in (SOME true, make_queue groups jobs' urgent total) end
| SOME _ => (SOME false, queue)
| NONE => (NONE, queue));
-fun dequeue thread urgent_only (queue as Queue {groups, jobs, urgent}) =
+fun dequeue thread urgent_only (queue as Queue {groups, jobs, urgent, total}) =
if not urgent_only orelse urgent > 0 then
(case Task_Graph.get_first (ready_job_urgent urgent_only) jobs of
SOME (result as (task, _)) =>
let
val jobs' = set_job task (Running thread) jobs;
val urgent' = if pri_of_task task >= urgent_pri then urgent - 1 else urgent;
- in (SOME result, make_queue groups jobs' urgent') end
+ in (SOME result, make_queue groups jobs' urgent' total) end
| NONE => (NONE, queue))
else (NONE, queue);
(* dequeue wrt. dynamic dependencies *)
-fun dequeue_deps thread deps (queue as Queue {groups, jobs, urgent}) =
+fun dequeue_deps thread deps (queue as Queue {groups, jobs, urgent, total}) =
let
fun ready [] rest = (NONE, rev rest)
| ready (task :: tasks) rest =
@@ -388,7 +398,7 @@
let
val jobs' = set_job task (Running thread) jobs;
val urgent' = if pri_of_task task >= urgent_pri then urgent - 1 else urgent;
- in ((SOME res, deps'), make_queue groups jobs' urgent') end;
+ in ((SOME res, deps'), make_queue groups jobs' urgent' total) end;
in
(case ready deps [] of
(SOME res, deps') => result res deps'
--- a/src/Pure/General/exn.scala Wed May 09 22:11:02 2018 +0100
+++ b/src/Pure/General/exn.scala Wed May 09 23:31:11 2018 +0100
@@ -31,13 +31,11 @@
def error(message: String): Nothing = throw ERROR(message)
- def cat_message(msg1: String, msg2: String): String =
- if (msg1 == "") msg2
- else if (msg2 == "") msg1
- else msg1 + "\n" + msg2
+ def cat_message(msgs: String*): String =
+ cat_lines(msgs.iterator.filterNot(_ == ""))
- def cat_error(msg1: String, msg2: String): Nothing =
- error(cat_message(msg1, msg2))
+ def cat_error(msgs: String*): Nothing =
+ error(cat_message(msgs:_*))
/* exceptions as values */
--- a/src/Pure/Isar/proof.ML Wed May 09 22:11:02 2018 +0100
+++ b/src/Pure/Isar/proof.ML Wed May 09 23:31:11 2018 +0100
@@ -1301,7 +1301,7 @@
local
fun future_terminal_proof proof1 proof2 done state =
- if Goal.future_enabled 3 andalso not (is_relevant state) then
+ if Future.proofs_enabled 3 andalso not (is_relevant state) then
state |> future_proof (fn state' =>
let val pos = Position.thread_data () in
Execution.fork {name = "Proof.future_terminal_proof", pos = pos, pri = ~1}
--- a/src/Pure/Isar/toplevel.ML Wed May 09 22:11:02 2018 +0100
+++ b/src/Pure/Isar/toplevel.ML Wed May 09 23:31:11 2018 +0100
@@ -668,19 +668,19 @@
let val trs = tl (Thy_Syntax.flat_element elem)
in fold (fn tr => fn t => get_timing tr + t) trs Time.zeroTime end;
-fun proof_future_enabled estimate st =
+fun future_proofs_enabled estimate st =
(case try proof_of st of
NONE => false
| SOME state =>
not (Proof.is_relevant state) andalso
(if can (Proof.assert_bottom true) state
- then Goal.future_enabled 1
- else Goal.future_enabled 2 orelse Goal.future_enabled_timing estimate));
+ then Future.proofs_enabled 1
+ else Future.proofs_enabled 2 orelse Future.proofs_enabled_timing estimate));
fun atom_result keywords tr st =
let
val st' =
- if Goal.future_enabled 1 andalso Keyword.is_diag keywords (name_of tr) then
+ if Future.proofs_enabled 1 andalso Keyword.is_diag keywords (name_of tr) then
(Execution.fork
{name = "Toplevel.diag", pos = pos_of tr, pri = ~1}
(fn () => command tr st); st)
@@ -696,7 +696,7 @@
val (body_elems, end_tr) = element_rest;
val estimate = timing_estimate elem;
in
- if not (proof_future_enabled estimate st')
+ if not (future_proofs_enabled estimate st')
then
let
val proof_trs = maps Thy_Syntax.flat_element body_elems @ [end_tr];
--- a/src/Pure/ML/ml_compiler.ML Wed May 09 22:11:02 2018 +0100
+++ b/src/Pure/ML/ml_compiler.ML Wed May 09 23:31:11 2018 +0100
@@ -120,7 +120,7 @@
|> map (fn (pos, markup, text) => Position.reported_text pos (markup ()) (text ()))
|> Output.report;
val _ =
- if not (null persistent_reports) andalso redirect andalso Multithreading.enabled ()
+ if not (null persistent_reports) andalso redirect andalso Future.enabled ()
then
Execution.print
{name = "ML_Compiler.report", pos = Position.thread_data (), pri = Task_Queue.urgent_pri}
--- a/src/Pure/ML/ml_statistics.scala Wed May 09 22:11:02 2018 +0100
+++ b/src/Pure/ML/ml_statistics.scala Wed May 09 23:31:11 2018 +0100
@@ -39,7 +39,8 @@
val tasks_fields: Fields =
("Future tasks",
- List("tasks_ready", "tasks_pending", "tasks_running", "tasks_passive", "tasks_urgent"))
+ List("tasks_ready", "tasks_pending", "tasks_running", "tasks_passive",
+ "tasks_urgent", "tasks_total"))
val workers_fields: Fields =
("Worker threads", List("workers_total", "workers_active", "workers_waiting"))
--- a/src/Pure/PIDE/command.ML Wed May 09 22:11:02 2018 +0100
+++ b/src/Pure/PIDE/command.ML Wed May 09 23:31:11 2018 +0100
@@ -190,7 +190,7 @@
end) ();
fun run keywords int tr st =
- if Goal.future_enabled 1 andalso Keyword.is_diag keywords (Toplevel.name_of tr) then
+ if Future.proofs_enabled 1 andalso Keyword.is_diag keywords (Toplevel.name_of tr) then
(Execution.fork {name = "Toplevel.diag", pos = Toplevel.pos_of tr, pri = ~1}
(fn () => Toplevel.command_exception int tr st); ([], SOME st))
else Toplevel.command_errors int tr st;
@@ -429,7 +429,7 @@
fun run_print execution_id (Print {name, delay, pri, exec_id, print_process, ...}) =
if ignore_process print_process then ()
- else if pri <= 0 orelse (Multithreading.enabled () andalso Options.default_bool "parallel_print")
+ else if pri <= 0 orelse (Future.enabled () andalso Options.default_bool "parallel_print")
then
let
val group = Future.worker_subgroup ();
--- a/src/Pure/PIDE/execution.ML Wed May 09 22:11:02 2018 +0100
+++ b/src/Pure/PIDE/execution.ML Wed May 09 23:31:11 2018 +0100
@@ -162,7 +162,7 @@
fun fork_prints exec_id =
(case Inttab.lookup (#2 (get_state ())) exec_id of
SOME (_, prints) =>
- if Multithreading.relevant prints then
+ if Future.relevant prints then
let val pos = Position.thread_data () in
List.app (fn {name, pri, body} =>
ignore (fork {name = name, pos = pos, pri = pri} body)) (rev prints)
--- a/src/Pure/ROOT.scala Wed May 09 22:11:02 2018 +0100
+++ b/src/Pure/ROOT.scala Wed May 09 23:31:11 2018 +0100
@@ -8,7 +8,7 @@
{
val ERROR = Exn.ERROR
val error = Exn.error _
- val cat_error = Exn.cat_error _
+ def cat_error(msgs: String*): Nothing = Exn.cat_error(msgs:_*)
def using[A <: { def close() }, B](x: A)(f: A => B): B = Library.using(x)(f)
val space_explode = Library.space_explode _
--- a/src/Pure/Thy/present.scala Wed May 09 22:11:02 2018 +0100
+++ b/src/Pure/Thy/present.scala Wed May 09 23:31:11 2018 +0100
@@ -272,6 +272,7 @@
if (!result.ok) {
cat_error(
+ Library.trim_line(result.err),
cat_lines(Latex.latex_errors(dir, root_name) ::: Bibtex.bibtex_errors(dir, root_name)),
"Failed to build document in " + File.path(dir.absolute_file))
}
--- a/src/Pure/Thy/thy_info.ML Wed May 09 22:11:02 2018 +0100
+++ b/src/Pure/Thy/thy_info.ML Wed May 09 23:31:11 2018 +0100
@@ -388,7 +388,7 @@
{document = document, symbols = symbols, bibtex_entries = bibtex_entries,
last_timing = last_timing};
val (_, tasks) = require_thys context [] qualifier master_dir imports String_Graph.empty;
- in if Multithreading.enabled () then schedule_futures tasks else schedule_seq tasks end;
+ in if Multithreading.max_threads () > 1 then schedule_futures tasks else schedule_seq tasks end;
fun use_thy name =
use_theories
--- a/src/Pure/goal.ML Wed May 09 22:11:02 2018 +0100
+++ b/src/Pure/goal.ML Wed May 09 23:31:11 2018 +0100
@@ -23,8 +23,6 @@
val finish: Proof.context -> thm -> thm
val norm_result: Proof.context -> thm -> thm
val skip_proofs_enabled: unit -> bool
- val future_enabled: int -> bool
- val future_enabled_timing: Time.time -> bool
val future_result: Proof.context -> thm future -> term -> thm
val prove_internal: Proof.context -> cterm list -> cterm -> (thm list -> tactic) -> thm
val is_schematic: term -> bool
@@ -111,14 +109,6 @@
else skip
end;
-fun future_enabled n =
- Multithreading.parallel_proofs_enabled n andalso
- is_some (Future.worker_task ());
-
-fun future_enabled_timing t =
- future_enabled 1 andalso
- Time.toReal t >= Options.default_real "parallel_subproofs_threshold";
-
(* future_result *)
@@ -176,7 +166,7 @@
val schematic = exists is_schematic props;
val immediate = is_none fork_pri;
- val future = future_enabled 1;
+ val future = Future.proofs_enabled 1;
val skip = not immediate andalso not schematic andalso future andalso skip_proofs_enabled ();
val pos = Position.thread_data ();
@@ -258,7 +248,7 @@
fun prove_sorry ctxt xs asms prop tac =
if Config.get ctxt quick_and_dirty then
prove ctxt xs asms prop (fn _ => ALLGOALS (Skip_Proof.cheat_tac ctxt))
- else (if future_enabled 1 then prove_future_pri ctxt ~2 else prove ctxt) xs asms prop tac;
+ else (if Future.proofs_enabled 1 then prove_future_pri ctxt ~2 else prove ctxt) xs asms prop tac;
fun prove_sorry_global thy xs asms prop tac =
Drule.export_without_context
--- a/src/Pure/par_tactical.ML Wed May 09 22:11:02 2018 +0100
+++ b/src/Pure/par_tactical.ML Wed May 09 23:31:11 2018 +0100
@@ -46,7 +46,7 @@
Drule.strip_imp_prems (Thm.cprop_of st)
|> map (Thm.adjust_maxidx_cterm ~1);
in
- if Multithreading.relevant goals andalso forall (fn g => Thm.maxidx_of_cterm g = ~1) goals then
+ if Future.relevant goals andalso forall (fn g => Thm.maxidx_of_cterm g = ~1) goals then
let
fun try_goal g =
(case SINGLE tac (Goal.init g) of