src/Pure/Concurrent/future.ML

 
 (** scheduling **)
 
 (* global state *)
 
-val queue = ref Task_Queue.empty;
-val next = ref 0;
-val workers = ref ([]: (Thread.thread * bool) list);
-val scheduler = ref (NONE: Thread.thread option);
-val excessive = ref 0;
-val canceled = ref ([]: Task_Queue.group list);
-val do_shutdown = ref false;
+val queue = Unsynchronized.ref Task_Queue.empty;
+val next = Unsynchronized.ref 0;
+val workers = Unsynchronized.ref ([]: (Thread.thread * bool) list);
+val scheduler = Unsynchronized.ref (NONE: Thread.thread option);
+val excessive = Unsynchronized.ref 0;
+val canceled = Unsynchronized.ref ([]: Task_Queue.group list);
+val do_shutdown = Unsynchronized.ref false;
 
 
 (* synchronization *)
 
 val scheduler_event = ConditionVar.conditionVar ();

         | Exn.Result _ => true)
       end;
   in (result, job) end;
 
 fun do_cancel group = (*requires SYNCHRONIZED*)
- (change canceled (insert Task_Queue.eq_group group); broadcast scheduler_event);
+ (Unsynchronized.change canceled (insert Task_Queue.eq_group group);
+  broadcast scheduler_event);
 
 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 "finish" (fn () =>
       let
-        val maximal = change_result queue (Task_Queue.finish task);
+        val maximal = Unsynchronized.change_result queue (Task_Queue.finish task);
         val _ =
           if ok then ()
           else if Task_Queue.cancel (! queue) group then ()
           else do_cancel group;
         val _ = broadcast work_finished;

 
 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));
+  Unsynchronized.change workers
+    (AList.update Thread.equal (Thread.self (), active));
 
 
 (* worker threads *)
 
 fun worker_wait cond = (*requires SYNCHRONIZED*)
   (change_active false; wait cond; change_active true);
 
 fun worker_next () = (*requires SYNCHRONIZED*)
   if ! excessive > 0 then
-    (dec excessive;
-     change workers (filter_out (fn (thread, _) => Thread.equal (thread, Thread.self ())));
+    (Unsynchronized.dec excessive;
+     Unsynchronized.change workers
+      (filter_out (fn (thread, _) => Thread.equal (thread, Thread.self ())));
      broadcast scheduler_event;
      NONE)
   else if count_active () > Multithreading.max_threads_value () then
     (worker_wait scheduler_event; worker_next ())
   else
-    (case change_result queue (Task_Queue.dequeue (Thread.self ())) of
+    (case Unsynchronized.change_result queue (Task_Queue.dequeue (Thread.self ())) of
       NONE => (worker_wait work_available; worker_next ())
     | some => some);
 
 fun worker_loop name =
   (case SYNCHRONIZED name (fn () => worker_next ()) of
     NONE => ()
   | SOME work => (execute name work; worker_loop name));
 
 fun worker_start name = (*requires SYNCHRONIZED*)
-  change workers (cons (SimpleThread.fork false (fn () =>
+  Unsynchronized.change workers (cons (SimpleThread.fork false (fn () =>
      (broadcast scheduler_event; worker_loop name)), true));
 
 
 (* scheduler *)
 
-val last_status = ref Time.zeroTime;
+val last_status = Unsynchronized.ref Time.zeroTime;
 val next_status = Time.fromMilliseconds 500;
 val next_round = Time.fromMilliseconds 50;
 
 fun scheduler_next () = (*requires SYNCHRONIZED*)
   let

     val mm = if m = 9999 then 1 else m * 2;
     val l = length (! workers);
     val _ = excessive := l - mm;
     val _ =
       if mm > l then
-        funpow (mm - l) (fn () => worker_start ("worker " ^ string_of_int (inc next))) ()
+        funpow (mm - l) (fn () =>
+          worker_start ("worker " ^ string_of_int (Unsynchronized.inc next))) ()
       else ();
 
     (*canceled groups*)
     val _ =
       if null (! canceled) then ()
       else
        (Multithreading.tracing 1 (fn () =>
           string_of_int (length (! canceled)) ^ " canceled groups");
-        change canceled (filter_out (Task_Queue.cancel (! queue)));
+        Unsynchronized.change canceled (filter_out (Task_Queue.cancel (! queue)));
         broadcast_work ());
 
     (*delay loop*)
     val _ = Exn.release (wait_timeout next_round scheduler_event);
 

         SOME group => group
       | NONE => Task_Queue.new_group (worker_group ()));
     val (result, job) = future_job group e;
     val task = SYNCHRONIZED "enqueue" (fn () =>
       let
-        val (task, minimal) = change_result queue (Task_Queue.enqueue group deps pri job);
+        val (task, minimal) =
+          Unsynchronized.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 Future {task = task, group = group, result = result} end;
 

   Synchronized.guarded_access (result_of x) (fn NONE => NONE | some => SOME ((), some));
 
 fun join_next deps = (*requires SYNCHRONIZED*)
   if null deps then NONE
   else
-    (case change_result queue (Task_Queue.dequeue_towards (Thread.self ()) deps) of
+    (case Unsynchronized.change_result queue (Task_Queue.dequeue_towards (Thread.self ()) deps) of
       (NONE, []) => NONE
     | (NONE, deps') => (worker_wait work_finished; join_next deps')
     | (SOME work, deps') => SOME (work, deps'));
 
 fun join_work deps =