src/Pure/Concurrent/future.ML
author wenzelm
Sat Mar 21 13:11:12 2009 +0100 (2009-03-21 ago)
changeset 30618 046f4f986fb5
parent 30612 cb6421b6a18f
child 30666 d6248d4508d5
permissions -rw-r--r--
restricted interrupts for tasks running as future worker thread -- attempt to prevent interrupt race conditions;
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@30618
    47
  val interruptible_task: ('a -> 'b) -> 'a -> 'b
wenzelm@28206
    48
  val interrupt_task: string -> unit
wenzelm@29431
    49
  val cancel_group: group -> unit
wenzelm@28972
    50
  val cancel: 'a future -> unit
wenzelm@28203
    51
  val shutdown: unit -> unit
wenzelm@28156
    52
end;
wenzelm@28156
    53
wenzelm@28156
    54
structure Future: FUTURE =
wenzelm@28156
    55
struct
wenzelm@28156
    56
wenzelm@28177
    57
(** future values **)
wenzelm@28177
    58
wenzelm@28645
    59
fun enabled () =
wenzelm@29118
    60
  Multithreading.enabled () andalso
wenzelm@28645
    61
    not (Multithreading.self_critical ());
wenzelm@28645
    62
wenzelm@28645
    63
wenzelm@28167
    64
(* identifiers *)
wenzelm@28167
    65
wenzelm@29119
    66
type task = Task_Queue.task;
wenzelm@29119
    67
type group = Task_Queue.group;
wenzelm@28167
    68
wenzelm@28979
    69
local val tag = Universal.tag () : (string * task) option Universal.tag in
wenzelm@28177
    70
  fun thread_data () = the_default NONE (Thread.getLocal tag);
wenzelm@28390
    71
  fun setmp_thread_data data f x = Library.setmp_thread_data tag (thread_data ()) (SOME data) f x;
wenzelm@28167
    72
end;
wenzelm@28167
    73
wenzelm@28167
    74
wenzelm@28167
    75
(* datatype future *)
wenzelm@28167
    76
wenzelm@28972
    77
datatype 'a future = Future of
wenzelm@28167
    78
 {task: task,
wenzelm@28177
    79
  group: group,
wenzelm@28167
    80
  result: 'a Exn.result option ref};
wenzelm@28167
    81
wenzelm@28167
    82
fun task_of (Future {task, ...}) = task;
wenzelm@28167
    83
fun group_of (Future {group, ...}) = group;
wenzelm@28167
    84
wenzelm@28558
    85
fun peek (Future {result, ...}) = ! result;
wenzelm@28558
    86
fun is_finished x = is_some (peek x);
wenzelm@28320
    87
wenzelm@28997
    88
fun value x = Future
wenzelm@29119
    89
 {task = Task_Queue.new_task 0,
wenzelm@29119
    90
  group = Task_Queue.new_group (),
wenzelm@28997
    91
  result = ref (SOME (Exn.Result x))};
wenzelm@28997
    92
wenzelm@28167
    93
wenzelm@28177
    94
wenzelm@28177
    95
(** scheduling **)
wenzelm@28177
    96
wenzelm@28177
    97
(* global state *)
wenzelm@28177
    98
wenzelm@29119
    99
val queue = ref Task_Queue.empty;
wenzelm@28468
   100
val next = ref 0;
wenzelm@28192
   101
val workers = ref ([]: (Thread.thread * bool) list);
wenzelm@28177
   102
val scheduler = ref (NONE: Thread.thread option);
wenzelm@28177
   103
val excessive = ref 0;
wenzelm@29119
   104
val canceled = ref ([]: Task_Queue.group list);
wenzelm@28206
   105
val do_shutdown = ref false;
wenzelm@28177
   106
wenzelm@28177
   107
wenzelm@28177
   108
(* synchronization *)
wenzelm@28156
   109
wenzelm@28156
   110
local
wenzelm@28156
   111
  val lock = Mutex.mutex ();
wenzelm@28156
   112
  val cond = ConditionVar.conditionVar ();
wenzelm@28156
   113
in
wenzelm@28156
   114
wenzelm@28575
   115
fun SYNCHRONIZED name = SimpleThread.synchronized name lock;
wenzelm@28156
   116
wenzelm@29119
   117
fun wait () = (*requires SYNCHRONIZED*)
wenzelm@28206
   118
  ConditionVar.wait (cond, lock);
wenzelm@28206
   119
wenzelm@29119
   120
fun wait_timeout timeout = (*requires SYNCHRONIZED*)
wenzelm@29341
   121
  ignore (ConditionVar.waitUntil (cond, lock, Time.+ (Time.now (), timeout)));
wenzelm@28166
   122
wenzelm@28166
   123
fun notify_all () = (*requires SYNCHRONIZED*)
wenzelm@28166
   124
  ConditionVar.broadcast cond;
wenzelm@28156
   125
wenzelm@28156
   126
end;
wenzelm@28156
   127
wenzelm@28156
   128
wenzelm@28382
   129
(* worker activity *)
wenzelm@28382
   130
wenzelm@28382
   131
fun trace_active () =
wenzelm@28382
   132
  let
wenzelm@28382
   133
    val ws = ! workers;
wenzelm@28382
   134
    val m = string_of_int (length ws);
wenzelm@28382
   135
    val n = string_of_int (length (filter #2 ws));
wenzelm@28382
   136
  in Multithreading.tracing 1 (fn () => "SCHEDULE: " ^ m ^ " workers, " ^ n ^ " active") end;
wenzelm@28382
   137
wenzelm@28382
   138
fun change_active active = (*requires SYNCHRONIZED*)
wenzelm@28382
   139
  change workers (AList.update Thread.equal (Thread.self (), active));
wenzelm@28382
   140
wenzelm@28382
   141
wenzelm@29366
   142
(* execute jobs *)
wenzelm@28156
   143
wenzelm@29341
   144
fun do_cancel group = (*requires SYNCHRONIZED*)
wenzelm@29341
   145
  change canceled (insert Task_Queue.eq_group group);
wenzelm@29341
   146
wenzelm@29366
   147
fun execute name (task, group, jobs) =
wenzelm@28167
   148
  let
wenzelm@28382
   149
    val _ = trace_active ();
wenzelm@29384
   150
    val valid = Task_Queue.is_valid group;
wenzelm@29366
   151
    val ok = setmp_thread_data (name, task) (fn () =>
wenzelm@29384
   152
      fold (fn job => fn ok => job valid andalso ok) jobs true) ();
wenzelm@28192
   153
    val _ = SYNCHRONIZED "execute" (fn () =>
wenzelm@29119
   154
     (change queue (Task_Queue.finish task);
wenzelm@28186
   155
      if ok then ()
wenzelm@29119
   156
      else if Task_Queue.cancel (! queue) group then ()
wenzelm@29341
   157
      else do_cancel group;
wenzelm@28177
   158
      notify_all ()));
wenzelm@28167
   159
  in () end;
wenzelm@28167
   160
wenzelm@28167
   161
wenzelm@28167
   162
(* worker threads *)
wenzelm@28167
   163
wenzelm@29119
   164
fun worker_wait () = (*requires SYNCHRONIZED*)
wenzelm@29119
   165
  (change_active false; wait (); change_active true);
wenzelm@28162
   166
wenzelm@29119
   167
fun worker_next () = (*requires SYNCHRONIZED*)
wenzelm@28167
   168
  if ! excessive > 0 then
wenzelm@28167
   169
    (dec excessive;
wenzelm@28192
   170
     change workers (filter_out (fn (thread, _) => Thread.equal (thread, Thread.self ())));
wenzelm@28203
   171
     notify_all ();
wenzelm@28167
   172
     NONE)
wenzelm@28166
   173
  else
wenzelm@29119
   174
    (case change_result queue Task_Queue.dequeue of
wenzelm@29119
   175
      NONE => (worker_wait (); worker_next ())
wenzelm@28166
   176
    | some => some);
wenzelm@28156
   177
wenzelm@28167
   178
fun worker_loop name =
wenzelm@29119
   179
  (case SYNCHRONIZED name worker_next of
wenzelm@29119
   180
    NONE => ()
wenzelm@28167
   181
  | SOME work => (execute name work; worker_loop name));
wenzelm@28156
   182
wenzelm@28167
   183
fun worker_start name = (*requires SYNCHRONIZED*)
wenzelm@28242
   184
  change workers (cons (SimpleThread.fork false (fn () => worker_loop name), true));
wenzelm@28156
   185
wenzelm@28156
   186
wenzelm@28156
   187
(* scheduler *)
wenzelm@28156
   188
wenzelm@28206
   189
fun scheduler_next () = (*requires SYNCHRONIZED*)
wenzelm@28156
   190
  let
wenzelm@28206
   191
    (*worker threads*)
wenzelm@28191
   192
    val _ =
wenzelm@28192
   193
      (case List.partition (Thread.isActive o #1) (! workers) of
wenzelm@28191
   194
        (_, []) => ()
wenzelm@28191
   195
      | (active, inactive) =>
wenzelm@28191
   196
          (workers := active; Multithreading.tracing 0 (fn () =>
wenzelm@28192
   197
            "SCHEDULE: disposed " ^ string_of_int (length inactive) ^ " dead worker threads")));
wenzelm@28382
   198
    val _ = trace_active ();
wenzelm@28191
   199
wenzelm@28206
   200
    val m = if ! do_shutdown then 0 else Multithreading.max_threads_value ();
wenzelm@28167
   201
    val l = length (! workers);
wenzelm@28167
   202
    val _ = excessive := l - m;
wenzelm@28203
   203
    val _ =
wenzelm@28468
   204
      if m > l then funpow (m - l) (fn () => worker_start ("worker " ^ string_of_int (inc next))) ()
wenzelm@28203
   205
      else ();
wenzelm@28206
   206
wenzelm@28206
   207
    (*canceled groups*)
wenzelm@29119
   208
    val _ =  change canceled (filter_out (Task_Queue.cancel (! queue)));
wenzelm@28206
   209
wenzelm@28206
   210
    (*shutdown*)
wenzelm@28206
   211
    val continue = not (! do_shutdown andalso null (! workers));
wenzelm@28206
   212
    val _ = if continue then () else scheduler := NONE;
wenzelm@28167
   213
wenzelm@28206
   214
    val _ = notify_all ();
wenzelm@29341
   215
    val _ = interruptible (fn () => wait_timeout (Time.fromSeconds 1)) ()
wenzelm@29341
   216
      handle Exn.Interrupt => List.app do_cancel (Task_Queue.cancel_all (! queue));
wenzelm@28206
   217
  in continue end;
wenzelm@28206
   218
wenzelm@28206
   219
fun scheduler_loop () =
wenzelm@29119
   220
  while SYNCHRONIZED "scheduler" scheduler_next do ();
wenzelm@28156
   221
wenzelm@28203
   222
fun scheduler_active () = (*requires SYNCHRONIZED*)
wenzelm@28203
   223
  (case ! scheduler of NONE => false | SOME thread => Thread.isActive thread);
wenzelm@28203
   224
wenzelm@28464
   225
fun scheduler_check name = SYNCHRONIZED name (fn () =>
wenzelm@28206
   226
  if not (scheduler_active ()) then
wenzelm@29119
   227
    (do_shutdown := false; scheduler := SOME (SimpleThread.fork false scheduler_loop))
wenzelm@28206
   228
  else if ! do_shutdown then error "Scheduler shutdown in progress"
wenzelm@28206
   229
  else ());
wenzelm@28156
   230
wenzelm@28156
   231
wenzelm@29366
   232
wenzelm@29366
   233
(** futures **)
wenzelm@28156
   234
wenzelm@29366
   235
(* future job: fill result *)
wenzelm@29366
   236
wenzelm@29366
   237
fun future_job group (e: unit -> 'a) =
wenzelm@28156
   238
  let
wenzelm@28166
   239
    val result = ref (NONE: 'a Exn.result option);
wenzelm@30612
   240
    val job = Multithreading.with_attributes Multithreading.restricted_interrupts
wenzelm@28177
   241
      (fn _ => fn ok =>
wenzelm@28532
   242
        let
wenzelm@28532
   243
          val res = if ok then Exn.capture e () else Exn.Exn Exn.Interrupt;
wenzelm@28548
   244
          val _ = result := SOME res;
wenzelm@28532
   245
          val res_ok =
wenzelm@28532
   246
            (case res of
wenzelm@28532
   247
              Exn.Result _ => true
wenzelm@29119
   248
            | Exn.Exn Exn.Interrupt => (Task_Queue.invalidate_group group; true)
wenzelm@28532
   249
            | _ => false);
wenzelm@28548
   250
        in res_ok end);
wenzelm@29366
   251
  in (result, job) end;
wenzelm@28177
   252
wenzelm@29366
   253
wenzelm@29366
   254
(* fork *)
wenzelm@29366
   255
wenzelm@29366
   256
fun fork_future opt_group deps pri e =
wenzelm@29366
   257
  let
wenzelm@29366
   258
    val _ = scheduler_check "future check";
wenzelm@29366
   259
wenzelm@29366
   260
    val group = (case opt_group of SOME group => group | NONE => Task_Queue.new_group ());
wenzelm@29366
   261
    val (result, job) = future_job group e;
wenzelm@28192
   262
    val task = SYNCHRONIZED "future" (fn () =>
wenzelm@29366
   263
      change_result queue (Task_Queue.enqueue group deps pri job) before notify_all ());
wenzelm@28166
   264
  in Future {task = task, group = group, result = result} end;
wenzelm@28162
   265
wenzelm@29366
   266
fun fork e = fork_future NONE [] 0 e;
wenzelm@29366
   267
fun fork_group group e = fork_future (SOME group) [] 0 e;
wenzelm@29366
   268
fun fork_deps deps e = fork_future NONE (map task_of deps) 0 e;
wenzelm@29366
   269
fun fork_pri pri e = fork_future NONE [] pri e;
wenzelm@28186
   270
wenzelm@28186
   271
wenzelm@29366
   272
(* join *)
wenzelm@29366
   273
wenzelm@29551
   274
local
wenzelm@29551
   275
wenzelm@29366
   276
fun get_result x = the_default (Exn.Exn (SYS_ERROR "unfinished future")) (peek x);
wenzelm@28186
   277
wenzelm@29551
   278
fun join_next pending = (*requires SYNCHRONIZED*)
wenzelm@29551
   279
  if forall is_finished pending then NONE
wenzelm@29551
   280
  else
wenzelm@29551
   281
    (case change_result queue Task_Queue.dequeue of
wenzelm@29551
   282
      NONE => (worker_wait (); join_next pending)
wenzelm@29551
   283
    | some => some);
wenzelm@29551
   284
wenzelm@29551
   285
fun join_loop name pending =
wenzelm@29551
   286
  (case SYNCHRONIZED name (fn () => join_next pending) of
wenzelm@29551
   287
    NONE => ()
wenzelm@29551
   288
  | SOME work => (execute name work; join_loop name pending));
wenzelm@29551
   289
wenzelm@29551
   290
in
wenzelm@29551
   291
wenzelm@29366
   292
fun join_results xs =
wenzelm@29366
   293
  if forall is_finished xs then map get_result xs
wenzelm@29366
   294
  else uninterruptible (fn _ => fn () =>
wenzelm@29366
   295
    let
wenzelm@29366
   296
      val _ = scheduler_check "join check";
wenzelm@29366
   297
      val _ = Multithreading.self_critical () andalso
wenzelm@29366
   298
        error "Cannot join future values within critical section";
wenzelm@28177
   299
wenzelm@29551
   300
      fun join_deps _ [] = ()
wenzelm@29551
   301
        | join_deps name deps =
wenzelm@29366
   302
            (case SYNCHRONIZED name (fn () =>
wenzelm@29366
   303
                change_result queue (Task_Queue.dequeue_towards deps)) of
wenzelm@29366
   304
              NONE => ()
wenzelm@29551
   305
            | SOME (work, deps') => (execute name work; join_deps name deps'));
wenzelm@29551
   306
wenzelm@29366
   307
      val _ =
wenzelm@29366
   308
        (case thread_data () of
wenzelm@29366
   309
          NONE =>
wenzelm@29366
   310
            (*alien thread -- refrain from contending for resources*)
wenzelm@29366
   311
            while not (forall is_finished xs)
wenzelm@29366
   312
            do SYNCHRONIZED "join_thread" (fn () => wait ())
wenzelm@29366
   313
        | SOME (name, task) =>
wenzelm@29366
   314
            (*proper task -- actively work towards results*)
wenzelm@29366
   315
            let
wenzelm@29366
   316
              val pending = filter_out is_finished xs;
wenzelm@29366
   317
              val deps = map task_of pending;
wenzelm@29366
   318
              val _ = SYNCHRONIZED "join" (fn () =>
wenzelm@29366
   319
                (change queue (Task_Queue.depend deps task); notify_all ()));
wenzelm@29551
   320
              val _ = join_deps ("join_deps: " ^ name) deps;
wenzelm@29551
   321
              val _ = join_loop ("join_loop: " ^ name) (filter_out is_finished pending);
wenzelm@29366
   322
            in () end);
wenzelm@28186
   323
wenzelm@29366
   324
    in map get_result xs end) ();
wenzelm@28186
   325
wenzelm@29551
   326
end;
wenzelm@29551
   327
wenzelm@28647
   328
fun join_result x = singleton join_results x;
wenzelm@28647
   329
fun join x = Exn.release (join_result x);
wenzelm@28156
   330
wenzelm@29366
   331
wenzelm@29366
   332
(* map *)
wenzelm@29366
   333
wenzelm@29384
   334
fun map_future f x =
wenzelm@29366
   335
  let
wenzelm@29366
   336
    val _ = scheduler_check "map_future check";
wenzelm@29366
   337
wenzelm@29384
   338
    val task = task_of x;
wenzelm@29384
   339
    val group = Task_Queue.new_group ();
wenzelm@29384
   340
    val (result, job) = future_job group (fn () => f (join x));
wenzelm@29384
   341
wenzelm@29366
   342
    val extended = SYNCHRONIZED "map_future" (fn () =>
wenzelm@29366
   343
      (case Task_Queue.extend task job (! queue) of
wenzelm@29366
   344
        SOME queue' => (queue := queue'; true)
wenzelm@29366
   345
      | NONE => false));
wenzelm@29366
   346
  in
wenzelm@29384
   347
    if extended then Future {task = task, group = group, result = result}
wenzelm@29384
   348
    else fork_future NONE [task] (Task_Queue.pri_of_task task) (fn () => f (join x))
wenzelm@29366
   349
  end;
wenzelm@28979
   350
wenzelm@28191
   351
wenzelm@29431
   352
(* cancellation *)
wenzelm@28202
   353
wenzelm@30618
   354
fun interruptible_task f x =
wenzelm@30618
   355
  if Multithreading.available then
wenzelm@30618
   356
    Multithreading.with_attributes
wenzelm@30618
   357
      (if is_some (thread_data ())
wenzelm@30618
   358
       then Multithreading.restricted_interrupts
wenzelm@30618
   359
       else Multithreading.regular_interrupts)
wenzelm@30618
   360
      (fn _ => f) x
wenzelm@30618
   361
  else interruptible f x;
wenzelm@30618
   362
wenzelm@28202
   363
(*interrupt: permissive signal, may get ignored*)
wenzelm@28197
   364
fun interrupt_task id = SYNCHRONIZED "interrupt"
wenzelm@29119
   365
  (fn () => Task_Queue.interrupt_external (! queue) id);
wenzelm@28191
   366
wenzelm@28206
   367
(*cancel: present and future group members will be interrupted eventually*)
wenzelm@29431
   368
fun cancel_group group =
wenzelm@28464
   369
 (scheduler_check "cancel check";
wenzelm@29431
   370
  SYNCHRONIZED "cancel" (fn () => (do_cancel group; notify_all ())));
wenzelm@28206
   371
wenzelm@29431
   372
fun cancel x = cancel_group (group_of x);
wenzelm@28206
   373
wenzelm@29366
   374
wenzelm@29366
   375
(** global join and shutdown **)
wenzelm@29366
   376
wenzelm@28203
   377
fun shutdown () =
wenzelm@28276
   378
  if Multithreading.available then
wenzelm@28464
   379
   (scheduler_check "shutdown check";
wenzelm@28276
   380
    SYNCHRONIZED "shutdown" (fn () =>
wenzelm@29119
   381
     (while not (scheduler_active ()) do wait ();
wenzelm@29119
   382
      while not (Task_Queue.is_empty (! queue)) do wait ();
wenzelm@28276
   383
      do_shutdown := true;
wenzelm@28276
   384
      notify_all ();
wenzelm@29119
   385
      while not (null (! workers)) do wait ();
wenzelm@29119
   386
      while scheduler_active () do wait ();
wenzelm@28470
   387
      OS.Process.sleep (Time.fromMilliseconds 300))))
wenzelm@28276
   388
  else ();
wenzelm@28203
   389
wenzelm@29366
   390
wenzelm@29366
   391
(*final declarations of this structure!*)
wenzelm@29366
   392
val map = map_future;
wenzelm@29366
   393
wenzelm@28156
   394
end;
wenzelm@28972
   395
wenzelm@28972
   396
type 'a future = 'a Future.future;
wenzelm@28972
   397