--- a/src/Pure/Concurrent/future.ML Wed Feb 02 13:44:40 2011 +0100
+++ b/src/Pure/Concurrent/future.ML Wed Feb 02 15:04:09 2011 +0100
@@ -32,19 +32,16 @@
signature FUTURE =
sig
- type task = Task_Queue.task
- type group = Task_Queue.group
- val is_worker: unit -> bool
val worker_task: unit -> Task_Queue.task option
val worker_group: unit -> Task_Queue.group option
val worker_subgroup: unit -> Task_Queue.group
type 'a future
- val task_of: 'a future -> task
- val group_of: 'a future -> group
+ val task_of: 'a future -> Task_Queue.task
val peek: 'a future -> 'a Exn.result option
val is_finished: 'a future -> bool
- val forks: {name: string, group: group option, deps: task list, pri: int} ->
- (unit -> 'a) list -> 'a future list
+ val forks:
+ {name: string, group: Task_Queue.group option, deps: Task_Queue.task list, pri: int} ->
+ (unit -> 'a) list -> 'a future list
val fork_pri: int -> (unit -> 'a) -> 'a future
val fork: (unit -> 'a) -> 'a future
val join_results: 'a future list -> 'a Exn.result list
@@ -52,12 +49,12 @@
val join: 'a future -> 'a
val value: 'a -> 'a future
val map: ('a -> 'b) -> 'a future -> 'b future
- val promise_group: group -> 'a future
+ val promise_group: Task_Queue.group -> 'a future
val promise: unit -> 'a future
val fulfill_result: 'a future -> 'a Exn.result -> unit
val fulfill: 'a future -> 'a -> unit
val interruptible_task: ('a -> 'b) -> 'a -> 'b
- val cancel_group: group -> unit
+ val cancel_group: Task_Queue.group -> unit
val cancel: 'a future -> unit
val shutdown: unit -> unit
val status: (unit -> 'a) -> 'a
@@ -70,20 +67,15 @@
(* identifiers *)
-type task = Task_Queue.task;
-type group = Task_Queue.group;
-
local
- val tag = Universal.tag () : (task * group) option Universal.tag;
+ val tag = Universal.tag () : Task_Queue.task option Universal.tag;
in
- fun thread_data () = the_default NONE (Thread.getLocal tag);
- fun setmp_thread_data data f x =
- Library.setmp_thread_data tag (thread_data ()) (SOME data) f x;
+ fun worker_task () = the_default NONE (Thread.getLocal tag);
+ fun setmp_worker_task data f x =
+ Library.setmp_thread_data tag (worker_task ()) (SOME data) f x;
end;
-val is_worker = is_some o thread_data;
-val worker_task = Option.map #1 o thread_data;
-val worker_group = Option.map #2 o thread_data;
+val worker_group = Option.map Task_Queue.group_of_task o worker_task;
fun worker_subgroup () = Task_Queue.new_group (worker_group ());
fun worker_joining e =
@@ -103,12 +95,10 @@
datatype 'a future = Future of
{promised: bool,
- task: task,
- group: group,
+ task: Task_Queue.task,
result: 'a result};
fun task_of (Future {task, ...}) = task;
-fun group_of (Future {group, ...}) = group;
fun result_of (Future {result, ...}) = result;
fun peek x = Single_Assignment.peek (result_of x);
@@ -204,12 +194,13 @@
(Unsynchronized.change canceled (insert Task_Queue.eq_group group);
broadcast scheduler_event);
-fun execute (task, group, jobs) =
+fun execute (task, jobs) =
let
+ val group = Task_Queue.group_of_task task;
val valid = not (Task_Queue.is_canceled group);
val ok =
Task_Queue.running task (fn () =>
- setmp_thread_data (task, group) (fn () =>
+ setmp_worker_task task (fn () =>
fold (fn job => fn ok => job valid andalso ok) jobs true) ());
val _ = Multithreading.tracing 1 (fn () =>
let
@@ -416,7 +407,7 @@
let
val (result, job) = future_job grp e;
val ((task, minimal'), queue') = Task_Queue.enqueue name grp deps pri job queue;
- val future = Future {promised = false, task = task, group = grp, result = result};
+ val future = Future {promised = false, task = task, result = result};
in (future, (minimal orelse minimal', queue')) end;
in
SYNCHRONIZED "enqueue" (fn () =>
@@ -443,7 +434,7 @@
NONE => Exn.Exn (Fail "Unfinished future")
| SOME res =>
if Exn.is_interrupt_exn res then
- (case Exn.flatten_list (Task_Queue.group_status (group_of x)) of
+ (case Exn.flatten_list (Task_Queue.group_status (Task_Queue.group_of_task (task_of x))) of
[] => res
| exns => Exn.Exn (Exn.EXCEPTIONS exns))
else res);
@@ -470,7 +461,7 @@
if forall is_finished xs then ()
else if Multithreading.self_critical () then
error "Cannot join future values within critical section"
- else if is_some (thread_data ()) then
+ else if is_some (worker_task ()) then
join_work (Task_Queue.init_deps (map task_of xs))
else List.app (ignore o Single_Assignment.await o result_of) xs;
in map get_result xs end;
@@ -485,15 +476,16 @@
fun value (x: 'a) =
let
- val group = Task_Queue.new_group NONE;
+ val task = Task_Queue.dummy_task ();
+ val group = Task_Queue.group_of_task task;
val result = Single_Assignment.var "value" : 'a result;
val _ = assign_result group result (Exn.Result x);
- in Future {promised = false, task = Task_Queue.dummy_task, group = group, result = result} end;
+ in Future {promised = false, task = task, result = result} end;
fun map_future f x =
let
val task = task_of x;
- val group = Task_Queue.new_group (SOME (group_of x));
+ val group = Task_Queue.new_group (SOME (Task_Queue.group_of_task task));
val (result, job) = future_job group (fn () => f (join x));
val extended = SYNCHRONIZED "extend" (fn () =>
@@ -501,7 +493,7 @@
SOME queue' => (queue := queue'; true)
| NONE => false));
in
- if extended then Future {promised = false, task = task, group = group, result = result}
+ if extended then Future {promised = false, task = task, result = result}
else
singleton
(forks {name = "Future.map", group = SOME group,
@@ -522,14 +514,15 @@
else reraise exn;
val task = SYNCHRONIZED "enqueue_passive" (fn () =>
Unsynchronized.change_result queue (Task_Queue.enqueue_passive group abort));
- in Future {promised = true, task = task, group = group, result = result} end;
+ in Future {promised = true, task = task, result = result} end;
fun promise () = promise_group (worker_subgroup ());
-fun fulfill_result (Future {promised, task, group, result}) res =
+fun fulfill_result (Future {promised, task, result}) res =
if not promised then raise Fail "Not a promised future"
else
let
+ val group = Task_Queue.group_of_task task;
fun job ok = assign_result group result (if ok then res else Exn.interrupt_exn);
val _ =
Multithreading.with_attributes Multithreading.no_interrupts (fn _ =>
@@ -538,7 +531,7 @@
SYNCHRONIZED "fulfill_result" (fn () =>
Unsynchronized.change_result queue
(Task_Queue.dequeue_passive (Thread.self ()) task));
- in if still_passive then execute (task, group, [job]) else () end);
+ in if still_passive then execute (task, [job]) else () end);
val _ =
worker_waiting (Task_Queue.init_deps [task])
(fn () => Single_Assignment.await result);
@@ -552,7 +545,7 @@
fun interruptible_task f x =
if Multithreading.available then
Multithreading.with_attributes
- (if is_worker ()
+ (if is_some (worker_task ())
then Multithreading.private_interrupts
else Multithreading.public_interrupts)
(fn _ => f x)
@@ -563,7 +556,7 @@
(if cancel_now group then () else cancel_later group;
signal work_available; scheduler_check ()));
-fun cancel x = cancel_group (group_of x);
+fun cancel x = cancel_group (Task_Queue.group_of_task (task_of x));
(* shutdown *)
--- a/src/Pure/Concurrent/task_queue.ML Wed Feb 02 13:44:40 2011 +0100
+++ b/src/Pure/Concurrent/task_queue.ML Wed Feb 02 15:04:09 2011 +0100
@@ -6,12 +6,6 @@
signature TASK_QUEUE =
sig
- type task
- val dummy_task: task
- val name_of_task: task -> string
- val pri_of_task: task -> int
- val str_of_task: task -> string
- val timing_of_task: task -> Time.time * Time.time * string list
type group
val new_group: group option -> group
val group_id: group -> int
@@ -20,6 +14,13 @@
val is_canceled: group -> bool
val group_status: group -> exn list
val str_of_group: group -> string
+ type task
+ val dummy_task: unit -> task
+ val group_of_task: task -> group
+ val name_of_task: task -> string
+ val pri_of_task: task -> int
+ val str_of_task: task -> string
+ val timing_of_task: task -> Time.time * Time.time * string list
type queue
val empty: queue
val all_passive: queue -> bool
@@ -32,12 +33,12 @@
queue -> (task * bool) * queue
val extend: task -> (bool -> bool) -> queue -> queue option
val dequeue_passive: Thread.thread -> task -> queue -> bool * queue
- val dequeue: Thread.thread -> queue -> (task * group * (bool -> bool) list) option * queue
+ val dequeue: Thread.thread -> queue -> (task * (bool -> bool) list) option * queue
type deps
val init_deps: task list -> deps
val finished_deps: deps -> bool
val dequeue_deps: Thread.thread -> deps -> queue ->
- (((task * group * (bool -> bool) list) option * deps) * queue)
+ (((task * (bool -> bool) list) option * deps) * queue)
val running: task -> (unit -> 'a) -> 'a
val joining: task -> (unit -> 'a) -> 'a
val waiting: task -> deps -> (unit -> 'a) -> 'a
@@ -49,51 +50,9 @@
val new_id = Synchronized.counter ();
-(** grouped tasks **)
-
-(* tasks *)
-
-type timing = Time.time * Time.time * string list; (*run, wait, wait dependencies*)
-
-fun new_timing () =
- Synchronized.var "timing" ((Time.zeroTime, Time.zeroTime, []): timing);
-
-abstype task = Task of
- {name: string,
- id: int,
- pri: int option,
- timing: timing Synchronized.var}
-with
-
-val dummy_task = Task {name = "", id = ~1, pri = NONE, timing = new_timing ()};
-fun new_task name pri = Task {name = name, id = new_id (), pri = pri, timing = new_timing ()};
+(** nested groups of tasks **)
-fun name_of_task (Task {name, ...}) = name;
-fun pri_of_task (Task {pri, ...}) = the_default 0 pri;
-fun str_of_task (Task {name, id, ...}) =
- if name = "" then string_of_int id else string_of_int id ^ " (" ^ name ^ ")";
-
-fun timing_of_task (Task {timing, ...}) = Synchronized.value timing;
-
-fun update_timing update (Task {timing, ...}) e =
- let
- val start = Time.now ();
- val result = Exn.capture e ();
- val t = Time.- (Time.now (), start);
- val _ = Synchronized.change timing (update t);
- in Exn.release result end;
-
-fun task_ord (Task {id = id1, pri = pri1, ...}, Task {id = id2, pri = pri2, ...}) =
- prod_ord (rev_order o option_ord int_ord) int_ord ((pri1, id1), (pri2, id2));
-
-val eq_task = is_equal o task_ord;
-
-end;
-
-structure Task_Graph = Graph(type key = task val ord = task_ord);
-
-
-(* nested groups *)
+(* groups *)
abstype group = Group of
{parent: group option,
@@ -134,6 +93,54 @@
end;
+(* tasks *)
+
+type timing = Time.time * Time.time * string list; (*run, wait, wait dependencies*)
+
+fun new_timing () =
+ Synchronized.var "timing" ((Time.zeroTime, Time.zeroTime, []): timing);
+
+abstype task = Task of
+ {group: group,
+ name: string,
+ id: int,
+ pri: int option,
+ timing: timing Synchronized.var}
+with
+
+fun dummy_task () =
+ Task {group = new_group NONE, name = "", id = 0, pri = NONE, timing = new_timing ()};
+
+fun new_task group name pri =
+ Task {group = group, name = name, id = new_id (), pri = pri, timing = new_timing ()};
+
+fun group_of_task (Task {group, ...}) = group;
+fun name_of_task (Task {name, ...}) = name;
+fun pri_of_task (Task {pri, ...}) = the_default 0 pri;
+fun str_of_task (Task {name, id, ...}) =
+ if name = "" then string_of_int id else string_of_int id ^ " (" ^ name ^ ")";
+
+fun timing_of_task (Task {timing, ...}) = Synchronized.value timing;
+
+fun update_timing update (Task {timing, ...}) e =
+ let
+ val start = Time.now ();
+ val result = Exn.capture e ();
+ val t = Time.- (Time.now (), start);
+ val _ = Synchronized.change timing (update t);
+ in Exn.release result end;
+
+fun task_ord (Task {id = id1, pri = pri1, ...}, Task {id = id2, pri = pri2, ...}) =
+ prod_ord (rev_order o option_ord int_ord) int_ord ((pri1, id1), (pri2, id2));
+
+val eq_task = is_equal o task_ord;
+
+end;
+
+structure Task_Graph = Graph(type key = task val ord = task_ord);
+
+
+
(** queue of jobs and groups **)
(* jobs *)
@@ -143,11 +150,10 @@
Running of Thread.thread |
Passive of unit -> bool;
-type jobs = (group * job) Task_Graph.T;
+type jobs = job Task_Graph.T;
-fun get_group (jobs: jobs) task = #1 (Task_Graph.get_node jobs task);
-fun get_job (jobs: jobs) task = #2 (Task_Graph.get_node jobs task);
-fun set_job task job (jobs: jobs) = Task_Graph.map_node task (fn (group, _) => (group, job)) jobs;
+fun get_job (jobs: jobs) task = Task_Graph.get_node jobs task;
+fun set_job task job (jobs: jobs) = Task_Graph.map_node task (K job) jobs;
fun add_job task dep (jobs: jobs) =
Task_Graph.add_edge (dep, task) jobs handle Task_Graph.UNDEF _ => jobs;
@@ -169,18 +175,18 @@
(* job status *)
-fun ready_job task ((group, Job list), ([], _)) = SOME (task, group, rev list)
- | ready_job task ((group, Passive abort), ([], _)) =
- if is_canceled group then SOME (task, group, [fn _ => abort ()])
+fun ready_job task (Job list, ([], _)) = SOME (task, rev list)
+ | ready_job task (Passive abort, ([], _)) =
+ if is_canceled (group_of_task task) then SOME (task, [fn _ => abort ()])
else NONE
| ready_job _ _ = NONE;
-fun active_job (_, Job _) = SOME ()
- | active_job (_, Running _) = SOME ()
- | active_job (group, Passive _) = if is_canceled group then SOME () else NONE;
+fun active_job (_, (Job _, _)) = SOME ()
+ | active_job (_, (Running _, _)) = SOME ()
+ | active_job (task, (Passive _, _)) =
+ if is_canceled (group_of_task task) then SOME () else NONE;
-fun all_passive (Queue {jobs, ...}) =
- is_none (Task_Graph.get_first (active_job o #1 o #2) jobs);
+fun all_passive (Queue {jobs, ...}) = is_none (Task_Graph.get_first active_job jobs);
(* queue status *)
@@ -188,7 +194,7 @@
fun status (Queue {jobs, ...}) =
let
val (x, y, z, w) =
- Task_Graph.fold (fn (_, ((_, job), (deps, _))) => fn (x, y, z, w) =>
+ Task_Graph.fold (fn (_, (job, (deps, _))) => fn (x, y, z, w) =>
(case job of
Job _ => if null deps then (x + 1, y, z, w) else (x, y + 1, z, w)
| Running _ => (x, y, z + 1, w)
@@ -213,12 +219,16 @@
fun cancel_all (Queue {jobs, ...}) =
let
- fun cancel_job (group, job) (groups, running) =
- (cancel_group group Exn.Interrupt;
+ fun cancel_job (task, (job, _)) (groups, running) =
+ let
+ val group = group_of_task task;
+ val _ = cancel_group group Exn.Interrupt;
+ in
(case job of
Running t => (insert eq_group group groups, insert Thread.equal t running)
- | _ => (groups, running)));
- val (running_groups, running) = Task_Graph.fold (cancel_job o #1 o #2) jobs ([], []);
+ | _ => (groups, running))
+ end;
+ val (running_groups, running) = Task_Graph.fold cancel_job jobs ([], []);
val _ = List.app Simple_Thread.interrupt running;
in running_groups end;
@@ -227,7 +237,7 @@
fun finish task (Queue {groups, jobs}) =
let
- val group = get_group jobs task;
+ val group = group_of_task task;
val groups' = groups
|> fold (fn gid => Inttab.remove_list eq_task (gid, task)) (group_ancestry group);
val jobs' = Task_Graph.del_node task jobs;
@@ -239,19 +249,19 @@
fun enqueue_passive group abort (Queue {groups, jobs}) =
let
- val task = new_task "passive" NONE;
+ val task = new_task group "passive" NONE;
val groups' = groups
|> fold (fn gid => Inttab.cons_list (gid, task)) (group_ancestry group);
- val jobs' = jobs |> Task_Graph.new_node (task, (group, Passive abort));
+ val jobs' = jobs |> Task_Graph.new_node (task, Passive abort);
in (task, make_queue groups' jobs') end;
fun enqueue name group deps pri job (Queue {groups, jobs}) =
let
- val task = new_task name (SOME pri);
+ val task = new_task group name (SOME pri);
val groups' = groups
|> fold (fn gid => Inttab.cons_list (gid, task)) (group_ancestry group);
val jobs' = jobs
- |> Task_Graph.new_node (task, (group, Job [job]))
+ |> Task_Graph.new_node (task, Job [job])
|> fold (add_job task) deps
|> fold (fold (add_job task) o get_deps jobs) deps;
val minimal = null (get_deps jobs' task);
@@ -274,7 +284,7 @@
fun dequeue thread (queue as Queue {groups, jobs}) =
(case Task_Graph.get_first (uncurry ready_job) jobs of
- SOME (result as (task, _, _)) =>
+ SOME (result as (task, _)) =>
let val jobs' = set_job task (Running thread) jobs
in (SOME result, make_queue groups jobs') end
| NONE => (NONE, queue));
@@ -294,7 +304,7 @@
let
fun ready task = ready_job task (Task_Graph.get_entry jobs task);
val tasks = filter (can (Task_Graph.get_node jobs)) deps;
- fun result (res as (task, _, _)) =
+ fun result (res as (task, _)) =
let val jobs' = set_job task (Running thread) jobs
in ((SOME res, Deps tasks), make_queue groups jobs') end;
in