src/Pure/Concurrent/future.ML
author wenzelm
Tue Sep 09 23:30:05 2008 +0200 (2008-09-09 ago)
changeset 28186 6a8417f36837
parent 28177 8c0335bc9336
child 28191 9e5f556409c6
permissions -rw-r--r--
cancel: check_scheduler;
adapted to simplified TaskQueue.cancel;
improved join/join_all: actively work towards results, i.e. do not yield unnecessarily;
misc tuning;
     1 (*  Title:      Pure/Concurrent/future.ML
     2     ID:         $Id$
     3     Author:     Makarius
     4 
     5 Functional threads as future values.
     6 *)
     7 
     8 signature FUTURE =
     9 sig
    10   type task = TaskQueue.task
    11   type group = TaskQueue.group
    12   type 'a T
    13   val task_of: 'a T -> task
    14   val group_of: 'a T -> group
    15   val shutdown_request: unit -> unit
    16   val future: bool -> task list -> (unit -> 'a) -> 'a T
    17   val fork: (unit -> 'a) -> 'a T
    18   val cancel: 'a T -> unit
    19   val join_all: 'a T list -> 'a list
    20   val join: 'a T -> 'a
    21 end;
    22 
    23 structure Future: FUTURE =
    24 struct
    25 
    26 (** future values **)
    27 
    28 (* identifiers *)
    29 
    30 type task = TaskQueue.task;
    31 type group = TaskQueue.group;
    32 
    33 local val tag = Universal.tag () : (task * group) option Universal.tag in
    34   fun thread_data () = the_default NONE (Thread.getLocal tag);
    35   fun set_thread_data x = Thread.setLocal (tag, x);
    36 end;
    37 
    38 
    39 (* datatype future *)
    40 
    41 datatype 'a T = Future of
    42  {task: task,
    43   group: group,
    44   result: 'a Exn.result option ref};
    45 
    46 fun task_of (Future {task, ...}) = task;
    47 fun group_of (Future {group, ...}) = group;
    48 
    49 
    50 
    51 (** scheduling **)
    52 
    53 (* global state *)
    54 
    55 val queue = ref TaskQueue.empty;
    56 val workers = ref ([]: Thread.thread list);
    57 val scheduler = ref (NONE: Thread.thread option);
    58 
    59 val excessive = ref 0;
    60 val active = ref 0;
    61 
    62 fun trace_active () =
    63   Multithreading.tracing 1 (fn () => "SCHEDULE: " ^ string_of_int (! active) ^ " active");
    64 
    65 
    66 (* requests *)
    67 
    68 datatype request = Shutdown | Cancel of group;
    69 val requests = Mailbox.create () : request Mailbox.T;
    70 
    71 fun shutdown_request () = Mailbox.send requests Shutdown;
    72 fun cancel_request group = Mailbox.send requests (Cancel group);
    73 
    74 
    75 (* synchronization *)
    76 
    77 local
    78   val lock = Mutex.mutex ();
    79   val cond = ConditionVar.conditionVar ();
    80 in
    81 
    82 fun SYNCHRONIZED e = uninterruptible (fn restore_attributes => fn () =>
    83   let
    84     val _ = Mutex.lock lock;
    85     val result = Exn.capture (restore_attributes e) ();
    86     val _ = Mutex.unlock lock;
    87   in Exn.release result end) ();
    88 
    89 fun wait name = (*requires SYNCHRONIZED*)
    90   let
    91     val _ = Multithreading.tracing 4 (fn () => name ^ " : waiting");
    92     val _ = ConditionVar.wait (cond, lock);
    93     val _ = Multithreading.tracing 4 (fn () => name ^ " : notified");
    94   in () end;
    95 
    96 fun notify_all () = (*requires SYNCHRONIZED*)
    97   ConditionVar.broadcast cond;
    98 
    99 end;
   100 
   101 
   102 (* execute *)
   103 
   104 fun execute name (task, group, run) =
   105   let
   106     val _ = set_thread_data (SOME (task, group));
   107     val _ = Multithreading.tracing 4 (fn () => name ^ ": running");
   108     val ok = run ();
   109     val _ = Multithreading.tracing 4 (fn () => name ^ ": finished");
   110     val _ = set_thread_data NONE;
   111     val _ = SYNCHRONIZED (fn () =>
   112      (change queue (TaskQueue.finish task);
   113       if ok then ()
   114       else if change_result queue (TaskQueue.cancel group) then ()
   115       else cancel_request group;
   116       notify_all ()));
   117   in () end;
   118 
   119 
   120 (* worker threads *)
   121 
   122 fun change_active b = (*requires SYNCHRONIZED*)
   123   (change active (fn n => if b then n + 1 else n - 1); trace_active ());
   124 
   125 fun worker_wait name = (*requires SYNCHRONIZED*)
   126   (change_active false; wait name; change_active true);
   127 
   128 fun worker_next name = (*requires SYNCHRONIZED*)
   129   if ! excessive > 0 then
   130     (dec excessive;
   131      change_active false;
   132      change workers (remove Thread.equal (Thread.self ()));
   133      NONE)
   134   else
   135     (case change_result queue TaskQueue.dequeue of
   136       NONE => (worker_wait name; worker_next name)
   137     | some => some);
   138 
   139 fun worker_loop name =
   140   (case SYNCHRONIZED (fn () => worker_next name) of
   141     NONE => ()
   142   | SOME work => (execute name work; worker_loop name));
   143 
   144 fun worker_start name = (*requires SYNCHRONIZED*)
   145  (change_active true;
   146   change workers (cons (Thread.fork (fn () => worker_loop name, Multithreading.no_interrupts))));
   147 
   148 
   149 (* scheduler *)
   150 
   151 fun scheduler_fork () = SYNCHRONIZED (fn () =>
   152   let
   153     val _ = trace_active ();
   154     val m = Multithreading.max_threads_value ();
   155     val l = length (! workers);
   156     val _ = excessive := l - m;
   157   in List.app (fn i => worker_start ("worker " ^ string_of_int i)) (l upto m - 1) end);
   158 
   159 fun scheduler_loop canceled =
   160   let
   161     val canceled' = SYNCHRONIZED (fn () =>
   162       filter_out (change_result queue o TaskQueue.cancel) canceled);
   163     val _ = scheduler_fork ();
   164   in
   165     (case Mailbox.receive_timeout (Time.fromSeconds 1) requests of
   166       SOME Shutdown => ()   (* FIXME proper worker shutdown *)
   167     | SOME (Cancel group) => scheduler_loop (group :: canceled')
   168     | NONE => scheduler_loop canceled')
   169   end;
   170 
   171 fun check_scheduler () = SYNCHRONIZED (fn () =>
   172   if (case ! scheduler of NONE => false | SOME thread => Thread.isActive thread) then ()
   173   else scheduler := SOME (Thread.fork (fn () => scheduler_loop [], Multithreading.no_interrupts)));
   174 
   175 
   176 (* future values *)
   177 
   178 fun future new_group deps (e: unit -> 'a) =
   179   let
   180     val _ = check_scheduler ();
   181 
   182     val group =
   183       (case (new_group, thread_data ()) of
   184         (false, SOME (_, group)) => group
   185       | _ => TaskQueue.new_group ());
   186 
   187     val result = ref (NONE: 'a Exn.result option);
   188     val run = Multithreading.with_attributes (Thread.getAttributes ())
   189       (fn _ => fn ok =>
   190         let val res = if ok then Exn.capture e () else Exn.Exn Interrupt
   191         in result := SOME res; is_some (Exn.get_result res) end);
   192 
   193     val task = SYNCHRONIZED (fn () =>
   194       change_result queue (TaskQueue.enqueue group deps run) before notify_all ());
   195   in Future {task = task, group = group, result = result} end;
   196 
   197 fun fork e = future false [] e;
   198 
   199 fun cancel x = (check_scheduler (); cancel_request (group_of x));
   200 
   201 
   202 (* join *)
   203 
   204 fun join_all xs =
   205   let
   206     val _ = check_scheduler ();
   207 
   208     fun unfinished () =
   209       xs |> map_filter (fn Future {task, result = ref NONE, ...} => SOME task | _ => NONE);
   210 
   211     (*alien thread -- refrain from contending for resources*)
   212     fun passive_join () = (*requires SYNCHRONIZED*)
   213       (case unfinished () of [] => ()
   214       | _ => (wait "join"; passive_join ()));
   215 
   216     (*proper worker thread -- actively work towards results*)
   217     fun active_join () = (*requires SYNCHRONIZED*)
   218       (case unfinished () of [] => ()
   219       | tasks =>
   220           (case change_result queue (TaskQueue.dequeue_towards tasks) of
   221             NONE => (worker_wait "join"; active_join ())
   222           | SOME work => (execute "join" work; active_join ())));
   223 
   224     val _ =
   225       (case thread_data () of
   226         NONE => SYNCHRONIZED passive_join
   227       | SOME (task, _) => SYNCHRONIZED (fn () =>
   228          (change queue (TaskQueue.depend (unfinished ()) task); active_join ())));
   229 
   230     val res = xs |> map (fn Future {result = ref (SOME res), ...} => res);
   231   in
   232     (case get_first (fn Exn.Exn Interrupt => NONE | Exn.Exn e => SOME e | _ => NONE) res of
   233       NONE => map Exn.release res
   234     | SOME e => raise e)
   235   end;
   236 
   237 fun join x = singleton join_all x;
   238 
   239 end;