src/Pure/Concurrent/future.ML
author wenzelm
Tue Dec 16 16:25:18 2008 +0100 (2008-12-16 ago)
changeset 29119 99941fd0cb0e
parent 29118 8f2481aa363d
child 29341 6bb007a0f9f2
permissions -rw-r--r--
renamed structure TaskQueue to Task_Queue;
generalized fork_background to fork_pri;
reduced tracing;
map: inherit task priority;
removed unused focus;
wenzelm@28156
     1
(*  Title:      Pure/Concurrent/future.ML
wenzelm@28156
     2
    Author:     Makarius
wenzelm@28156
     3
wenzelm@28201
     4
Future values.
wenzelm@28201
     5
wenzelm@28201
     6
Notes:
wenzelm@28201
     7
wenzelm@28201
     8
  * Futures are similar to delayed evaluation, i.e. delay/force is
wenzelm@28201
     9
    generalized to fork/join (and variants).  The idea is to model
wenzelm@28201
    10
    parallel value-oriented computations, but *not* communicating
wenzelm@28201
    11
    processes.
wenzelm@28201
    12
wenzelm@28201
    13
  * Futures are grouped; failure of one group member causes the whole
wenzelm@28201
    14
    group to be interrupted eventually.
wenzelm@28201
    15
wenzelm@28201
    16
  * Forked futures are evaluated spontaneously by a farm of worker
wenzelm@28201
    17
    threads in the background; join resynchronizes the computation and
wenzelm@28201
    18
    delivers results (values or exceptions).
wenzelm@28201
    19
wenzelm@28201
    20
  * The pool of worker threads is limited, usually in correlation with
wenzelm@28201
    21
    the number of physical cores on the machine.  Note that allocation
wenzelm@28201
    22
    of runtime resources is distorted either if workers yield CPU time
wenzelm@28201
    23
    (e.g. via system sleep or wait operations), or if non-worker
wenzelm@28201
    24
    threads contend for significant runtime resources independently.
wenzelm@28156
    25
*)
wenzelm@28156
    26
wenzelm@28156
    27
signature FUTURE =
wenzelm@28156
    28
sig
wenzelm@28645
    29
  val enabled: unit -> bool
wenzelm@29119
    30
  type task = Task_Queue.task
wenzelm@29119
    31
  type group = Task_Queue.group
wenzelm@28979
    32
  val thread_data: unit -> (string * task) option
wenzelm@28972
    33
  type 'a future
wenzelm@28972
    34
  val task_of: 'a future -> task
wenzelm@28972
    35
  val group_of: 'a future -> group
wenzelm@28972
    36
  val peek: 'a future -> 'a Exn.result option
wenzelm@28972
    37
  val is_finished: 'a future -> bool
wenzelm@28997
    38
  val value: 'a -> 'a future
wenzelm@28972
    39
  val fork: (unit -> 'a) -> 'a future
wenzelm@28979
    40
  val fork_group: group -> (unit -> 'a) -> 'a future
wenzelm@28979
    41
  val fork_deps: 'b future list -> (unit -> 'a) -> 'a future
wenzelm@29119
    42
  val fork_pri: int -> (unit -> 'a) -> 'a future
wenzelm@28972
    43
  val join_results: 'a future list -> 'a Exn.result list
wenzelm@28972
    44
  val join_result: 'a future -> 'a Exn.result
wenzelm@28972
    45
  val join: 'a future -> 'a
wenzelm@28972
    46
  val map: ('a -> 'b) -> 'a future -> 'b future
wenzelm@28206
    47
  val interrupt_task: string -> unit
wenzelm@28972
    48
  val cancel: 'a future -> unit
wenzelm@28203
    49
  val shutdown: unit -> unit
wenzelm@28156
    50
end;
wenzelm@28156
    51
wenzelm@28156
    52
structure Future: FUTURE =
wenzelm@28156
    53
struct
wenzelm@28156
    54
wenzelm@28177
    55
(** future values **)
wenzelm@28177
    56
wenzelm@28645
    57
fun enabled () =
wenzelm@29118
    58
  Multithreading.enabled () andalso
wenzelm@28645
    59
    not (Multithreading.self_critical ());
wenzelm@28645
    60
wenzelm@28645
    61
wenzelm@28167
    62
(* identifiers *)
wenzelm@28167
    63
wenzelm@29119
    64
type task = Task_Queue.task;
wenzelm@29119
    65
type group = Task_Queue.group;
wenzelm@28167
    66
wenzelm@28979
    67
local val tag = Universal.tag () : (string * task) option Universal.tag in
wenzelm@28177
    68
  fun thread_data () = the_default NONE (Thread.getLocal tag);
wenzelm@28390
    69
  fun setmp_thread_data data f x = Library.setmp_thread_data tag (thread_data ()) (SOME data) f x;
wenzelm@28167
    70
end;
wenzelm@28167
    71
wenzelm@28167
    72
wenzelm@28167
    73
(* datatype future *)
wenzelm@28167
    74
wenzelm@28972
    75
datatype 'a future = Future of
wenzelm@28167
    76
 {task: task,
wenzelm@28177
    77
  group: group,
wenzelm@28167
    78
  result: 'a Exn.result option ref};
wenzelm@28167
    79
wenzelm@28167
    80
fun task_of (Future {task, ...}) = task;
wenzelm@28167
    81
fun group_of (Future {group, ...}) = group;
wenzelm@28167
    82
wenzelm@28558
    83
fun peek (Future {result, ...}) = ! result;
wenzelm@28558
    84
fun is_finished x = is_some (peek x);
wenzelm@28320
    85
wenzelm@28997
    86
fun value x = Future
wenzelm@29119
    87
 {task = Task_Queue.new_task 0,
wenzelm@29119
    88
  group = Task_Queue.new_group (),
wenzelm@28997
    89
  result = ref (SOME (Exn.Result x))};
wenzelm@28997
    90
wenzelm@28167
    91
wenzelm@28177
    92
wenzelm@28177
    93
(** scheduling **)
wenzelm@28177
    94
wenzelm@28177
    95
(* global state *)
wenzelm@28177
    96
wenzelm@29119
    97
val queue = ref Task_Queue.empty;
wenzelm@28468
    98
val next = ref 0;
wenzelm@28192
    99
val workers = ref ([]: (Thread.thread * bool) list);
wenzelm@28177
   100
val scheduler = ref (NONE: Thread.thread option);
wenzelm@28177
   101
val excessive = ref 0;
wenzelm@29119
   102
val canceled = ref ([]: Task_Queue.group list);
wenzelm@28206
   103
val do_shutdown = ref false;
wenzelm@28177
   104
wenzelm@28177
   105
wenzelm@28177
   106
(* synchronization *)
wenzelm@28156
   107
wenzelm@28156
   108
local
wenzelm@28156
   109
  val lock = Mutex.mutex ();
wenzelm@28156
   110
  val cond = ConditionVar.conditionVar ();
wenzelm@28156
   111
in
wenzelm@28156
   112
wenzelm@28575
   113
fun SYNCHRONIZED name = SimpleThread.synchronized name lock;
wenzelm@28156
   114
wenzelm@29119
   115
fun wait () = (*requires SYNCHRONIZED*)
wenzelm@28206
   116
  ConditionVar.wait (cond, lock);
wenzelm@28206
   117
wenzelm@29119
   118
fun wait_timeout timeout = (*requires SYNCHRONIZED*)
wenzelm@28206
   119
  ConditionVar.waitUntil (cond, lock, Time.+ (Time.now (), timeout));
wenzelm@28166
   120
wenzelm@28166
   121
fun notify_all () = (*requires SYNCHRONIZED*)
wenzelm@28166
   122
  ConditionVar.broadcast cond;
wenzelm@28156
   123
wenzelm@28156
   124
end;
wenzelm@28156
   125
wenzelm@28156
   126
wenzelm@28382
   127
(* worker activity *)
wenzelm@28382
   128
wenzelm@28382
   129
fun trace_active () =
wenzelm@28382
   130
  let
wenzelm@28382
   131
    val ws = ! workers;
wenzelm@28382
   132
    val m = string_of_int (length ws);
wenzelm@28382
   133
    val n = string_of_int (length (filter #2 ws));
wenzelm@28382
   134
  in Multithreading.tracing 1 (fn () => "SCHEDULE: " ^ m ^ " workers, " ^ n ^ " active") end;
wenzelm@28382
   135
wenzelm@28382
   136
fun change_active active = (*requires SYNCHRONIZED*)
wenzelm@28382
   137
  change workers (AList.update Thread.equal (Thread.self (), active));
wenzelm@28382
   138
wenzelm@28382
   139
wenzelm@28177
   140
(* execute *)
wenzelm@28156
   141
wenzelm@28167
   142
fun execute name (task, group, run) =
wenzelm@28167
   143
  let
wenzelm@28382
   144
    val _ = trace_active ();
wenzelm@28979
   145
    val ok = setmp_thread_data (name, task) run ();
wenzelm@28192
   146
    val _ = SYNCHRONIZED "execute" (fn () =>
wenzelm@29119
   147
     (change queue (Task_Queue.finish task);
wenzelm@28186
   148
      if ok then ()
wenzelm@29119
   149
      else if Task_Queue.cancel (! queue) group then ()
wenzelm@28206
   150
      else change canceled (cons group);
wenzelm@28177
   151
      notify_all ()));
wenzelm@28167
   152
  in () end;
wenzelm@28167
   153
wenzelm@28167
   154
wenzelm@28167
   155
(* worker threads *)
wenzelm@28167
   156
wenzelm@29119
   157
fun worker_wait () = (*requires SYNCHRONIZED*)
wenzelm@29119
   158
  (change_active false; wait (); change_active true);
wenzelm@28162
   159
wenzelm@29119
   160
fun worker_next () = (*requires SYNCHRONIZED*)
wenzelm@28167
   161
  if ! excessive > 0 then
wenzelm@28167
   162
    (dec excessive;
wenzelm@28192
   163
     change workers (filter_out (fn (thread, _) => Thread.equal (thread, Thread.self ())));
wenzelm@28203
   164
     notify_all ();
wenzelm@28167
   165
     NONE)
wenzelm@28166
   166
  else
wenzelm@29119
   167
    (case change_result queue Task_Queue.dequeue of
wenzelm@29119
   168
      NONE => (worker_wait (); worker_next ())
wenzelm@28166
   169
    | some => some);
wenzelm@28156
   170
wenzelm@28167
   171
fun worker_loop name =
wenzelm@29119
   172
  (case SYNCHRONIZED name worker_next of
wenzelm@29119
   173
    NONE => ()
wenzelm@28167
   174
  | SOME work => (execute name work; worker_loop name));
wenzelm@28156
   175
wenzelm@28167
   176
fun worker_start name = (*requires SYNCHRONIZED*)
wenzelm@28242
   177
  change workers (cons (SimpleThread.fork false (fn () => worker_loop name), true));
wenzelm@28156
   178
wenzelm@28156
   179
wenzelm@28156
   180
(* scheduler *)
wenzelm@28156
   181
wenzelm@28206
   182
fun scheduler_next () = (*requires SYNCHRONIZED*)
wenzelm@28156
   183
  let
wenzelm@28206
   184
    (*worker threads*)
wenzelm@28191
   185
    val _ =
wenzelm@28192
   186
      (case List.partition (Thread.isActive o #1) (! workers) of
wenzelm@28191
   187
        (_, []) => ()
wenzelm@28191
   188
      | (active, inactive) =>
wenzelm@28191
   189
          (workers := active; Multithreading.tracing 0 (fn () =>
wenzelm@28192
   190
            "SCHEDULE: disposed " ^ string_of_int (length inactive) ^ " dead worker threads")));
wenzelm@28382
   191
    val _ = trace_active ();
wenzelm@28191
   192
wenzelm@28206
   193
    val m = if ! do_shutdown then 0 else Multithreading.max_threads_value ();
wenzelm@28167
   194
    val l = length (! workers);
wenzelm@28167
   195
    val _ = excessive := l - m;
wenzelm@28203
   196
    val _ =
wenzelm@28468
   197
      if m > l then funpow (m - l) (fn () => worker_start ("worker " ^ string_of_int (inc next))) ()
wenzelm@28203
   198
      else ();
wenzelm@28206
   199
wenzelm@28206
   200
    (*canceled groups*)
wenzelm@29119
   201
    val _ =  change canceled (filter_out (Task_Queue.cancel (! queue)));
wenzelm@28206
   202
wenzelm@28206
   203
    (*shutdown*)
wenzelm@28206
   204
    val continue = not (! do_shutdown andalso null (! workers));
wenzelm@28206
   205
    val _ = if continue then () else scheduler := NONE;
wenzelm@28167
   206
wenzelm@28206
   207
    val _ = notify_all ();
wenzelm@29119
   208
    val _ = wait_timeout (Time.fromSeconds 3);
wenzelm@28206
   209
  in continue end;
wenzelm@28206
   210
wenzelm@28206
   211
fun scheduler_loop () =
wenzelm@29119
   212
  while SYNCHRONIZED "scheduler" scheduler_next do ();
wenzelm@28156
   213
wenzelm@28203
   214
fun scheduler_active () = (*requires SYNCHRONIZED*)
wenzelm@28203
   215
  (case ! scheduler of NONE => false | SOME thread => Thread.isActive thread);
wenzelm@28203
   216
wenzelm@28464
   217
fun scheduler_check name = SYNCHRONIZED name (fn () =>
wenzelm@28206
   218
  if not (scheduler_active ()) then
wenzelm@29119
   219
    (do_shutdown := false; scheduler := SOME (SimpleThread.fork false scheduler_loop))
wenzelm@28206
   220
  else if ! do_shutdown then error "Scheduler shutdown in progress"
wenzelm@28206
   221
  else ());
wenzelm@28156
   222
wenzelm@28156
   223
wenzelm@28191
   224
(* future values: fork independent computation *)
wenzelm@28156
   225
wenzelm@28304
   226
fun future opt_group deps pri (e: unit -> 'a) =
wenzelm@28156
   227
  let
wenzelm@28464
   228
    val _ = scheduler_check "future check";
wenzelm@28177
   229
wenzelm@29119
   230
    val group = (case opt_group of SOME group => group | NONE => Task_Queue.new_group ());
wenzelm@28177
   231
wenzelm@28166
   232
    val result = ref (NONE: 'a Exn.result option);
wenzelm@28177
   233
    val run = Multithreading.with_attributes (Thread.getAttributes ())
wenzelm@28177
   234
      (fn _ => fn ok =>
wenzelm@28532
   235
        let
wenzelm@28532
   236
          val res = if ok then Exn.capture e () else Exn.Exn Exn.Interrupt;
wenzelm@28548
   237
          val _ = result := SOME res;
wenzelm@28532
   238
          val res_ok =
wenzelm@28532
   239
            (case res of
wenzelm@28532
   240
              Exn.Result _ => true
wenzelm@29119
   241
            | Exn.Exn Exn.Interrupt => (Task_Queue.invalidate_group group; true)
wenzelm@28532
   242
            | _ => false);
wenzelm@28548
   243
        in res_ok end);
wenzelm@28177
   244
wenzelm@28192
   245
    val task = SYNCHRONIZED "future" (fn () =>
wenzelm@29119
   246
      change_result queue (Task_Queue.enqueue group deps pri run) before notify_all ());
wenzelm@28166
   247
  in Future {task = task, group = group, result = result} end;
wenzelm@28162
   248
wenzelm@29119
   249
fun fork e = future NONE [] 0 e;
wenzelm@29119
   250
fun fork_group group e = future (SOME group) [] 0 e;
wenzelm@29119
   251
fun fork_deps deps e = future NONE (map task_of deps) 0 e;
wenzelm@29119
   252
fun fork_pri pri e = future NONE [] pri e;
wenzelm@28186
   253
wenzelm@28186
   254
wenzelm@28191
   255
(* join: retrieve results *)
wenzelm@28186
   256
wenzelm@28331
   257
fun join_results [] = []
wenzelm@28532
   258
  | join_results xs = uninterruptible (fn _ => fn () =>
wenzelm@28331
   259
      let
wenzelm@28464
   260
        val _ = scheduler_check "join check";
wenzelm@28331
   261
        val _ = Multithreading.self_critical () andalso
wenzelm@28647
   262
          exists (not o is_finished) xs andalso
wenzelm@28331
   263
          error "Cannot join future values within critical section";
wenzelm@28177
   264
wenzelm@28386
   265
        fun join_loop _ [] = ()
wenzelm@28386
   266
          | join_loop name tasks =
wenzelm@28386
   267
              (case SYNCHRONIZED name (fn () =>
wenzelm@29119
   268
                  change_result queue (Task_Queue.dequeue_towards tasks)) of
wenzelm@28382
   269
                NONE => ()
wenzelm@28386
   270
              | SOME (work, tasks') => (execute name work; join_loop name tasks'));
wenzelm@28331
   271
        val _ =
wenzelm@28331
   272
          (case thread_data () of
wenzelm@28382
   273
            NONE =>
wenzelm@28382
   274
              (*alien thread -- refrain from contending for resources*)
wenzelm@28382
   275
              while exists (not o is_finished) xs
wenzelm@29119
   276
              do SYNCHRONIZED "join_thread" (fn () => wait ())
wenzelm@28979
   277
          | SOME (name, task) =>
wenzelm@28382
   278
              (*proper task -- actively work towards results*)
wenzelm@28382
   279
              let
wenzelm@28382
   280
                val unfinished = xs |> map_filter
wenzelm@28382
   281
                  (fn Future {task, result = ref NONE, ...} => SOME task | _ => NONE);
wenzelm@28382
   282
                val _ = SYNCHRONIZED "join" (fn () =>
wenzelm@29119
   283
                  (change queue (Task_Queue.depend unfinished task); notify_all ()));
wenzelm@28386
   284
                val _ = join_loop ("join_loop: " ^ name) unfinished;
wenzelm@28382
   285
                val _ =
wenzelm@28382
   286
                  while exists (not o is_finished) xs
wenzelm@29119
   287
                  do SYNCHRONIZED "join_task" (fn () => worker_wait ());
wenzelm@28382
   288
              in () end);
wenzelm@28186
   289
wenzelm@28532
   290
      in xs |> map (fn Future {result = ref (SOME res), ...} => res) end) ();
wenzelm@28186
   291
wenzelm@28647
   292
fun join_result x = singleton join_results x;
wenzelm@28647
   293
fun join x = Exn.release (join_result x);
wenzelm@28156
   294
wenzelm@29119
   295
fun map f x =
wenzelm@29119
   296
  let val task = task_of x
wenzelm@29119
   297
  in future NONE [task] (Task_Queue.pri_of_task task) (fn () => f (join x)) end;
wenzelm@28979
   298
wenzelm@28191
   299
wenzelm@28202
   300
(* misc operations *)
wenzelm@28202
   301
wenzelm@28202
   302
(*interrupt: permissive signal, may get ignored*)
wenzelm@28197
   303
fun interrupt_task id = SYNCHRONIZED "interrupt"
wenzelm@29119
   304
  (fn () => Task_Queue.interrupt_external (! queue) id);
wenzelm@28191
   305
wenzelm@28206
   306
(*cancel: present and future group members will be interrupted eventually*)
wenzelm@28206
   307
fun cancel x =
wenzelm@28464
   308
 (scheduler_check "cancel check";
wenzelm@28208
   309
  SYNCHRONIZED "cancel" (fn () => (change canceled (cons (group_of x)); notify_all ())));
wenzelm@28206
   310
wenzelm@28206
   311
wenzelm@28203
   312
(*global join and shutdown*)
wenzelm@28203
   313
fun shutdown () =
wenzelm@28276
   314
  if Multithreading.available then
wenzelm@28464
   315
   (scheduler_check "shutdown check";
wenzelm@28276
   316
    SYNCHRONIZED "shutdown" (fn () =>
wenzelm@29119
   317
     (while not (scheduler_active ()) do wait ();
wenzelm@29119
   318
      while not (Task_Queue.is_empty (! queue)) do wait ();
wenzelm@28276
   319
      do_shutdown := true;
wenzelm@28276
   320
      notify_all ();
wenzelm@29119
   321
      while not (null (! workers)) do wait ();
wenzelm@29119
   322
      while scheduler_active () do wait ();
wenzelm@28470
   323
      OS.Process.sleep (Time.fromMilliseconds 300))))
wenzelm@28276
   324
  else ();
wenzelm@28203
   325
wenzelm@28156
   326
end;
wenzelm@28972
   327
wenzelm@28972
   328
type 'a future = 'a Future.future;
wenzelm@28972
   329