src/Pure/Concurrent/task_queue.ML

 Ordered queue of grouped tasks.
 *)
 
 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
   val eq_group: group * group -> bool
   val cancel_group: group -> exn -> unit
   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
   val status: queue -> {ready: int, pending: int, running: int, passive: int}
   val cancel: queue -> group -> bool

   val enqueue_passive: group -> (unit -> bool) -> queue -> task * queue
   val enqueue: string -> group -> task list -> int -> (bool -> bool) ->
     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
 end;
 

 struct
 
 val new_id = Synchronized.counter ();
 
 
-(** grouped tasks **)
+(** nested groups of tasks **)
+
+(* groups *)
+
+abstype group = Group of
+ {parent: group option,
+  id: int,
+  status: exn list Synchronized.var}
+with
+
+fun make_group (parent, id, status) = Group {parent = parent, id = id, status = status};
+
+fun new_group parent = make_group (parent, new_id (), Synchronized.var "group" []);
+
+fun group_id (Group {id, ...}) = id;
+fun eq_group (group1, group2) = group_id group1 = group_id group2;
+
+fun group_ancestry (Group {parent, id, ...}) =
+  id :: (case parent of NONE => [] | SOME group => group_ancestry group);
+
+
+(* group status *)
+
+fun cancel_group (Group {status, ...}) exn =
+  Synchronized.change status
+    (fn exns =>
+      if not (Exn.is_interrupt exn) orelse null exns then exn :: exns
+      else exns);
+
+fun is_canceled (Group {parent, status, ...}) =
+  not (null (Synchronized.value status)) orelse
+    (case parent of NONE => false | SOME group => is_canceled group);
+
+fun group_status (Group {parent, status, ...}) =
+  Synchronized.value status @
+    (case parent of NONE => [] | SOME group => group_status group);
+
+fun str_of_group group =
+  (is_canceled group ? enclose "(" ")") (string_of_int (group_id group));
+
+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
- {name: string,
+ {group: group,
+  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 ()};
-
+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 ^ ")";
 

 end;
 
 structure Task_Graph = Graph(type key = task val ord = task_ord);
 
 
-(* nested groups *)
-
-abstype group = Group of
- {parent: group option,
-  id: int,
-  status: exn list Synchronized.var}
-with
-
-fun make_group (parent, id, status) = Group {parent = parent, id = id, status = status};
-
-fun new_group parent = make_group (parent, new_id (), Synchronized.var "group" []);
-
-fun group_id (Group {id, ...}) = id;
-fun eq_group (group1, group2) = group_id group1 = group_id group2;
-
-fun group_ancestry (Group {parent, id, ...}) =
-  id :: (case parent of NONE => [] | SOME group => group_ancestry group);
-
-
-(* group status *)
-
-fun cancel_group (Group {status, ...}) exn =
-  Synchronized.change status
-    (fn exns =>
-      if not (Exn.is_interrupt exn) orelse null exns then exn :: exns
-      else exns);
-
-fun is_canceled (Group {parent, status, ...}) =
-  not (null (Synchronized.value status)) orelse
-    (case parent of NONE => false | SOME group => is_canceled group);
-
-fun group_status (Group {parent, status, ...}) =
-  Synchronized.value status @
-    (case parent of NONE => [] | SOME group => group_status group);
-
-fun str_of_group group =
-  (is_canceled group ? enclose "(" ")") (string_of_int (group_id group));
-
-end;
-
 
 (** queue of jobs and groups **)
 
 (* jobs *)
 
 datatype job =
   Job of (bool -> bool) list |
   Running of Thread.thread |
   Passive of unit -> bool;
 
-type jobs = (group * 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;
+type jobs = job Task_Graph.T;
+
+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;
 
 fun get_deps (jobs: jobs) task =

 val empty = make_queue Inttab.empty Task_Graph.empty;
 
 
 (* 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 all_passive (Queue {jobs, ...}) =
-  is_none (Task_Graph.get_first (active_job o #1 o #2) jobs);
+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 jobs);
 
 
 (* queue status *)
 
 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)
           | Passive _ => (x, y, z, w + 1)))
         jobs (0, 0, 0, 0);

     val _ = List.app Simple_Thread.interrupt running;
   in null running end;
 
 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;
 
 
 (* finish *)
 
 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;
     val maximal = null (Task_Graph.imm_succs jobs task);
   in (maximal, make_queue groups' jobs') end;

 
 (* enqueue *)
 
 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);
   in ((task, minimal), make_queue groups' jobs') end;
 

       in (true, make_queue groups jobs') end
   | _ => (false, queue));
 
 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));
 
 

 
 fun dequeue_deps thread (Deps deps) (queue as Queue {groups, jobs}) =
   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
     (case get_first ready tasks of
       SOME res => result res