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