src/Pure/Concurrent/future.ML

 (*  Title:      Pure/Concurrent/future.ML
     Author:     Makarius
 
-Future values.
+Future values, see also
+http://www4.in.tum.de/~wenzelm/papers/parallel-isabelle.pdf
 
 Notes:
 
   * Futures are similar to delayed evaluation, i.e. delay/force is
     generalized to fork/join (and variants).  The idea is to model

   ConditionVar.broadcast work_finished);
 
 end;
 
 
-(* worker activity *)
-
-fun count_active ws =
-  fold (fn (_, active) => fn i => if active then i + 1 else i) ws 0;
-
-fun change_active active = (*requires SYNCHRONIZED*)
-  change workers (AList.update Thread.equal (Thread.self (), active));
-
-fun overloaded () =
-  count_active (! workers) > Multithreading.max_threads_value ();
-
-
 (* execute future jobs *)
 
 fun future_job group (e: unit -> 'a) =
   let
     val result = ref (NONE: 'a Exn.result option);

 fun execute name (task, group, jobs) =
   let
     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 () =>
+    val _ = SYNCHRONIZED "finish" (fn () =>
       let
         val maximal = change_result queue (Task_Queue.finish task);
         val _ =
           if ok then ()
           else if Task_Queue.cancel (! queue) group then ()

         val _ = if maximal then () else broadcast work_available;
       in () end);
   in () end;
 
 
+(* worker activity *)
+
+fun count_active () = (*requires SYNCHRONIZED*)
+  fold (fn (_, active) => fn i => if active then i + 1 else i) (! workers) 0;
+
+fun change_active active = (*requires SYNCHRONIZED*)
+  change workers (AList.update Thread.equal (Thread.self (), active));
+
+
 (* worker threads *)
 
 fun worker_wait cond = (*requires SYNCHRONIZED*)
  (change_active false; broadcast scheduler_event;
   wait cond;

   if ! excessive > 0 then
     (dec excessive;
      change workers (filter_out (fn (thread, _) => Thread.equal (thread, Thread.self ())));
      broadcast scheduler_event;
      NONE)
-  else if overloaded () then (worker_wait scheduler_event; worker_next ())
+  else if count_active () > Multithreading.max_threads_value () then
+    (worker_wait scheduler_event; worker_next ())
   else
     (case change_result queue Task_Queue.dequeue of
       NONE => (worker_wait work_available; worker_next ())
     | some => some);
 

         else
          (last_status := now; Multithreading.tracing 1 (fn () =>
             let
               val {ready, pending, running} = Task_Queue.status (! queue);
               val total = length (! workers);
-              val active = count_active (! workers);
+              val active = count_active ();
             in
               "SCHEDULE: " ^
                 string_of_int ready ^ " ready, " ^
                 string_of_int pending ^ " pending, " ^
                 string_of_int running ^ " running; " ^

     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 () =>
+    val task = SYNCHRONIZED "enqueue" (fn () =>
       let
         val (task, minimal) = change_result queue (Task_Queue.enqueue group deps pri job);
         val _ = if minimal then signal work_available else ();
         val _ = scheduler_check ();
       in task end);

 
 in
 
 fun join_results xs =
   if forall is_finished xs then map get_result xs
+  else if Multithreading.self_critical () then
+    error "Cannot join future values within critical section"
   else uninterruptible (fn _ => fn () =>
-    let
-      val _ = Multithreading.self_critical () andalso
-        error "Cannot join future values within critical section";
-      val _ =
-        if is_worker () then join_work (map task_of xs)
-        else List.app join_wait xs;
-    in map get_result xs end) ();
+     (if is_worker ()
+      then join_work (map task_of xs)
+      else List.app join_wait xs;
+      map get_result xs)) ();
 
 end;
 
 fun join_result x = singleton join_results x;
 fun join x = Exn.release (join_result x);

   let
     val task = task_of x;
     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 () =>
+    val extended = SYNCHRONIZED "extend" (fn () =>
       (case Task_Queue.extend task job (! queue) of
         SOME queue' => (queue := queue'; true)
       | NONE => false));
   in
     if extended then Future {task = task, group = group, result = result}