src/Pure/Concurrent/task_queue.ML

   val pri_of_task: task -> int
   val str_of_task: task -> string
   type group
   val group_id: group -> int
   val eq_group: group * group -> bool
-  val new_group: unit -> group
-  val group_exns: group -> exn list
+  val new_group: group option -> group
+  val group_status: group -> exn list
   val str_of_group: group -> string
   type queue
   val empty: queue
   val is_empty: queue -> bool
   val status: queue -> {ready: int, pending: int, running: int}

   val dequeue: queue -> (task * group * (bool -> bool) list) option * queue
   val dequeue_towards: task list -> queue ->
     (((task * group * (bool -> bool) list) * task list) option * queue)
   val interrupt: queue -> task -> unit
   val interrupt_external: queue -> string -> unit
+  val is_canceled: group -> bool
   val cancel_group: group -> exn -> unit
   val cancel: queue -> group -> bool
   val cancel_all: queue -> group list
   val finish: task -> queue -> queue
 end;

 
 fun task_ord (Task t1, Task t2) = prod_ord (rev_order o int_ord) int_ord (t1, t2);
 structure Task_Graph = Graph(type key = task val ord = task_ord);
 
 
-(* groups *)
-
-datatype group = Group of serial * exn list ref;
-
-fun group_id (Group (gid, _)) = gid;
-fun eq_group (Group (gid1, _), Group (gid2, _)) = gid1 = gid2;
-
-fun new_group () = Group (serial (), ref []);
-
-fun group_exns (Group (_, ref exns)) = exns;
-
-fun str_of_group (Group (i, ref exns)) =
-  if null exns then string_of_int i else enclose "(" ")" (string_of_int i);
+(* nested groups *)
+
+datatype group = Group of
+ {parent: group option,
+  id: serial,
+  status: exn list ref};
+
+fun make_group (parent, id, status) = Group {parent = parent, id = id, status = status};
+
+fun new_group parent = make_group (parent, serial (), ref []);
+
+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);
+
+
+fun cancel_group (Group {status, ...}) exn = CRITICAL (fn () =>
+  (case exn of
+    Exn.Interrupt => if null (! status) then status := [exn] else ()
+  | _ => change status (cons exn)));
+
+fun group_status (Group {parent, status, ...}) = (*non-critical*)
+  ! status @ (case parent of NONE => [] | SOME group => group_status group);
+
+fun is_canceled (Group {parent, status, ...}) = (*non-critical*)
+  not (null (! status)) orelse (case parent of NONE => false | SOME group => is_canceled group);
+
+fun str_of_group group =
+  (is_canceled group ? enclose "(" ")") (string_of_int (group_id group));
 
 
 (* jobs *)
 
 datatype job =

 
 val empty = make_queue Inttab.empty Task_Graph.empty No_Result;
 fun is_empty (Queue {jobs, ...}) = Task_Graph.is_empty jobs;
 
 
-(* status *)
+(* queue status *)
 
 fun status (Queue {jobs, ...}) =
   let
     val (x, y, z) =
       Task_Graph.fold (fn (task, ((_, job), (deps, _))) => fn (x, y, z) =>

           | Running _ => (x, y, z + 1)))
         jobs (0, 0, 0);
   in {ready = x, pending = y, running = z} end;
 
 
+(* cancel -- peers and sub-groups *)
+
+fun cancel (Queue {groups, jobs, ...}) group =
+  let
+    val _ = cancel_group group Exn.Interrupt;
+    val tasks = Inttab.lookup_list groups (group_id group);
+    val running =
+      fold (get_job jobs #> (fn Running t => insert Thread.equal t | _ => I)) tasks [];
+    val _ = List.app SimpleThread.interrupt running;
+  in null running end;
+
+fun cancel_all (Queue {jobs, ...}) =
+  let
+    fun cancel_job (group, job) (groups, running) =
+      (cancel_group group Exn.Interrupt;
+        (case job of Running t => (insert eq_group group groups, insert Thread.equal t running)
+        | _ => (groups, running)));
+    val (groups, running) = Task_Graph.fold (cancel_job o #1 o #2) jobs ([], []);
+    val _ = List.app SimpleThread.interrupt running;
+  in groups end;
+
+
 (* enqueue *)
 
-fun enqueue (group as Group (gid, _)) deps pri job (Queue {groups, jobs, cache}) =
+fun enqueue group deps pri job (Queue {groups, jobs, cache}) =
   let
     val task = new_task pri;
-    val groups' = Inttab.cons_list (gid, task) groups;
+    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])) |> fold (add_job task) deps;
+      |> Task_Graph.new_node (task, (group, Job [job]))
+      |> fold (add_job task) deps;
     val cache' =
       (case cache of
         Result last =>
           if task_ord (last, task) = LESS
           then cache else Unknown

 fun dequeue_towards deps (queue as Queue {groups, jobs, ...}) =
   let
     fun ready task =
       (case Task_Graph.get_node jobs task of
         (group, Job list) =>
-          if null (Task_Graph.imm_preds jobs task) then SOME (task, group, rev list)
+          if null (Task_Graph.imm_preds jobs task)
+          then SOME (task, group, rev list)
           else NONE
       | _ => NONE);
 
     val tasks = filter (can (Task_Graph.get_node jobs)) deps;
     fun result (res as (task, _, _)) =

 
 
 (* sporadic interrupts *)
 
 fun interrupt (Queue {jobs, ...}) task =
-  (case try (get_job jobs) task of SOME (Running thread) => SimpleThread.interrupt thread | _ => ());
+  (case try (get_job jobs) task of
+    SOME (Running thread) => SimpleThread.interrupt thread
+  | _ => ());
 
 fun interrupt_external (queue as Queue {jobs, ...}) str =
   (case Int.fromString str of
     SOME i =>
       (case Task_Graph.get_first NONE

   | NONE => ());
 
 
 (* termination *)
 
-fun cancel_group (Group (_, r)) exn = CRITICAL (fn () =>
-  (case exn of
-    Exn.Interrupt => if null (! r) then r := [exn] else ()
-  | _ => change r (cons exn)));
-
-fun cancel (Queue {groups, jobs, ...}) (group as Group (gid, _)) =
-  let
-    val _ = cancel_group group Exn.Interrupt;
-    val tasks = Inttab.lookup_list groups gid;
-    val running = fold (get_job jobs #> (fn Running t => insert Thread.equal t | _ => I)) tasks [];
-    val _ = List.app SimpleThread.interrupt running;
-  in null running end;
-
-fun cancel_all (Queue {jobs, ...}) =
-  let
-    fun cancel_job (group, job) (groups, running) =
-      (cancel_group group Exn.Interrupt;
-        (case job of Running t => (insert eq_group group groups, insert Thread.equal t running)
-        | _ => (groups, running)));
-    val (groups, running) = Task_Graph.fold (cancel_job o #1 o #2) jobs ([], []);
-    val _ = List.app SimpleThread.interrupt running;
-  in groups end;
-
 fun finish task (Queue {groups, jobs, cache}) =
   let
-    val Group (gid, _) = get_group jobs task;
-    val groups' = Inttab.remove_list (op =) (gid, task) groups;
+    val group = get_group jobs task;
+    val groups' = groups
+      |> fold (fn gid => Inttab.remove_list (op =) (gid, task)) (group_ancestry group);
     val jobs' = Task_Graph.del_node task jobs;
     val cache' =
       if null (Task_Graph.imm_succs jobs task) then cache
       else Unknown;
   in make_queue groups' jobs' cache' end;