src/Pure/Concurrent/future.ML
author wenzelm
Wed Jan 16 16:26:36 2013 +0100 (2013-01-16)
changeset 50911 ee7fe4230642
parent 50909 b2fb1ab1475d
child 50914 fe4714886d92
permissions -rw-r--r--
more explicit treatment of (optional) exception properties, notably for "serial" -- avoid conflict with startPosition = offset;
wenzelm@28156
     1
(*  Title:      Pure/Concurrent/future.ML
wenzelm@28156
     2
    Author:     Makarius
wenzelm@28156
     3
wenzelm@44268
     4
Value-oriented parallelism via futures and promises.  See also
wenzelm@32246
     5
http://www4.in.tum.de/~wenzelm/papers/parallel-isabelle.pdf
wenzelm@37904
     6
http://www4.in.tum.de/~wenzelm/papers/parallel-ml.pdf
wenzelm@28201
     7
wenzelm@28201
     8
Notes:
wenzelm@28201
     9
wenzelm@28201
    10
  * Futures are similar to delayed evaluation, i.e. delay/force is
wenzelm@44268
    11
    generalized to fork/join.  The idea is to model parallel
wenzelm@44268
    12
    value-oriented computations (not communicating processes).
wenzelm@28201
    13
wenzelm@28201
    14
  * Forked futures are evaluated spontaneously by a farm of worker
wenzelm@28201
    15
    threads in the background; join resynchronizes the computation and
wenzelm@28201
    16
    delivers results (values or exceptions).
wenzelm@28201
    17
wenzelm@28201
    18
  * The pool of worker threads is limited, usually in correlation with
wenzelm@28201
    19
    the number of physical cores on the machine.  Note that allocation
wenzelm@44268
    20
    of runtime resources may be distorted either if workers yield CPU
wenzelm@44268
    21
    time (e.g. via system sleep or wait operations), or if non-worker
wenzelm@28201
    22
    threads contend for significant runtime resources independently.
wenzelm@44268
    23
    There is a limited number of replacement worker threads that get
wenzelm@44268
    24
    activated in certain explicit wait conditions.
wenzelm@34277
    25
wenzelm@44268
    26
  * Future tasks are organized in groups, which are block-structured.
wenzelm@44268
    27
    When forking a new new task, the default is to open an individual
wenzelm@44268
    28
    subgroup, unless some common group is specified explicitly.
wenzelm@49908
    29
    Failure of one group member causes peer and subgroup members to be
wenzelm@49908
    30
    interrupted eventually.  Interrupted tasks that lack regular
wenzelm@49908
    31
    result information, will pick up parallel exceptions from the
wenzelm@49908
    32
    cumulative group context (as Par_Exn).
wenzelm@44268
    33
wenzelm@44299
    34
  * Future task groups may be canceled: present and future group
wenzelm@44299
    35
    members will be interrupted eventually.
wenzelm@44299
    36
wenzelm@44268
    37
  * Promised "passive" futures are fulfilled by external means.  There
wenzelm@44268
    38
    is no associated evaluation task, but other futures can depend on
wenzelm@44268
    39
    them via regular join operations.
wenzelm@28156
    40
*)
wenzelm@28156
    41
wenzelm@28156
    42
signature FUTURE =
wenzelm@28156
    43
sig
wenzelm@44300
    44
  type task = Task_Queue.task
wenzelm@44300
    45
  type group = Task_Queue.group
wenzelm@44300
    46
  val new_group: group option -> group
wenzelm@44300
    47
  val worker_task: unit -> task option
wenzelm@44300
    48
  val worker_group: unit -> group option
wenzelm@44300
    49
  val worker_subgroup: unit -> group
wenzelm@28972
    50
  type 'a future
wenzelm@44300
    51
  val task_of: 'a future -> task
wenzelm@28972
    52
  val peek: 'a future -> 'a Exn.result option
wenzelm@28972
    53
  val is_finished: 'a future -> bool
wenzelm@50280
    54
  val ML_statistics: bool Unsynchronized.ref
wenzelm@44301
    55
  val interruptible_task: ('a -> 'b) -> 'a -> 'b
wenzelm@47404
    56
  val cancel_group: group -> unit
wenzelm@47404
    57
  val cancel: 'a future -> unit
wenzelm@44427
    58
  type params = {name: string, group: group option, deps: task list, pri: int, interrupts: bool}
wenzelm@44427
    59
  val default_params: params
wenzelm@44427
    60
  val forks: params -> (unit -> 'a) list -> 'a future list
wenzelm@29119
    61
  val fork_pri: int -> (unit -> 'a) -> 'a future
wenzelm@32724
    62
  val fork: (unit -> 'a) -> 'a future
wenzelm@28972
    63
  val join_results: 'a future list -> 'a Exn.result list
wenzelm@28972
    64
  val join_result: 'a future -> 'a Exn.result
wenzelm@44330
    65
  val joins: 'a future list -> 'a list
wenzelm@28972
    66
  val join: 'a future -> 'a
wenzelm@44294
    67
  val value_result: 'a Exn.result -> 'a future
wenzelm@34277
    68
  val value: 'a -> 'a future
wenzelm@44427
    69
  val cond_forks: params -> (unit -> 'a) list -> 'a future list
wenzelm@28972
    70
  val map: ('a -> 'b) -> 'a future -> 'b future
wenzelm@44300
    71
  val promise_group: group -> (unit -> unit) -> 'a future
wenzelm@44298
    72
  val promise: (unit -> unit) -> 'a future
wenzelm@34277
    73
  val fulfill_result: 'a future -> 'a Exn.result -> unit
wenzelm@34277
    74
  val fulfill: 'a future -> 'a -> unit
wenzelm@49906
    75
  val terminate: group -> unit
wenzelm@28203
    76
  val shutdown: unit -> unit
wenzelm@28156
    77
end;
wenzelm@28156
    78
wenzelm@28156
    79
structure Future: FUTURE =
wenzelm@28156
    80
struct
wenzelm@28156
    81
wenzelm@28177
    82
(** future values **)
wenzelm@28177
    83
wenzelm@44300
    84
type task = Task_Queue.task;
wenzelm@44300
    85
type group = Task_Queue.group;
wenzelm@44300
    86
val new_group = Task_Queue.new_group;
wenzelm@44300
    87
wenzelm@44300
    88
wenzelm@28167
    89
(* identifiers *)
wenzelm@28167
    90
wenzelm@32058
    91
local
wenzelm@44300
    92
  val tag = Universal.tag () : task option Universal.tag;
wenzelm@32058
    93
in
wenzelm@41683
    94
  fun worker_task () = the_default NONE (Thread.getLocal tag);
wenzelm@44110
    95
  fun setmp_worker_task task f x = setmp_thread_data tag (worker_task ()) (SOME task) f x;
wenzelm@28167
    96
end;
wenzelm@28167
    97
wenzelm@41683
    98
val worker_group = Option.map Task_Queue.group_of_task o worker_task;
wenzelm@44300
    99
fun worker_subgroup () = new_group (worker_group ());
wenzelm@34277
   100
wenzelm@41679
   101
fun worker_joining e =
wenzelm@41679
   102
  (case worker_task () of
wenzelm@41679
   103
    NONE => e ()
wenzelm@41679
   104
  | SOME task => Task_Queue.joining task e);
wenzelm@41679
   105
wenzelm@41680
   106
fun worker_waiting deps e =
wenzelm@41670
   107
  (case worker_task () of
wenzelm@41670
   108
    NONE => e ()
wenzelm@41680
   109
  | SOME task => Task_Queue.waiting task deps e);
wenzelm@41670
   110
wenzelm@28167
   111
wenzelm@28167
   112
(* datatype future *)
wenzelm@28167
   113
wenzelm@35016
   114
type 'a result = 'a Exn.result Single_Assignment.var;
wenzelm@35016
   115
wenzelm@28972
   116
datatype 'a future = Future of
wenzelm@34277
   117
 {promised: bool,
wenzelm@44300
   118
  task: task,
wenzelm@35016
   119
  result: 'a result};
wenzelm@28167
   120
wenzelm@28167
   121
fun task_of (Future {task, ...}) = task;
wenzelm@32253
   122
fun result_of (Future {result, ...}) = result;
wenzelm@28167
   123
wenzelm@35016
   124
fun peek x = Single_Assignment.peek (result_of x);
wenzelm@28558
   125
fun is_finished x = is_some (peek x);
wenzelm@28320
   126
wenzelm@28167
   127
wenzelm@28177
   128
wenzelm@28177
   129
(** scheduling **)
wenzelm@28177
   130
wenzelm@28177
   131
(* synchronization *)
wenzelm@28156
   132
wenzelm@32219
   133
val scheduler_event = ConditionVar.conditionVar ();
wenzelm@32219
   134
val work_available = ConditionVar.conditionVar ();
wenzelm@32219
   135
val work_finished = ConditionVar.conditionVar ();
wenzelm@32219
   136
wenzelm@28156
   137
local
wenzelm@28156
   138
  val lock = Mutex.mutex ();
wenzelm@28156
   139
in
wenzelm@28156
   140
wenzelm@37216
   141
fun SYNCHRONIZED name = Simple_Thread.synchronized name lock;
wenzelm@28156
   142
wenzelm@32219
   143
fun wait cond = (*requires SYNCHRONIZED*)
wenzelm@32295
   144
  Multithreading.sync_wait NONE NONE cond lock;
wenzelm@28206
   145
wenzelm@32295
   146
fun wait_timeout timeout cond = (*requires SYNCHRONIZED*)
wenzelm@32295
   147
  Multithreading.sync_wait NONE (SOME (Time.+ (Time.now (), timeout))) cond lock;
wenzelm@28166
   148
wenzelm@32219
   149
fun signal cond = (*requires SYNCHRONIZED*)
wenzelm@32219
   150
  ConditionVar.signal cond;
wenzelm@32219
   151
wenzelm@32219
   152
fun broadcast cond = (*requires SYNCHRONIZED*)
wenzelm@28166
   153
  ConditionVar.broadcast cond;
wenzelm@28156
   154
wenzelm@32248
   155
fun broadcast_work () = (*requires SYNCHRONIZED*)
wenzelm@32248
   156
 (ConditionVar.broadcast work_available;
wenzelm@32225
   157
  ConditionVar.broadcast work_finished);
wenzelm@32225
   158
wenzelm@28156
   159
end;
wenzelm@28156
   160
wenzelm@28156
   161
wenzelm@33410
   162
(* global state *)
wenzelm@33410
   163
wenzelm@33410
   164
val queue = Unsynchronized.ref Task_Queue.empty;
wenzelm@33410
   165
val next = Unsynchronized.ref 0;
wenzelm@33410
   166
val scheduler = Unsynchronized.ref (NONE: Thread.thread option);
wenzelm@44300
   167
val canceled = Unsynchronized.ref ([]: group list);
wenzelm@33410
   168
val do_shutdown = Unsynchronized.ref false;
wenzelm@33410
   169
val max_workers = Unsynchronized.ref 0;
wenzelm@33410
   170
val max_active = Unsynchronized.ref 0;
wenzelm@33411
   171
val worker_trend = Unsynchronized.ref 0;
wenzelm@33410
   172
wenzelm@50280
   173
val status_ticks = Unsynchronized.ref 0;
wenzelm@50280
   174
val last_round = Unsynchronized.ref Time.zeroTime;
wenzelm@50280
   175
val next_round = seconds 0.05;
wenzelm@50280
   176
wenzelm@33410
   177
datatype worker_state = Working | Waiting | Sleeping;
wenzelm@33410
   178
val workers = Unsynchronized.ref ([]: (Thread.thread * worker_state Unsynchronized.ref) list);
wenzelm@33410
   179
wenzelm@33410
   180
fun count_workers state = (*requires SYNCHRONIZED*)
wenzelm@33410
   181
  fold (fn (_, state_ref) => fn i => if ! state_ref = state then i + 1 else i) (! workers) 0;
wenzelm@33410
   182
wenzelm@33410
   183
wenzelm@50280
   184
wenzelm@50280
   185
(* status *)
wenzelm@50280
   186
wenzelm@50280
   187
val ML_statistics = Unsynchronized.ref false;
wenzelm@50280
   188
wenzelm@50280
   189
fun report_status () = (*requires SYNCHRONIZED*)
wenzelm@50280
   190
  if ! ML_statistics then
wenzelm@50280
   191
    let
wenzelm@50280
   192
      val {ready, pending, running, passive} = Task_Queue.status (! queue);
wenzelm@50280
   193
      val total = length (! workers);
wenzelm@50280
   194
      val active = count_workers Working;
wenzelm@50280
   195
      val waiting = count_workers Waiting;
wenzelm@50280
   196
      val stats =
wenzelm@50280
   197
       [("now", signed_string_of_real (Time.toReal (Time.now ()))),
wenzelm@50280
   198
        ("tasks_ready", Markup.print_int ready),
wenzelm@50280
   199
        ("tasks_pending", Markup.print_int pending),
wenzelm@50280
   200
        ("tasks_running", Markup.print_int running),
wenzelm@50280
   201
        ("tasks_passive", Markup.print_int passive),
wenzelm@50280
   202
        ("workers_total", Markup.print_int total),
wenzelm@50280
   203
        ("workers_active", Markup.print_int active),
wenzelm@50280
   204
        ("workers_waiting", Markup.print_int waiting)] @
wenzelm@50280
   205
        ML_Statistics.get ();
wenzelm@50664
   206
    in
wenzelm@50683
   207
      Output.protocol_message (Markup.ML_statistics :: stats) ""
wenzelm@50664
   208
        handle Fail msg => warning msg
wenzelm@50664
   209
    end
wenzelm@50280
   210
  else ();
wenzelm@50280
   211
wenzelm@50280
   212
wenzelm@44110
   213
(* cancellation primitives *)
wenzelm@32099
   214
wenzelm@34279
   215
fun cancel_now group = (*requires SYNCHRONIZED*)
wenzelm@44341
   216
  let
wenzelm@47404
   217
    val running = Task_Queue.cancel (! queue) group;
wenzelm@49894
   218
    val _ = running |> List.app (fn thread =>
wenzelm@49894
   219
      if Simple_Thread.is_self thread then ()
wenzelm@49894
   220
      else Simple_Thread.interrupt_unsynchronized thread);
wenzelm@47404
   221
  in running end;
wenzelm@44341
   222
wenzelm@44341
   223
fun cancel_all () = (*requires SYNCHRONIZED*)
wenzelm@44341
   224
  let
wenzelm@44341
   225
    val (groups, threads) = Task_Queue.cancel_all (! queue);
wenzelm@44341
   226
    val _ = List.app Simple_Thread.interrupt_unsynchronized threads;
wenzelm@44341
   227
  in groups end;
wenzelm@34279
   228
wenzelm@34279
   229
fun cancel_later group = (*requires SYNCHRONIZED*)
wenzelm@32738
   230
 (Unsynchronized.change canceled (insert Task_Queue.eq_group group);
wenzelm@32738
   231
  broadcast scheduler_event);
wenzelm@29341
   232
wenzelm@44301
   233
fun interruptible_task f x =
wenzelm@44301
   234
  (if Multithreading.available then
wenzelm@44301
   235
    Multithreading.with_attributes
wenzelm@44301
   236
      (if is_some (worker_task ())
wenzelm@44301
   237
       then Multithreading.private_interrupts
wenzelm@44301
   238
       else Multithreading.public_interrupts)
wenzelm@44301
   239
      (fn _ => f x)
wenzelm@44301
   240
   else interruptible f x)
wenzelm@44301
   241
  before Multithreading.interrupted ();
wenzelm@44301
   242
wenzelm@44301
   243
wenzelm@44110
   244
(* worker threads *)
wenzelm@44110
   245
wenzelm@44110
   246
fun worker_exec (task, jobs) =
wenzelm@28167
   247
  let
wenzelm@41683
   248
    val group = Task_Queue.group_of_task task;
wenzelm@32102
   249
    val valid = not (Task_Queue.is_canceled group);
wenzelm@41670
   250
    val ok =
wenzelm@41670
   251
      Task_Queue.running task (fn () =>
wenzelm@41683
   252
        setmp_worker_task task (fn () =>
wenzelm@41670
   253
          fold (fn job => fn ok => job valid andalso ok) jobs true) ());
wenzelm@41776
   254
    val _ = Multithreading.tracing 2 (fn () =>
wenzelm@41670
   255
      let
wenzelm@43951
   256
        val s = Task_Queue.str_of_task_groups task;
wenzelm@41670
   257
        fun micros time = string_of_int (Time.toNanoseconds time div 1000);
wenzelm@41680
   258
        val (run, wait, deps) = Task_Queue.timing_of_task task;
wenzelm@41680
   259
      in "TASK " ^ s ^ " " ^ micros run ^ " " ^ micros wait ^ " (" ^ commas deps ^ ")" end);
wenzelm@32246
   260
    val _ = SYNCHRONIZED "finish" (fn () =>
wenzelm@32219
   261
      let
wenzelm@32738
   262
        val maximal = Unsynchronized.change_result queue (Task_Queue.finish task);
wenzelm@44295
   263
        val test = Exn.capture Multithreading.interrupted ();
wenzelm@32219
   264
        val _ =
wenzelm@44295
   265
          if ok andalso not (Exn.is_interrupt_exn test) then ()
wenzelm@44299
   266
          else if null (cancel_now group) then ()
wenzelm@34279
   267
          else cancel_later group;
wenzelm@32219
   268
        val _ = broadcast work_finished;
wenzelm@33413
   269
        val _ = if maximal then () else signal work_available;
wenzelm@32219
   270
      in () end);
wenzelm@28167
   271
  in () end;
wenzelm@28167
   272
wenzelm@33410
   273
fun worker_wait active cond = (*requires SYNCHRONIZED*)
wenzelm@33406
   274
  let
wenzelm@33410
   275
    val state =
wenzelm@33410
   276
      (case AList.lookup Thread.equal (! workers) (Thread.self ()) of
wenzelm@33410
   277
        SOME state => state
wenzelm@33410
   278
      | NONE => raise Fail "Unregistered worker thread");
wenzelm@33410
   279
    val _ = state := (if active then Waiting else Sleeping);
wenzelm@33406
   280
    val _ = wait cond;
wenzelm@33410
   281
    val _ = state := Working;
wenzelm@33406
   282
  in () end;
wenzelm@28162
   283
wenzelm@33415
   284
fun worker_next () = (*requires SYNCHRONIZED*)
wenzelm@33406
   285
  if length (! workers) > ! max_workers then
wenzelm@33406
   286
    (Unsynchronized.change workers (AList.delete Thread.equal (Thread.self ()));
wenzelm@33415
   287
     signal work_available;
wenzelm@28167
   288
     NONE)
wenzelm@33410
   289
  else if count_workers Working > ! max_active then
wenzelm@33415
   290
    (worker_wait false work_available; worker_next ())
wenzelm@28166
   291
  else
wenzelm@32738
   292
    (case Unsynchronized.change_result queue (Task_Queue.dequeue (Thread.self ())) of
wenzelm@33415
   293
      NONE => (worker_wait false work_available; worker_next ())
wenzelm@33413
   294
    | some => (signal work_available; some));
wenzelm@28156
   295
wenzelm@28167
   296
fun worker_loop name =
wenzelm@33415
   297
  (case SYNCHRONIZED name (fn () => worker_next ()) of
wenzelm@29119
   298
    NONE => ()
wenzelm@44295
   299
  | SOME work => (worker_exec work; worker_loop name));
wenzelm@28156
   300
wenzelm@33407
   301
fun worker_start name = (*requires SYNCHRONIZED*)
wenzelm@37216
   302
  Unsynchronized.change workers (cons (Simple_Thread.fork false (fn () => worker_loop name),
wenzelm@33410
   303
    Unsynchronized.ref Working));
wenzelm@28156
   304
wenzelm@28156
   305
wenzelm@28156
   306
(* scheduler *)
wenzelm@28156
   307
wenzelm@28206
   308
fun scheduler_next () = (*requires SYNCHRONIZED*)
wenzelm@28156
   309
  let
wenzelm@33407
   310
    val now = Time.now ();
wenzelm@33407
   311
    val tick = Time.<= (Time.+ (! last_round, next_round), now);
wenzelm@33407
   312
    val _ = if tick then last_round := now else ();
wenzelm@33407
   313
wenzelm@33415
   314
wenzelm@50280
   315
    (* runtime status *)
wenzelm@33415
   316
wenzelm@32226
   317
    val _ =
wenzelm@33407
   318
      if tick then Unsynchronized.change status_ticks (fn i => (i + 1) mod 10) else ();
wenzelm@33407
   319
    val _ =
wenzelm@50280
   320
      if tick andalso ! status_ticks = 0 then report_status () else ();
wenzelm@32053
   321
wenzelm@28191
   322
    val _ =
wenzelm@32219
   323
      if forall (Thread.isActive o #1) (! workers) then ()
wenzelm@32095
   324
      else
wenzelm@33409
   325
        let
wenzelm@37682
   326
          val (alive, dead) = List.partition (Thread.isActive o #1) (! workers);
wenzelm@33409
   327
          val _ = workers := alive;
wenzelm@33409
   328
        in
wenzelm@33409
   329
          Multithreading.tracing 0 (fn () =>
wenzelm@33409
   330
            "SCHEDULE: disposed " ^ string_of_int (length dead) ^ " dead worker threads")
wenzelm@33409
   331
        end;
wenzelm@28191
   332
wenzelm@33415
   333
wenzelm@33415
   334
    (* worker pool adjustments *)
wenzelm@33415
   335
wenzelm@33415
   336
    val max_active0 = ! max_active;
wenzelm@33415
   337
    val max_workers0 = ! max_workers;
wenzelm@33415
   338
wenzelm@28206
   339
    val m = if ! do_shutdown then 0 else Multithreading.max_threads_value ();
wenzelm@33406
   340
    val _ = max_active := m;
wenzelm@33406
   341
wenzelm@33411
   342
    val mm =
wenzelm@33411
   343
      if ! do_shutdown then 0
wenzelm@33411
   344
      else if m = 9999 then 1
wenzelm@33413
   345
      else Int.min (Int.max (count_workers Working + 2 * count_workers Waiting, m), 4 * m);
wenzelm@33411
   346
    val _ =
wenzelm@33411
   347
      if tick andalso mm > ! max_workers then
wenzelm@33411
   348
        Unsynchronized.change worker_trend (fn w => if w < 0 then 0 else w + 1)
wenzelm@33411
   349
      else if tick andalso mm < ! max_workers then
wenzelm@33411
   350
        Unsynchronized.change worker_trend (fn w => if w > 0 then 0 else w - 1)
wenzelm@33411
   351
      else ();
wenzelm@33411
   352
    val _ =
wenzelm@33415
   353
      if mm = 0 orelse ! worker_trend > 50 orelse ! worker_trend < ~50 then
wenzelm@33415
   354
        max_workers := mm
wenzelm@44173
   355
      else if ! worker_trend > 5 andalso ! max_workers < 2 * m orelse ! max_workers = 0 then
wenzelm@33415
   356
        max_workers := Int.min (mm, 2 * m)
wenzelm@33411
   357
      else ();
wenzelm@33406
   358
wenzelm@33407
   359
    val missing = ! max_workers - length (! workers);
wenzelm@28203
   360
    val _ =
wenzelm@33407
   361
      if missing > 0 then
wenzelm@33415
   362
        funpow missing (fn () =>
wenzelm@33415
   363
          ignore (worker_start ("worker " ^ string_of_int (Unsynchronized.inc next)))) ()
wenzelm@28203
   364
      else ();
wenzelm@28206
   365
wenzelm@33415
   366
    val _ =
wenzelm@33415
   367
      if ! max_active = max_active0 andalso ! max_workers = max_workers0 then ()
wenzelm@33415
   368
      else signal work_available;
wenzelm@33415
   369
wenzelm@33415
   370
wenzelm@33415
   371
    (* canceled groups *)
wenzelm@33415
   372
wenzelm@32225
   373
    val _ =
wenzelm@32225
   374
      if null (! canceled) then ()
wenzelm@32293
   375
      else
wenzelm@32293
   376
       (Multithreading.tracing 1 (fn () =>
wenzelm@32293
   377
          string_of_int (length (! canceled)) ^ " canceled groups");
wenzelm@44299
   378
        Unsynchronized.change canceled (filter_out (null o cancel_now));
wenzelm@32293
   379
        broadcast_work ());
wenzelm@28206
   380
wenzelm@33415
   381
wenzelm@33415
   382
    (* delay loop *)
wenzelm@33415
   383
wenzelm@32295
   384
    val _ = Exn.release (wait_timeout next_round scheduler_event);
wenzelm@28167
   385
wenzelm@33415
   386
wenzelm@33415
   387
    (* shutdown *)
wenzelm@33415
   388
wenzelm@34277
   389
    val _ = if Task_Queue.all_passive (! queue) then do_shutdown := true else ();
wenzelm@32219
   390
    val continue = not (! do_shutdown andalso null (! workers));
wenzelm@50429
   391
    val _ = if continue then () else (report_status (); scheduler := NONE);
wenzelm@33415
   392
wenzelm@32219
   393
    val _ = broadcast scheduler_event;
wenzelm@32295
   394
  in continue end
wenzelm@39232
   395
  handle exn =>
wenzelm@39232
   396
    if Exn.is_interrupt exn then
wenzelm@39232
   397
     (Multithreading.tracing 1 (fn () => "Interrupt");
wenzelm@44341
   398
      List.app cancel_later (cancel_all ());
wenzelm@39232
   399
      broadcast_work (); true)
wenzelm@39232
   400
    else reraise exn;
wenzelm@32295
   401
wenzelm@28206
   402
fun scheduler_loop () =
wenzelm@44173
   403
 (while
wenzelm@33416
   404
    Multithreading.with_attributes
wenzelm@33416
   405
      (Multithreading.sync_interrupts Multithreading.public_interrupts)
wenzelm@33416
   406
      (fn _ => SYNCHRONIZED "scheduler" (fn () => scheduler_next ()))
wenzelm@50429
   407
  do (); last_round := Time.zeroTime);
wenzelm@28156
   408
wenzelm@28203
   409
fun scheduler_active () = (*requires SYNCHRONIZED*)
wenzelm@28203
   410
  (case ! scheduler of NONE => false | SOME thread => Thread.isActive thread);
wenzelm@28203
   411
wenzelm@32228
   412
fun scheduler_check () = (*requires SYNCHRONIZED*)
wenzelm@32228
   413
 (do_shutdown := false;
wenzelm@32248
   414
  if scheduler_active () then ()
wenzelm@37216
   415
  else scheduler := SOME (Simple_Thread.fork false scheduler_loop));
wenzelm@28156
   416
wenzelm@44301
   417
wenzelm@44301
   418
wenzelm@44301
   419
(** futures **)
wenzelm@44301
   420
wenzelm@44301
   421
(* cancel *)
wenzelm@44301
   422
wenzelm@49906
   423
fun cancel_group_unsynchronized group = (*requires SYNCHRONIZED*)
wenzelm@44299
   424
  let
wenzelm@47421
   425
    val _ = if null (cancel_now group) then () else cancel_later group;
wenzelm@47421
   426
    val _ = signal work_available;
wenzelm@47421
   427
    val _ = scheduler_check ();
wenzelm@49906
   428
  in () end;
wenzelm@49906
   429
wenzelm@49906
   430
fun cancel_group group =
wenzelm@49906
   431
  SYNCHRONIZED "cancel_group" (fn () => cancel_group_unsynchronized group);
wenzelm@44299
   432
wenzelm@44301
   433
fun cancel x = cancel_group (Task_Queue.group_of_task (task_of x));
wenzelm@29366
   434
wenzelm@28156
   435
wenzelm@44110
   436
(* future jobs *)
wenzelm@44110
   437
wenzelm@44249
   438
fun assign_result group result raw_res =
wenzelm@44110
   439
  let
wenzelm@44249
   440
    val res =
wenzelm@44249
   441
      (case raw_res of
wenzelm@50911
   442
        Exn.Exn exn => Exn.Exn (Par_Exn.identify [] exn)
wenzelm@44249
   443
      | _ => raw_res);
wenzelm@44110
   444
    val _ = Single_Assignment.assign result res
wenzelm@44110
   445
      handle exn as Fail _ =>
wenzelm@44110
   446
        (case Single_Assignment.peek result of
wenzelm@44110
   447
          SOME (Exn.Exn e) => reraise (if Exn.is_interrupt e then e else exn)
wenzelm@44110
   448
        | _ => reraise exn);
wenzelm@44110
   449
    val ok =
wenzelm@44110
   450
      (case the (Single_Assignment.peek result) of
wenzelm@44111
   451
        Exn.Exn exn =>
wenzelm@44111
   452
          (SYNCHRONIZED "cancel" (fn () => Task_Queue.cancel_group group exn); false)
wenzelm@44110
   453
      | Exn.Res _ => true);
wenzelm@44110
   454
  in ok end;
wenzelm@44110
   455
wenzelm@44113
   456
fun future_job group interrupts (e: unit -> 'a) =
wenzelm@44110
   457
  let
wenzelm@44110
   458
    val result = Single_Assignment.var "future" : 'a result;
wenzelm@44110
   459
    val pos = Position.thread_data ();
wenzelm@44110
   460
    fun job ok =
wenzelm@44110
   461
      let
wenzelm@44110
   462
        val res =
wenzelm@44110
   463
          if ok then
wenzelm@44110
   464
            Exn.capture (fn () =>
wenzelm@44113
   465
              Multithreading.with_attributes
wenzelm@44113
   466
                (if interrupts
wenzelm@44113
   467
                 then Multithreading.private_interrupts else Multithreading.no_interrupts)
wenzelm@44295
   468
                (fn _ => Position.setmp_thread_data pos e ())) ()
wenzelm@44110
   469
          else Exn.interrupt_exn;
wenzelm@44110
   470
      in assign_result group result res end;
wenzelm@44110
   471
  in (result, job) end;
wenzelm@44110
   472
wenzelm@44110
   473
wenzelm@29366
   474
(* fork *)
wenzelm@29366
   475
wenzelm@44427
   476
type params = {name: string, group: group option, deps: task list, pri: int, interrupts: bool};
wenzelm@44427
   477
val default_params: params = {name = "", group = NONE, deps = [], pri = 0, interrupts = true};
wenzelm@44113
   478
wenzelm@44427
   479
fun forks ({name, group, deps, pri, interrupts}: params) es =
wenzelm@41674
   480
  if null es then []
wenzelm@41674
   481
  else
wenzelm@41674
   482
    let
wenzelm@41674
   483
      val grp =
wenzelm@41674
   484
        (case group of
wenzelm@41674
   485
          NONE => worker_subgroup ()
wenzelm@41674
   486
        | SOME grp => grp);
wenzelm@41708
   487
      fun enqueue e queue =
wenzelm@41674
   488
        let
wenzelm@44113
   489
          val (result, job) = future_job grp interrupts e;
wenzelm@41708
   490
          val (task, queue') = Task_Queue.enqueue name grp deps pri job queue;
wenzelm@41683
   491
          val future = Future {promised = false, task = task, result = result};
wenzelm@41708
   492
        in (future, queue') end;
wenzelm@41674
   493
    in
wenzelm@41674
   494
      SYNCHRONIZED "enqueue" (fn () =>
wenzelm@41674
   495
        let
wenzelm@41708
   496
          val (futures, queue') = fold_map enqueue es (! queue);
wenzelm@41708
   497
          val _ = queue := queue';
wenzelm@41708
   498
          val minimal = forall (not o Task_Queue.known_task queue') deps;
wenzelm@41674
   499
          val _ = if minimal then signal work_available else ();
wenzelm@41674
   500
          val _ = scheduler_check ();
wenzelm@41674
   501
        in futures end)
wenzelm@41674
   502
    end;
wenzelm@28162
   503
wenzelm@44113
   504
fun fork_pri pri e =
wenzelm@44301
   505
  (singleton o forks) {name = "fork", group = NONE, deps = [], pri = pri, interrupts = true} e;
wenzelm@44113
   506
wenzelm@41672
   507
fun fork e = fork_pri 0 e;
wenzelm@28186
   508
wenzelm@28186
   509
wenzelm@29366
   510
(* join *)
wenzelm@29366
   511
wenzelm@32099
   512
fun get_result x =
wenzelm@32099
   513
  (case peek x of
wenzelm@37852
   514
    NONE => Exn.Exn (Fail "Unfinished future")
wenzelm@39232
   515
  | SOME res =>
wenzelm@39232
   516
      if Exn.is_interrupt_exn res then
wenzelm@44247
   517
        (case Task_Queue.group_status (Task_Queue.group_of_task (task_of x)) of
wenzelm@44247
   518
          NONE => res
wenzelm@44247
   519
        | SOME exn => Exn.Exn exn)
wenzelm@39232
   520
      else res);
wenzelm@28186
   521
wenzelm@49935
   522
local
wenzelm@49935
   523
wenzelm@32095
   524
fun join_next deps = (*requires SYNCHRONIZED*)
wenzelm@41695
   525
  if null deps then NONE
wenzelm@32224
   526
  else
wenzelm@41681
   527
    (case Unsynchronized.change_result queue (Task_Queue.dequeue_deps (Thread.self ()) deps) of
wenzelm@41695
   528
      (NONE, []) => NONE
wenzelm@41695
   529
    | (NONE, deps') =>
wenzelm@41695
   530
        (worker_waiting deps' (fn () => worker_wait true work_finished); join_next deps')
wenzelm@32224
   531
    | (SOME work, deps') => SOME (work, deps'));
wenzelm@32095
   532
wenzelm@32814
   533
fun execute_work NONE = ()
wenzelm@44110
   534
  | execute_work (SOME (work, deps')) =
wenzelm@44110
   535
      (worker_joining (fn () => worker_exec work); join_work deps')
wenzelm@32814
   536
and join_work deps =
wenzelm@43538
   537
  Multithreading.with_attributes Multithreading.no_interrupts
wenzelm@43538
   538
    (fn _ => execute_work (SYNCHRONIZED "join" (fn () => join_next deps)));
wenzelm@32814
   539
wenzelm@29551
   540
in
wenzelm@29551
   541
wenzelm@29366
   542
fun join_results xs =
wenzelm@41679
   543
  let
wenzelm@41679
   544
    val _ =
wenzelm@41679
   545
      if forall is_finished xs then ()
wenzelm@41679
   546
      else if Multithreading.self_critical () then
wenzelm@41679
   547
        error "Cannot join future values within critical section"
wenzelm@41695
   548
      else if is_some (worker_task ()) then join_work (map task_of xs)
wenzelm@41681
   549
      else List.app (ignore o Single_Assignment.await o result_of) xs;
wenzelm@41679
   550
  in map get_result xs end;
wenzelm@28186
   551
wenzelm@29551
   552
end;
wenzelm@29551
   553
wenzelm@28647
   554
fun join_result x = singleton join_results x;
wenzelm@44330
   555
fun joins xs = Par_Exn.release_all (join_results xs);
wenzelm@28647
   556
fun join x = Exn.release (join_result x);
wenzelm@28156
   557
wenzelm@29366
   558
wenzelm@44110
   559
(* fast-path versions -- bypassing task queue *)
wenzelm@34277
   560
wenzelm@44294
   561
fun value_result (res: 'a Exn.result) =
wenzelm@34277
   562
  let
wenzelm@45136
   563
    val task = Task_Queue.dummy_task;
wenzelm@41683
   564
    val group = Task_Queue.group_of_task task;
wenzelm@35016
   565
    val result = Single_Assignment.var "value" : 'a result;
wenzelm@44294
   566
    val _ = assign_result group result res;
wenzelm@41683
   567
  in Future {promised = false, task = task, result = result} end;
wenzelm@29366
   568
wenzelm@44294
   569
fun value x = value_result (Exn.Res x);
wenzelm@44294
   570
wenzelm@44330
   571
fun cond_forks args es =
wenzelm@44330
   572
  if Multithreading.enabled () then forks args es
wenzelm@44330
   573
  else map (fn e => value_result (Exn.interruptible_capture e ())) es;
wenzelm@44330
   574
wenzelm@29384
   575
fun map_future f x =
wenzelm@49935
   576
  if is_finished x then value (f (join x))
wenzelm@49935
   577
  else
wenzelm@49935
   578
    let
wenzelm@49935
   579
      val task = task_of x;
wenzelm@49935
   580
      val group = Task_Queue.group_of_task task;
wenzelm@49935
   581
      val (result, job) = future_job group true (fn () => f (join x));
wenzelm@29384
   582
wenzelm@49935
   583
      val extended = SYNCHRONIZED "extend" (fn () =>
wenzelm@49935
   584
        (case Task_Queue.extend task job (! queue) of
wenzelm@49935
   585
          SOME queue' => (queue := queue'; true)
wenzelm@49935
   586
        | NONE => false));
wenzelm@49935
   587
    in
wenzelm@49935
   588
      if extended then Future {promised = false, task = task, result = result}
wenzelm@49935
   589
      else
wenzelm@49935
   590
        (singleton o cond_forks)
wenzelm@49935
   591
          {name = "map_future", group = SOME group, deps = [task],
wenzelm@49935
   592
            pri = Task_Queue.pri_of_task task, interrupts = true}
wenzelm@49935
   593
          (fn () => f (join x))
wenzelm@49935
   594
    end;
wenzelm@28979
   595
wenzelm@28191
   596
wenzelm@34277
   597
(* promised futures -- fulfilled by external means *)
wenzelm@34277
   598
wenzelm@44298
   599
fun promise_group group abort : 'a future =
wenzelm@34277
   600
  let
wenzelm@35016
   601
    val result = Single_Assignment.var "promise" : 'a result;
wenzelm@44298
   602
    fun assign () = assign_result group result Exn.interrupt_exn
wenzelm@39243
   603
      handle Fail _ => true
wenzelm@39243
   604
        | exn =>
wenzelm@44298
   605
            if Exn.is_interrupt exn
wenzelm@44298
   606
            then raise Fail "Concurrent attempt to fulfill promise"
wenzelm@39243
   607
            else reraise exn;
wenzelm@44298
   608
    fun job () =
wenzelm@44298
   609
      Multithreading.with_attributes Multithreading.no_interrupts
wenzelm@47423
   610
        (fn _ => Exn.release (Exn.capture assign () before abort ()));
wenzelm@37854
   611
    val task = SYNCHRONIZED "enqueue_passive" (fn () =>
wenzelm@44298
   612
      Unsynchronized.change_result queue (Task_Queue.enqueue_passive group job));
wenzelm@41683
   613
  in Future {promised = true, task = task, result = result} end;
wenzelm@34277
   614
wenzelm@44298
   615
fun promise abort = promise_group (worker_subgroup ()) abort;
wenzelm@34277
   616
wenzelm@41683
   617
fun fulfill_result (Future {promised, task, result}) res =
wenzelm@39243
   618
  if not promised then raise Fail "Not a promised future"
wenzelm@39243
   619
  else
wenzelm@39243
   620
    let
wenzelm@41683
   621
      val group = Task_Queue.group_of_task task;
wenzelm@39243
   622
      fun job ok = assign_result group result (if ok then res else Exn.interrupt_exn);
wenzelm@39243
   623
      val _ =
wenzelm@39243
   624
        Multithreading.with_attributes Multithreading.no_interrupts (fn _ =>
wenzelm@39243
   625
          let
wenzelm@47423
   626
            val passive_job =
wenzelm@39243
   627
              SYNCHRONIZED "fulfill_result" (fn () =>
wenzelm@39243
   628
                Unsynchronized.change_result queue
wenzelm@39243
   629
                  (Task_Queue.dequeue_passive (Thread.self ()) task));
wenzelm@47423
   630
          in
wenzelm@47423
   631
            (case passive_job of
wenzelm@47423
   632
              SOME true => worker_exec (task, [job])
wenzelm@47423
   633
            | SOME false => ()
wenzelm@47423
   634
            | NONE => ignore (job (not (Task_Queue.is_canceled group))))
wenzelm@47423
   635
          end);
wenzelm@41681
   636
      val _ =
wenzelm@41695
   637
        if is_some (Single_Assignment.peek result) then ()
wenzelm@41695
   638
        else worker_waiting [task] (fn () => ignore (Single_Assignment.await result));
wenzelm@39243
   639
    in () end;
wenzelm@34277
   640
wenzelm@43761
   641
fun fulfill x res = fulfill_result x (Exn.Res res);
wenzelm@34277
   642
wenzelm@34277
   643
wenzelm@49906
   644
(* terminate *)
wenzelm@49906
   645
wenzelm@49906
   646
fun terminate group =
wenzelm@49906
   647
  let
wenzelm@49906
   648
    val tasks =
wenzelm@49906
   649
      SYNCHRONIZED "terminate" (fn () =>
wenzelm@49906
   650
        let val _ = cancel_group_unsynchronized group;
wenzelm@49906
   651
        in Task_Queue.group_tasks (! queue) group end);
wenzelm@49906
   652
  in
wenzelm@49906
   653
    if null tasks then ()
wenzelm@49906
   654
    else
wenzelm@49906
   655
      (singleton o forks)
wenzelm@49906
   656
        {name = "terminate", group = SOME (new_group NONE),
wenzelm@49906
   657
          deps = tasks, pri = 0, interrupts = false} I
wenzelm@49906
   658
      |> join
wenzelm@49906
   659
  end;
wenzelm@49906
   660
wenzelm@49906
   661
wenzelm@32228
   662
(* shutdown *)
wenzelm@29366
   663
wenzelm@28203
   664
fun shutdown () =
wenzelm@28276
   665
  if Multithreading.available then
wenzelm@28276
   666
    SYNCHRONIZED "shutdown" (fn () =>
wenzelm@32228
   667
     while scheduler_active () do
wenzelm@34279
   668
      (wait scheduler_event; broadcast_work ()))
wenzelm@28276
   669
  else ();
wenzelm@28203
   670
wenzelm@29366
   671
wenzelm@29366
   672
(*final declarations of this structure!*)
wenzelm@29366
   673
val map = map_future;
wenzelm@29366
   674
wenzelm@28156
   675
end;
wenzelm@28972
   676
wenzelm@28972
   677
type 'a future = 'a Future.future;
wenzelm@28972
   678