src/Pure/Concurrent/future.ML

 sig
   val enabled: unit -> bool
   type task = Task_Queue.task
   type group = Task_Queue.group
   val is_worker: unit -> bool
+  val worker_group: unit -> Task_Queue.group option
   type 'a future
   val task_of: 'a future -> task
   val group_of: 'a future -> group
   val peek: 'a future -> 'a Exn.result option
   val is_finished: 'a future -> bool
   val value: 'a -> 'a future
   val fork: (unit -> 'a) -> 'a future
   val fork_group: group -> (unit -> 'a) -> 'a future
   val fork_deps: 'b future list -> (unit -> 'a) -> 'a future
   val fork_pri: int -> (unit -> 'a) -> 'a future
-  val fork_local: int -> (unit -> 'a) -> 'a future
   val join_results: 'a future list -> 'a Exn.result list
   val join_result: 'a future -> 'a Exn.result
   val join: 'a future -> 'a
   val map: ('a -> 'b) -> 'a future -> 'b future
   val interruptible_task: ('a -> 'b) -> 'a -> 'b

   fun setmp_thread_data data f x =
     Library.setmp_thread_data tag (thread_data ()) (SOME data) f x;
 end;
 
 val is_worker = is_some o thread_data;
+val worker_group = Option.map #3 o thread_data;
 
 
 (* datatype future *)
 
 datatype 'a future = Future of

 fun peek (Future {result, ...}) = ! result;
 fun is_finished x = is_some (peek x);
 
 fun value x = Future
  {task = Task_Queue.new_task 0,
-  group = Task_Queue.new_group (),
+  group = Task_Queue.new_group NONE,
   result = ref (SOME (Exn.Result x))};
 
 
 
 (** scheduling **)

   change canceled (insert Task_Queue.eq_group group);
 
 fun execute name (task, group, jobs) =
   let
     val _ = trace_active ();
-    val valid = null (Task_Queue.group_exns group);
+    val valid = not (Task_Queue.is_canceled group);
     val ok = setmp_thread_data (name, task, group) (fn () =>
       fold (fn job => fn ok => job valid andalso ok) jobs true) ();
     val _ = SYNCHRONIZED "execute" (fn () =>
      (change queue (Task_Queue.finish task);
       if ok then ()

 
 fun fork_future opt_group deps pri e =
   let
     val _ = scheduler_check "future check";
 
-    val group = (case opt_group of SOME group => group | NONE => Task_Queue.new_group ());
+    val group =
+      (case opt_group of
+        SOME group => group
+      | NONE => Task_Queue.new_group (worker_group ()));
     val (result, job) = future_job group e;
     val task = SYNCHRONIZED "future" (fn () =>
       change_result queue (Task_Queue.enqueue group deps pri job) before notify_all ());
   in Future {task = task, group = group, result = result} end;
 
 fun fork e = fork_future NONE [] 0 e;
 fun fork_group group e = fork_future (SOME group) [] 0 e;
 fun fork_deps deps e = fork_future NONE (map task_of deps) 0 e;
 fun fork_pri pri e = fork_future NONE [] pri e;
-fun fork_local pri e = fork_future (Option.map #3 (thread_data ())) [] pri e;
 
 
 (* join *)
 
 local
 
 fun get_result x =
   (case peek x of
-    SOME (Exn.Exn Exn.Interrupt) => Exn.Exn (Exn.EXCEPTIONS (Task_Queue.group_exns (group_of x)))
-  | SOME res => res
-  | NONE => Exn.Exn (SYS_ERROR "unfinished future"));
+    NONE => Exn.Exn (SYS_ERROR "unfinished future")
+  | SOME (Exn.Exn Exn.Interrupt) =>
+      Exn.Exn (Exn.EXCEPTIONS (Exn.flatten_list (Task_Queue.group_status (group_of x))))
+  | SOME res => res);
 
 fun join_next deps = (*requires SYNCHRONIZED*)
   if overloaded () then (worker_wait (); join_next deps)
   else change_result queue (Task_Queue.dequeue_towards deps);
 

 fun map_future f x =
   let
     val _ = scheduler_check "map_future check";
 
     val task = task_of x;
-    val group = Task_Queue.new_group ();
+    val group = Task_Queue.new_group (SOME (group_of x));
     val (result, job) = future_job group (fn () => f (join x));
 
     val extended = SYNCHRONIZED "map_future" (fn () =>
       (case Task_Queue.extend task job (! queue) of
         SOME queue' => (queue := queue'; true)