src/Pure/Concurrent/future.ML
author wenzelm
Tue Mar 05 11:37:01 2013 +0100 (2013-03-05)
changeset 51354 45579fbe5a24
parent 51325 bcd6b1aa4db5
child 51637 ff2d241dcde1
permissions -rw-r--r--
removed unused Future.flat, while leaving its influence of Future.map (see bcd6b1aa4db5);
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@50974
    55
  val forked_proofs: int Unsynchronized.ref
wenzelm@44301
    56
  val interruptible_task: ('a -> 'b) -> 'a -> 'b
wenzelm@47404
    57
  val cancel_group: group -> unit
wenzelm@47404
    58
  val cancel: 'a future -> unit
wenzelm@50914
    59
  val error_msg: Position.T -> (serial * string) * string option -> unit
wenzelm@50914
    60
  val identify_result: Position.T -> 'a Exn.result -> 'a Exn.result
wenzelm@44427
    61
  type params = {name: string, group: group option, deps: task list, pri: int, interrupts: bool}
wenzelm@44427
    62
  val default_params: params
wenzelm@44427
    63
  val forks: params -> (unit -> 'a) list -> 'a future list
wenzelm@32724
    64
  val fork: (unit -> 'a) -> 'a future
wenzelm@28972
    65
  val join_results: 'a future list -> 'a Exn.result list
wenzelm@28972
    66
  val join_result: 'a future -> 'a Exn.result
wenzelm@44330
    67
  val joins: 'a future list -> 'a list
wenzelm@28972
    68
  val join: 'a future -> 'a
wenzelm@44294
    69
  val value_result: 'a Exn.result -> 'a future
wenzelm@34277
    70
  val value: 'a -> 'a future
wenzelm@44427
    71
  val cond_forks: params -> (unit -> 'a) list -> 'a future list
wenzelm@28972
    72
  val map: ('a -> 'b) -> 'a future -> 'b future
wenzelm@44300
    73
  val promise_group: group -> (unit -> unit) -> 'a future
wenzelm@44298
    74
  val promise: (unit -> unit) -> 'a future
wenzelm@34277
    75
  val fulfill_result: 'a future -> 'a Exn.result -> unit
wenzelm@34277
    76
  val fulfill: 'a future -> 'a -> unit
wenzelm@49906
    77
  val terminate: group -> unit
wenzelm@28203
    78
  val shutdown: unit -> unit
wenzelm@28156
    79
end;
wenzelm@28156
    80
wenzelm@28156
    81
structure Future: FUTURE =
wenzelm@28156
    82
struct
wenzelm@28156
    83
wenzelm@28177
    84
(** future values **)
wenzelm@28177
    85
wenzelm@44300
    86
type task = Task_Queue.task;
wenzelm@44300
    87
type group = Task_Queue.group;
wenzelm@44300
    88
val new_group = Task_Queue.new_group;
wenzelm@44300
    89
wenzelm@44300
    90
wenzelm@28167
    91
(* identifiers *)
wenzelm@28167
    92
wenzelm@32058
    93
local
wenzelm@44300
    94
  val tag = Universal.tag () : task option Universal.tag;
wenzelm@32058
    95
in
wenzelm@41683
    96
  fun worker_task () = the_default NONE (Thread.getLocal tag);
wenzelm@44110
    97
  fun setmp_worker_task task f x = setmp_thread_data tag (worker_task ()) (SOME task) f x;
wenzelm@28167
    98
end;
wenzelm@28167
    99
wenzelm@41683
   100
val worker_group = Option.map Task_Queue.group_of_task o worker_task;
wenzelm@44300
   101
fun worker_subgroup () = new_group (worker_group ());
wenzelm@34277
   102
wenzelm@41679
   103
fun worker_joining e =
wenzelm@41679
   104
  (case worker_task () of
wenzelm@41679
   105
    NONE => e ()
wenzelm@41679
   106
  | SOME task => Task_Queue.joining task e);
wenzelm@41679
   107
wenzelm@41680
   108
fun worker_waiting deps e =
wenzelm@41670
   109
  (case worker_task () of
wenzelm@41670
   110
    NONE => e ()
wenzelm@41680
   111
  | SOME task => Task_Queue.waiting task deps e);
wenzelm@41670
   112
wenzelm@28167
   113
wenzelm@28167
   114
(* datatype future *)
wenzelm@28167
   115
wenzelm@35016
   116
type 'a result = 'a Exn.result Single_Assignment.var;
wenzelm@35016
   117
wenzelm@28972
   118
datatype 'a future = Future of
wenzelm@34277
   119
 {promised: bool,
wenzelm@44300
   120
  task: task,
wenzelm@35016
   121
  result: 'a result};
wenzelm@28167
   122
wenzelm@28167
   123
fun task_of (Future {task, ...}) = task;
wenzelm@32253
   124
fun result_of (Future {result, ...}) = result;
wenzelm@28167
   125
wenzelm@35016
   126
fun peek x = Single_Assignment.peek (result_of x);
wenzelm@28558
   127
fun is_finished x = is_some (peek x);
wenzelm@28320
   128
wenzelm@28167
   129
wenzelm@28177
   130
wenzelm@28177
   131
(** scheduling **)
wenzelm@28177
   132
wenzelm@28177
   133
(* synchronization *)
wenzelm@28156
   134
wenzelm@32219
   135
val scheduler_event = ConditionVar.conditionVar ();
wenzelm@32219
   136
val work_available = ConditionVar.conditionVar ();
wenzelm@32219
   137
val work_finished = ConditionVar.conditionVar ();
wenzelm@32219
   138
wenzelm@28156
   139
local
wenzelm@28156
   140
  val lock = Mutex.mutex ();
wenzelm@28156
   141
in
wenzelm@28156
   142
wenzelm@37216
   143
fun SYNCHRONIZED name = Simple_Thread.synchronized name lock;
wenzelm@28156
   144
wenzelm@32219
   145
fun wait cond = (*requires SYNCHRONIZED*)
wenzelm@32295
   146
  Multithreading.sync_wait NONE NONE cond lock;
wenzelm@28206
   147
wenzelm@32295
   148
fun wait_timeout timeout cond = (*requires SYNCHRONIZED*)
wenzelm@32295
   149
  Multithreading.sync_wait NONE (SOME (Time.+ (Time.now (), timeout))) cond lock;
wenzelm@28166
   150
wenzelm@32219
   151
fun signal cond = (*requires SYNCHRONIZED*)
wenzelm@32219
   152
  ConditionVar.signal cond;
wenzelm@32219
   153
wenzelm@32219
   154
fun broadcast cond = (*requires SYNCHRONIZED*)
wenzelm@28166
   155
  ConditionVar.broadcast cond;
wenzelm@28156
   156
wenzelm@28156
   157
end;
wenzelm@28156
   158
wenzelm@28156
   159
wenzelm@33410
   160
(* global state *)
wenzelm@33410
   161
wenzelm@33410
   162
val queue = Unsynchronized.ref Task_Queue.empty;
wenzelm@33410
   163
val next = Unsynchronized.ref 0;
wenzelm@33410
   164
val scheduler = Unsynchronized.ref (NONE: Thread.thread option);
wenzelm@44300
   165
val canceled = Unsynchronized.ref ([]: group list);
wenzelm@33410
   166
val do_shutdown = Unsynchronized.ref false;
wenzelm@33410
   167
val max_workers = Unsynchronized.ref 0;
wenzelm@33410
   168
val max_active = Unsynchronized.ref 0;
wenzelm@33411
   169
val worker_trend = Unsynchronized.ref 0;
wenzelm@33410
   170
wenzelm@50280
   171
val status_ticks = Unsynchronized.ref 0;
wenzelm@50280
   172
val last_round = Unsynchronized.ref Time.zeroTime;
wenzelm@50280
   173
val next_round = seconds 0.05;
wenzelm@50280
   174
wenzelm@33410
   175
datatype worker_state = Working | Waiting | Sleeping;
wenzelm@33410
   176
val workers = Unsynchronized.ref ([]: (Thread.thread * worker_state Unsynchronized.ref) list);
wenzelm@33410
   177
wenzelm@33410
   178
fun count_workers state = (*requires SYNCHRONIZED*)
wenzelm@33410
   179
  fold (fn (_, state_ref) => fn i => if ! state_ref = state then i + 1 else i) (! workers) 0;
wenzelm@33410
   180
wenzelm@33410
   181
wenzelm@50280
   182
wenzelm@50280
   183
(* status *)
wenzelm@50280
   184
wenzelm@50280
   185
val ML_statistics = Unsynchronized.ref false;
wenzelm@50974
   186
val forked_proofs = Unsynchronized.ref 0;
wenzelm@50280
   187
wenzelm@50280
   188
fun report_status () = (*requires SYNCHRONIZED*)
wenzelm@50280
   189
  if ! ML_statistics then
wenzelm@50280
   190
    let
wenzelm@50280
   191
      val {ready, pending, running, passive} = Task_Queue.status (! queue);
wenzelm@50280
   192
      val total = length (! workers);
wenzelm@50280
   193
      val active = count_workers Working;
wenzelm@50280
   194
      val waiting = count_workers Waiting;
wenzelm@50280
   195
      val stats =
wenzelm@50280
   196
       [("now", signed_string_of_real (Time.toReal (Time.now ()))),
wenzelm@50974
   197
        ("tasks_proof", Markup.print_int (! forked_proofs)),
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@50975
   254
    val _ =
wenzelm@50975
   255
      if ! Multithreading.trace >= 2 then
wenzelm@50975
   256
        Output.protocol_message (Markup.task_statistics :: Task_Queue.task_statistics task) ""
wenzelm@50975
   257
          handle Fail msg => warning msg
wenzelm@50975
   258
      else ();
wenzelm@32246
   259
    val _ = SYNCHRONIZED "finish" (fn () =>
wenzelm@32219
   260
      let
wenzelm@32738
   261
        val maximal = Unsynchronized.change_result queue (Task_Queue.finish task);
wenzelm@44295
   262
        val test = Exn.capture Multithreading.interrupted ();
wenzelm@32219
   263
        val _ =
wenzelm@44295
   264
          if ok andalso not (Exn.is_interrupt_exn test) then ()
wenzelm@44299
   265
          else if null (cancel_now group) then ()
wenzelm@34279
   266
          else cancel_later group;
wenzelm@32219
   267
        val _ = broadcast work_finished;
wenzelm@33413
   268
        val _ = if maximal then () else signal work_available;
wenzelm@32219
   269
      in () end);
wenzelm@28167
   270
  in () end;
wenzelm@28167
   271
wenzelm@33410
   272
fun worker_wait active cond = (*requires SYNCHRONIZED*)
wenzelm@33406
   273
  let
wenzelm@33410
   274
    val state =
wenzelm@33410
   275
      (case AList.lookup Thread.equal (! workers) (Thread.self ()) of
wenzelm@33410
   276
        SOME state => state
wenzelm@33410
   277
      | NONE => raise Fail "Unregistered worker thread");
wenzelm@33410
   278
    val _ = state := (if active then Waiting else Sleeping);
wenzelm@33406
   279
    val _ = wait cond;
wenzelm@33410
   280
    val _ = state := Working;
wenzelm@33406
   281
  in () end;
wenzelm@28162
   282
wenzelm@33415
   283
fun worker_next () = (*requires SYNCHRONIZED*)
wenzelm@33406
   284
  if length (! workers) > ! max_workers then
wenzelm@33406
   285
    (Unsynchronized.change workers (AList.delete Thread.equal (Thread.self ()));
wenzelm@33415
   286
     signal work_available;
wenzelm@28167
   287
     NONE)
wenzelm@33410
   288
  else if count_workers Working > ! max_active then
wenzelm@33415
   289
    (worker_wait false work_available; worker_next ())
wenzelm@28166
   290
  else
wenzelm@32738
   291
    (case Unsynchronized.change_result queue (Task_Queue.dequeue (Thread.self ())) of
wenzelm@33415
   292
      NONE => (worker_wait false work_available; worker_next ())
wenzelm@33413
   293
    | some => (signal work_available; some));
wenzelm@28156
   294
wenzelm@28167
   295
fun worker_loop name =
wenzelm@33415
   296
  (case SYNCHRONIZED name (fn () => worker_next ()) of
wenzelm@29119
   297
    NONE => ()
wenzelm@44295
   298
  | SOME work => (worker_exec work; worker_loop name));
wenzelm@28156
   299
wenzelm@33407
   300
fun worker_start name = (*requires SYNCHRONIZED*)
wenzelm@37216
   301
  Unsynchronized.change workers (cons (Simple_Thread.fork false (fn () => worker_loop name),
wenzelm@33410
   302
    Unsynchronized.ref Working));
wenzelm@28156
   303
wenzelm@28156
   304
wenzelm@28156
   305
(* scheduler *)
wenzelm@28156
   306
wenzelm@28206
   307
fun scheduler_next () = (*requires SYNCHRONIZED*)
wenzelm@28156
   308
  let
wenzelm@33407
   309
    val now = Time.now ();
wenzelm@33407
   310
    val tick = Time.<= (Time.+ (! last_round, next_round), now);
wenzelm@33407
   311
    val _ = if tick then last_round := now else ();
wenzelm@33407
   312
wenzelm@33415
   313
wenzelm@50280
   314
    (* runtime status *)
wenzelm@33415
   315
wenzelm@32226
   316
    val _ =
wenzelm@51046
   317
      if tick then Unsynchronized.change status_ticks (fn i => i + 1) else ();
wenzelm@33407
   318
    val _ =
wenzelm@51046
   319
      if tick andalso ! status_ticks mod (if ! Multithreading.trace >= 1 then 2 else 10) = 0
wenzelm@51046
   320
      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@51279
   330
            "SCHEDULER: 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@51281
   379
        signal work_available);
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@51279
   397
     (Multithreading.tracing 1 (fn () => "SCHEDULER: Interrupt");
wenzelm@44341
   398
      List.app cancel_later (cancel_all ());
wenzelm@51281
   399
      signal work_available; 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@50914
   436
(* results *)
wenzelm@50914
   437
wenzelm@50914
   438
fun error_msg pos ((serial, msg), exec_id) =
wenzelm@50916
   439
  Position.setmp_thread_data pos (fn () =>
wenzelm@50931
   440
    let val id = Position.get_id pos in
wenzelm@50931
   441
      if is_none id orelse is_none exec_id orelse id = exec_id
wenzelm@50931
   442
      then Output.error_msg' (serial, msg) else ()
wenzelm@50931
   443
    end) ();
wenzelm@44110
   444
wenzelm@50914
   445
fun identify_result pos res =
wenzelm@50914
   446
  (case res of
wenzelm@50914
   447
    Exn.Exn exn =>
wenzelm@50914
   448
      let val exec_id =
wenzelm@50914
   449
        (case Position.get_id pos of
wenzelm@50914
   450
          NONE => []
wenzelm@50914
   451
        | SOME id => [(Markup.exec_idN, id)])
wenzelm@50914
   452
      in Exn.Exn (Par_Exn.identify exec_id exn) end
wenzelm@50914
   453
  | _ => res);
wenzelm@50914
   454
wenzelm@50914
   455
fun assign_result group result res =
wenzelm@44110
   456
  let
wenzelm@44110
   457
    val _ = Single_Assignment.assign result res
wenzelm@44110
   458
      handle exn as Fail _ =>
wenzelm@44110
   459
        (case Single_Assignment.peek result of
wenzelm@44110
   460
          SOME (Exn.Exn e) => reraise (if Exn.is_interrupt e then e else exn)
wenzelm@44110
   461
        | _ => reraise exn);
wenzelm@44110
   462
    val ok =
wenzelm@44110
   463
      (case the (Single_Assignment.peek result) of
wenzelm@44111
   464
        Exn.Exn exn =>
wenzelm@44111
   465
          (SYNCHRONIZED "cancel" (fn () => Task_Queue.cancel_group group exn); false)
wenzelm@44110
   466
      | Exn.Res _ => true);
wenzelm@44110
   467
  in ok end;
wenzelm@44110
   468
wenzelm@50914
   469
wenzelm@50914
   470
(* future jobs *)
wenzelm@50914
   471
wenzelm@44113
   472
fun future_job group interrupts (e: unit -> 'a) =
wenzelm@44110
   473
  let
wenzelm@44110
   474
    val result = Single_Assignment.var "future" : 'a result;
wenzelm@44110
   475
    val pos = Position.thread_data ();
wenzelm@44110
   476
    fun job ok =
wenzelm@44110
   477
      let
wenzelm@44110
   478
        val res =
wenzelm@44110
   479
          if ok then
wenzelm@44110
   480
            Exn.capture (fn () =>
wenzelm@44113
   481
              Multithreading.with_attributes
wenzelm@44113
   482
                (if interrupts
wenzelm@44113
   483
                 then Multithreading.private_interrupts else Multithreading.no_interrupts)
wenzelm@44295
   484
                (fn _ => Position.setmp_thread_data pos e ())) ()
wenzelm@44110
   485
          else Exn.interrupt_exn;
wenzelm@50914
   486
      in assign_result group result (identify_result pos res) end;
wenzelm@44110
   487
  in (result, job) end;
wenzelm@44110
   488
wenzelm@44110
   489
wenzelm@29366
   490
(* fork *)
wenzelm@29366
   491
wenzelm@44427
   492
type params = {name: string, group: group option, deps: task list, pri: int, interrupts: bool};
wenzelm@44427
   493
val default_params: params = {name = "", group = NONE, deps = [], pri = 0, interrupts = true};
wenzelm@44113
   494
wenzelm@44427
   495
fun forks ({name, group, deps, pri, interrupts}: params) es =
wenzelm@41674
   496
  if null es then []
wenzelm@41674
   497
  else
wenzelm@41674
   498
    let
wenzelm@41674
   499
      val grp =
wenzelm@41674
   500
        (case group of
wenzelm@41674
   501
          NONE => worker_subgroup ()
wenzelm@41674
   502
        | SOME grp => grp);
wenzelm@41708
   503
      fun enqueue e queue =
wenzelm@41674
   504
        let
wenzelm@44113
   505
          val (result, job) = future_job grp interrupts e;
wenzelm@41708
   506
          val (task, queue') = Task_Queue.enqueue name grp deps pri job queue;
wenzelm@41683
   507
          val future = Future {promised = false, task = task, result = result};
wenzelm@41708
   508
        in (future, queue') end;
wenzelm@41674
   509
    in
wenzelm@41674
   510
      SYNCHRONIZED "enqueue" (fn () =>
wenzelm@41674
   511
        let
wenzelm@41708
   512
          val (futures, queue') = fold_map enqueue es (! queue);
wenzelm@41708
   513
          val _ = queue := queue';
wenzelm@41708
   514
          val minimal = forall (not o Task_Queue.known_task queue') deps;
wenzelm@41674
   515
          val _ = if minimal then signal work_available else ();
wenzelm@41674
   516
          val _ = scheduler_check ();
wenzelm@41674
   517
        in futures end)
wenzelm@41674
   518
    end;
wenzelm@28162
   519
wenzelm@50983
   520
fun fork e =
wenzelm@50983
   521
  (singleton o forks) {name = "fork", group = NONE, deps = [], pri = 0, interrupts = true} e;
wenzelm@28186
   522
wenzelm@28186
   523
wenzelm@29366
   524
(* join *)
wenzelm@29366
   525
wenzelm@32099
   526
fun get_result x =
wenzelm@32099
   527
  (case peek x of
wenzelm@37852
   528
    NONE => Exn.Exn (Fail "Unfinished future")
wenzelm@39232
   529
  | SOME res =>
wenzelm@39232
   530
      if Exn.is_interrupt_exn res then
wenzelm@44247
   531
        (case Task_Queue.group_status (Task_Queue.group_of_task (task_of x)) of
wenzelm@44247
   532
          NONE => res
wenzelm@44247
   533
        | SOME exn => Exn.Exn exn)
wenzelm@39232
   534
      else res);
wenzelm@28186
   535
wenzelm@49935
   536
local
wenzelm@49935
   537
wenzelm@32095
   538
fun join_next deps = (*requires SYNCHRONIZED*)
wenzelm@41695
   539
  if null deps then NONE
wenzelm@32224
   540
  else
wenzelm@41681
   541
    (case Unsynchronized.change_result queue (Task_Queue.dequeue_deps (Thread.self ()) deps) of
wenzelm@41695
   542
      (NONE, []) => NONE
wenzelm@41695
   543
    | (NONE, deps') =>
wenzelm@41695
   544
        (worker_waiting deps' (fn () => worker_wait true work_finished); join_next deps')
wenzelm@32224
   545
    | (SOME work, deps') => SOME (work, deps'));
wenzelm@32095
   546
wenzelm@32814
   547
fun execute_work NONE = ()
wenzelm@44110
   548
  | execute_work (SOME (work, deps')) =
wenzelm@44110
   549
      (worker_joining (fn () => worker_exec work); join_work deps')
wenzelm@32814
   550
and join_work deps =
wenzelm@43538
   551
  Multithreading.with_attributes Multithreading.no_interrupts
wenzelm@43538
   552
    (fn _ => execute_work (SYNCHRONIZED "join" (fn () => join_next deps)));
wenzelm@32814
   553
wenzelm@29551
   554
in
wenzelm@29551
   555
wenzelm@29366
   556
fun join_results xs =
wenzelm@41679
   557
  let
wenzelm@41679
   558
    val _ =
wenzelm@41679
   559
      if forall is_finished xs then ()
wenzelm@41679
   560
      else if Multithreading.self_critical () then
wenzelm@41679
   561
        error "Cannot join future values within critical section"
wenzelm@41695
   562
      else if is_some (worker_task ()) then join_work (map task_of xs)
wenzelm@41681
   563
      else List.app (ignore o Single_Assignment.await o result_of) xs;
wenzelm@41679
   564
  in map get_result xs end;
wenzelm@28186
   565
wenzelm@29551
   566
end;
wenzelm@29551
   567
wenzelm@28647
   568
fun join_result x = singleton join_results x;
wenzelm@44330
   569
fun joins xs = Par_Exn.release_all (join_results xs);
wenzelm@28647
   570
fun join x = Exn.release (join_result x);
wenzelm@28156
   571
wenzelm@29366
   572
wenzelm@51325
   573
(* fast-path operations -- bypass task queue if possible *)
wenzelm@34277
   574
wenzelm@44294
   575
fun value_result (res: 'a Exn.result) =
wenzelm@34277
   576
  let
wenzelm@45136
   577
    val task = Task_Queue.dummy_task;
wenzelm@41683
   578
    val group = Task_Queue.group_of_task task;
wenzelm@35016
   579
    val result = Single_Assignment.var "value" : 'a result;
wenzelm@50914
   580
    val _ = assign_result group result (identify_result (Position.thread_data ()) res);
wenzelm@41683
   581
  in Future {promised = false, task = task, result = result} end;
wenzelm@29366
   582
wenzelm@44294
   583
fun value x = value_result (Exn.Res x);
wenzelm@44294
   584
wenzelm@44330
   585
fun cond_forks args es =
wenzelm@44330
   586
  if Multithreading.enabled () then forks args es
wenzelm@44330
   587
  else map (fn e => value_result (Exn.interruptible_capture e ())) es;
wenzelm@44330
   588
wenzelm@29384
   589
fun map_future f x =
wenzelm@51325
   590
  if is_finished x then value_result (Exn.interruptible_capture (f o join) x)
wenzelm@49935
   591
  else
wenzelm@49935
   592
    let
wenzelm@49935
   593
      val task = task_of x;
wenzelm@49935
   594
      val group = Task_Queue.group_of_task task;
wenzelm@49935
   595
      val (result, job) = future_job group true (fn () => f (join x));
wenzelm@29384
   596
wenzelm@49935
   597
      val extended = SYNCHRONIZED "extend" (fn () =>
wenzelm@49935
   598
        (case Task_Queue.extend task job (! queue) of
wenzelm@49935
   599
          SOME queue' => (queue := queue'; true)
wenzelm@49935
   600
        | NONE => false));
wenzelm@49935
   601
    in
wenzelm@49935
   602
      if extended then Future {promised = false, task = task, result = result}
wenzelm@49935
   603
      else
wenzelm@49935
   604
        (singleton o cond_forks)
wenzelm@49935
   605
          {name = "map_future", group = SOME group, deps = [task],
wenzelm@49935
   606
            pri = Task_Queue.pri_of_task task, interrupts = true}
wenzelm@49935
   607
          (fn () => f (join x))
wenzelm@49935
   608
    end;
wenzelm@28979
   609
wenzelm@28191
   610
wenzelm@34277
   611
(* promised futures -- fulfilled by external means *)
wenzelm@34277
   612
wenzelm@44298
   613
fun promise_group group abort : 'a future =
wenzelm@34277
   614
  let
wenzelm@35016
   615
    val result = Single_Assignment.var "promise" : 'a result;
wenzelm@44298
   616
    fun assign () = assign_result group result Exn.interrupt_exn
wenzelm@39243
   617
      handle Fail _ => true
wenzelm@39243
   618
        | exn =>
wenzelm@44298
   619
            if Exn.is_interrupt exn
wenzelm@44298
   620
            then raise Fail "Concurrent attempt to fulfill promise"
wenzelm@39243
   621
            else reraise exn;
wenzelm@44298
   622
    fun job () =
wenzelm@44298
   623
      Multithreading.with_attributes Multithreading.no_interrupts
wenzelm@47423
   624
        (fn _ => Exn.release (Exn.capture assign () before abort ()));
wenzelm@37854
   625
    val task = SYNCHRONIZED "enqueue_passive" (fn () =>
wenzelm@44298
   626
      Unsynchronized.change_result queue (Task_Queue.enqueue_passive group job));
wenzelm@41683
   627
  in Future {promised = true, task = task, result = result} end;
wenzelm@34277
   628
wenzelm@44298
   629
fun promise abort = promise_group (worker_subgroup ()) abort;
wenzelm@34277
   630
wenzelm@41683
   631
fun fulfill_result (Future {promised, task, result}) res =
wenzelm@39243
   632
  if not promised then raise Fail "Not a promised future"
wenzelm@39243
   633
  else
wenzelm@39243
   634
    let
wenzelm@41683
   635
      val group = Task_Queue.group_of_task task;
wenzelm@50914
   636
      val pos = Position.thread_data ();
wenzelm@50914
   637
      fun job ok =
wenzelm@50914
   638
        assign_result group result (if ok then identify_result pos res else Exn.interrupt_exn);
wenzelm@39243
   639
      val _ =
wenzelm@39243
   640
        Multithreading.with_attributes Multithreading.no_interrupts (fn _ =>
wenzelm@39243
   641
          let
wenzelm@47423
   642
            val passive_job =
wenzelm@39243
   643
              SYNCHRONIZED "fulfill_result" (fn () =>
wenzelm@39243
   644
                Unsynchronized.change_result queue
wenzelm@39243
   645
                  (Task_Queue.dequeue_passive (Thread.self ()) task));
wenzelm@47423
   646
          in
wenzelm@47423
   647
            (case passive_job of
wenzelm@47423
   648
              SOME true => worker_exec (task, [job])
wenzelm@47423
   649
            | SOME false => ()
wenzelm@47423
   650
            | NONE => ignore (job (not (Task_Queue.is_canceled group))))
wenzelm@47423
   651
          end);
wenzelm@41681
   652
      val _ =
wenzelm@41695
   653
        if is_some (Single_Assignment.peek result) then ()
wenzelm@41695
   654
        else worker_waiting [task] (fn () => ignore (Single_Assignment.await result));
wenzelm@39243
   655
    in () end;
wenzelm@34277
   656
wenzelm@43761
   657
fun fulfill x res = fulfill_result x (Exn.Res res);
wenzelm@34277
   658
wenzelm@34277
   659
wenzelm@49906
   660
(* terminate *)
wenzelm@49906
   661
wenzelm@49906
   662
fun terminate group =
wenzelm@49906
   663
  let
wenzelm@49906
   664
    val tasks =
wenzelm@49906
   665
      SYNCHRONIZED "terminate" (fn () =>
wenzelm@49906
   666
        let val _ = cancel_group_unsynchronized group;
wenzelm@49906
   667
        in Task_Queue.group_tasks (! queue) group end);
wenzelm@49906
   668
  in
wenzelm@49906
   669
    if null tasks then ()
wenzelm@49906
   670
    else
wenzelm@49906
   671
      (singleton o forks)
wenzelm@49906
   672
        {name = "terminate", group = SOME (new_group NONE),
wenzelm@49906
   673
          deps = tasks, pri = 0, interrupts = false} I
wenzelm@49906
   674
      |> join
wenzelm@49906
   675
  end;
wenzelm@49906
   676
wenzelm@49906
   677
wenzelm@32228
   678
(* shutdown *)
wenzelm@29366
   679
wenzelm@28203
   680
fun shutdown () =
wenzelm@51283
   681
  if not Multithreading.available then ()
wenzelm@51283
   682
  else if is_some (worker_task ()) then
wenzelm@51283
   683
    raise Fail "Cannot shutdown while running as worker thread"
wenzelm@51283
   684
  else
wenzelm@28276
   685
    SYNCHRONIZED "shutdown" (fn () =>
wenzelm@51279
   686
      while scheduler_active () do
wenzelm@51279
   687
       (Multithreading.tracing 1 (fn () => "SHUTDOWN: wait");
wenzelm@51283
   688
        wait scheduler_event));
wenzelm@28203
   689
wenzelm@29366
   690
wenzelm@29366
   691
(*final declarations of this structure!*)
wenzelm@29366
   692
val map = map_future;
wenzelm@29366
   693
wenzelm@28156
   694
end;
wenzelm@28972
   695
wenzelm@28972
   696
type 'a future = 'a Future.future;
wenzelm@28972
   697