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