src/Pure/Concurrent/future.ML
author wenzelm
Wed Apr 11 12:15:56 2012 +0200 (2012-04-11)
changeset 47421 9624408d8827
parent 47404 e6e5750f1311
child 47423 8a179a0493e3
permissions -rw-r--r--
always signal after cancel_group: passive tasks may have become active;
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@44268
    29
    Failure of one group member causes the immediate peers to be
wenzelm@44268
    30
    interrupted eventually (i.e. none by default).  Interrupted tasks
wenzelm@44268
    31
    that lack regular result information, will pick up parallel
wenzelm@44268
    32
    exceptions from the 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@44301
    54
  val interruptible_task: ('a -> 'b) -> 'a -> 'b
wenzelm@47404
    55
  val cancel_group: group -> unit
wenzelm@47404
    56
  val cancel: 'a future -> unit
wenzelm@44427
    57
  type params = {name: string, group: group option, deps: task list, pri: int, interrupts: bool}
wenzelm@44427
    58
  val default_params: params
wenzelm@44427
    59
  val forks: params -> (unit -> 'a) list -> 'a future list
wenzelm@29119
    60
  val fork_pri: int -> (unit -> 'a) -> 'a future
wenzelm@32724
    61
  val fork: (unit -> 'a) -> 'a future
wenzelm@28972
    62
  val join_results: 'a future list -> 'a Exn.result list
wenzelm@28972
    63
  val join_result: 'a future -> 'a Exn.result
wenzelm@44330
    64
  val joins: 'a future list -> 'a list
wenzelm@28972
    65
  val join: 'a future -> 'a
wenzelm@44301
    66
  val join_tasks: task list -> unit
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@28203
    75
  val shutdown: unit -> unit
wenzelm@38236
    76
  val status: (unit -> 'a) -> 'a
wenzelm@47404
    77
  val group_tasks: group -> task list
wenzelm@47404
    78
  val queue_status: unit -> {ready: int, pending: int, running: int, passive: int}
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@32248
   157
fun broadcast_work () = (*requires SYNCHRONIZED*)
wenzelm@32248
   158
 (ConditionVar.broadcast work_available;
wenzelm@32225
   159
  ConditionVar.broadcast work_finished);
wenzelm@32225
   160
wenzelm@28156
   161
end;
wenzelm@28156
   162
wenzelm@28156
   163
wenzelm@33410
   164
(* global state *)
wenzelm@33410
   165
wenzelm@33410
   166
val queue = Unsynchronized.ref Task_Queue.empty;
wenzelm@33410
   167
val next = Unsynchronized.ref 0;
wenzelm@33410
   168
val scheduler = Unsynchronized.ref (NONE: Thread.thread option);
wenzelm@44300
   169
val canceled = Unsynchronized.ref ([]: group list);
wenzelm@33410
   170
val do_shutdown = Unsynchronized.ref false;
wenzelm@33410
   171
val max_workers = Unsynchronized.ref 0;
wenzelm@33410
   172
val max_active = Unsynchronized.ref 0;
wenzelm@33411
   173
val worker_trend = Unsynchronized.ref 0;
wenzelm@33410
   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@44110
   182
(* cancellation primitives *)
wenzelm@32099
   183
wenzelm@34279
   184
fun cancel_now group = (*requires SYNCHRONIZED*)
wenzelm@44341
   185
  let
wenzelm@47404
   186
    val running = Task_Queue.cancel (! queue) group;
wenzelm@47404
   187
    val _ = List.app Simple_Thread.interrupt_unsynchronized running;
wenzelm@47404
   188
  in running end;
wenzelm@44341
   189
wenzelm@44341
   190
fun cancel_all () = (*requires SYNCHRONIZED*)
wenzelm@44341
   191
  let
wenzelm@44341
   192
    val (groups, threads) = Task_Queue.cancel_all (! queue);
wenzelm@44341
   193
    val _ = List.app Simple_Thread.interrupt_unsynchronized threads;
wenzelm@44341
   194
  in groups end;
wenzelm@34279
   195
wenzelm@34279
   196
fun cancel_later group = (*requires SYNCHRONIZED*)
wenzelm@32738
   197
 (Unsynchronized.change canceled (insert Task_Queue.eq_group group);
wenzelm@32738
   198
  broadcast scheduler_event);
wenzelm@29341
   199
wenzelm@44301
   200
fun interruptible_task f x =
wenzelm@44301
   201
  (if Multithreading.available then
wenzelm@44301
   202
    Multithreading.with_attributes
wenzelm@44301
   203
      (if is_some (worker_task ())
wenzelm@44301
   204
       then Multithreading.private_interrupts
wenzelm@44301
   205
       else Multithreading.public_interrupts)
wenzelm@44301
   206
      (fn _ => f x)
wenzelm@44301
   207
   else interruptible f x)
wenzelm@44301
   208
  before Multithreading.interrupted ();
wenzelm@44301
   209
wenzelm@44301
   210
wenzelm@44110
   211
(* worker threads *)
wenzelm@44110
   212
wenzelm@44110
   213
fun worker_exec (task, jobs) =
wenzelm@28167
   214
  let
wenzelm@41683
   215
    val group = Task_Queue.group_of_task task;
wenzelm@32102
   216
    val valid = not (Task_Queue.is_canceled group);
wenzelm@41670
   217
    val ok =
wenzelm@41670
   218
      Task_Queue.running task (fn () =>
wenzelm@41683
   219
        setmp_worker_task task (fn () =>
wenzelm@41670
   220
          fold (fn job => fn ok => job valid andalso ok) jobs true) ());
wenzelm@41776
   221
    val _ = Multithreading.tracing 2 (fn () =>
wenzelm@41670
   222
      let
wenzelm@43951
   223
        val s = Task_Queue.str_of_task_groups task;
wenzelm@41670
   224
        fun micros time = string_of_int (Time.toNanoseconds time div 1000);
wenzelm@41680
   225
        val (run, wait, deps) = Task_Queue.timing_of_task task;
wenzelm@41680
   226
      in "TASK " ^ s ^ " " ^ micros run ^ " " ^ micros wait ^ " (" ^ commas deps ^ ")" end);
wenzelm@32246
   227
    val _ = SYNCHRONIZED "finish" (fn () =>
wenzelm@32219
   228
      let
wenzelm@32738
   229
        val maximal = Unsynchronized.change_result queue (Task_Queue.finish task);
wenzelm@44295
   230
        val test = Exn.capture Multithreading.interrupted ();
wenzelm@32219
   231
        val _ =
wenzelm@44295
   232
          if ok andalso not (Exn.is_interrupt_exn test) then ()
wenzelm@44299
   233
          else if null (cancel_now group) then ()
wenzelm@34279
   234
          else cancel_later group;
wenzelm@32219
   235
        val _ = broadcast work_finished;
wenzelm@33413
   236
        val _ = if maximal then () else signal work_available;
wenzelm@32219
   237
      in () end);
wenzelm@28167
   238
  in () end;
wenzelm@28167
   239
wenzelm@33410
   240
fun worker_wait active cond = (*requires SYNCHRONIZED*)
wenzelm@33406
   241
  let
wenzelm@33410
   242
    val state =
wenzelm@33410
   243
      (case AList.lookup Thread.equal (! workers) (Thread.self ()) of
wenzelm@33410
   244
        SOME state => state
wenzelm@33410
   245
      | NONE => raise Fail "Unregistered worker thread");
wenzelm@33410
   246
    val _ = state := (if active then Waiting else Sleeping);
wenzelm@33406
   247
    val _ = wait cond;
wenzelm@33410
   248
    val _ = state := Working;
wenzelm@33406
   249
  in () end;
wenzelm@28162
   250
wenzelm@33415
   251
fun worker_next () = (*requires SYNCHRONIZED*)
wenzelm@33406
   252
  if length (! workers) > ! max_workers then
wenzelm@33406
   253
    (Unsynchronized.change workers (AList.delete Thread.equal (Thread.self ()));
wenzelm@33415
   254
     signal work_available;
wenzelm@28167
   255
     NONE)
wenzelm@33410
   256
  else if count_workers Working > ! max_active then
wenzelm@33415
   257
    (worker_wait false work_available; worker_next ())
wenzelm@28166
   258
  else
wenzelm@32738
   259
    (case Unsynchronized.change_result queue (Task_Queue.dequeue (Thread.self ())) of
wenzelm@33415
   260
      NONE => (worker_wait false work_available; worker_next ())
wenzelm@33413
   261
    | some => (signal work_available; some));
wenzelm@28156
   262
wenzelm@28167
   263
fun worker_loop name =
wenzelm@33415
   264
  (case SYNCHRONIZED name (fn () => worker_next ()) of
wenzelm@29119
   265
    NONE => ()
wenzelm@44295
   266
  | SOME work => (worker_exec work; worker_loop name));
wenzelm@28156
   267
wenzelm@33407
   268
fun worker_start name = (*requires SYNCHRONIZED*)
wenzelm@37216
   269
  Unsynchronized.change workers (cons (Simple_Thread.fork false (fn () => worker_loop name),
wenzelm@33410
   270
    Unsynchronized.ref Working));
wenzelm@28156
   271
wenzelm@28156
   272
wenzelm@28156
   273
(* scheduler *)
wenzelm@28156
   274
wenzelm@33407
   275
val status_ticks = Unsynchronized.ref 0;
wenzelm@33407
   276
wenzelm@33407
   277
val last_round = Unsynchronized.ref Time.zeroTime;
wenzelm@40301
   278
val next_round = seconds 0.05;
wenzelm@32226
   279
wenzelm@28206
   280
fun scheduler_next () = (*requires SYNCHRONIZED*)
wenzelm@28156
   281
  let
wenzelm@33407
   282
    val now = Time.now ();
wenzelm@33407
   283
    val tick = Time.<= (Time.+ (! last_round, next_round), now);
wenzelm@33407
   284
    val _ = if tick then last_round := now else ();
wenzelm@33407
   285
wenzelm@33415
   286
wenzelm@33415
   287
    (* queue and worker status *)
wenzelm@33415
   288
wenzelm@32226
   289
    val _ =
wenzelm@33407
   290
      if tick then Unsynchronized.change status_ticks (fn i => (i + 1) mod 10) else ();
wenzelm@33407
   291
    val _ =
wenzelm@33407
   292
      if tick andalso ! status_ticks = 0 then
wenzelm@33407
   293
        Multithreading.tracing 1 (fn () =>
wenzelm@33407
   294
          let
wenzelm@34277
   295
            val {ready, pending, running, passive} = Task_Queue.status (! queue);
wenzelm@33407
   296
            val total = length (! workers);
wenzelm@33410
   297
            val active = count_workers Working;
wenzelm@33410
   298
            val waiting = count_workers Waiting;
wenzelm@33407
   299
          in
wenzelm@33407
   300
            "SCHEDULE " ^ Time.toString now ^ ": " ^
wenzelm@33407
   301
              string_of_int ready ^ " ready, " ^
wenzelm@33407
   302
              string_of_int pending ^ " pending, " ^
wenzelm@34277
   303
              string_of_int running ^ " running, " ^
wenzelm@34277
   304
              string_of_int passive ^ " passive; " ^
wenzelm@33407
   305
              string_of_int total ^ " workers, " ^
wenzelm@33410
   306
              string_of_int active ^ " active, " ^
wenzelm@33410
   307
              string_of_int waiting ^ " waiting "
wenzelm@33407
   308
          end)
wenzelm@33407
   309
      else ();
wenzelm@32053
   310
wenzelm@28191
   311
    val _ =
wenzelm@32219
   312
      if forall (Thread.isActive o #1) (! workers) then ()
wenzelm@32095
   313
      else
wenzelm@33409
   314
        let
wenzelm@37682
   315
          val (alive, dead) = List.partition (Thread.isActive o #1) (! workers);
wenzelm@33409
   316
          val _ = workers := alive;
wenzelm@33409
   317
        in
wenzelm@33409
   318
          Multithreading.tracing 0 (fn () =>
wenzelm@33409
   319
            "SCHEDULE: disposed " ^ string_of_int (length dead) ^ " dead worker threads")
wenzelm@33409
   320
        end;
wenzelm@28191
   321
wenzelm@33415
   322
wenzelm@33415
   323
    (* worker pool adjustments *)
wenzelm@33415
   324
wenzelm@33415
   325
    val max_active0 = ! max_active;
wenzelm@33415
   326
    val max_workers0 = ! max_workers;
wenzelm@33415
   327
wenzelm@28206
   328
    val m = if ! do_shutdown then 0 else Multithreading.max_threads_value ();
wenzelm@33406
   329
    val _ = max_active := m;
wenzelm@33406
   330
wenzelm@33411
   331
    val mm =
wenzelm@33411
   332
      if ! do_shutdown then 0
wenzelm@33411
   333
      else if m = 9999 then 1
wenzelm@33413
   334
      else Int.min (Int.max (count_workers Working + 2 * count_workers Waiting, m), 4 * m);
wenzelm@33411
   335
    val _ =
wenzelm@33411
   336
      if tick andalso mm > ! max_workers then
wenzelm@33411
   337
        Unsynchronized.change worker_trend (fn w => if w < 0 then 0 else w + 1)
wenzelm@33411
   338
      else if tick andalso mm < ! max_workers then
wenzelm@33411
   339
        Unsynchronized.change worker_trend (fn w => if w > 0 then 0 else w - 1)
wenzelm@33411
   340
      else ();
wenzelm@33411
   341
    val _ =
wenzelm@33415
   342
      if mm = 0 orelse ! worker_trend > 50 orelse ! worker_trend < ~50 then
wenzelm@33415
   343
        max_workers := mm
wenzelm@44173
   344
      else if ! worker_trend > 5 andalso ! max_workers < 2 * m orelse ! max_workers = 0 then
wenzelm@33415
   345
        max_workers := Int.min (mm, 2 * m)
wenzelm@33411
   346
      else ();
wenzelm@33406
   347
wenzelm@33407
   348
    val missing = ! max_workers - length (! workers);
wenzelm@28203
   349
    val _ =
wenzelm@33407
   350
      if missing > 0 then
wenzelm@33415
   351
        funpow missing (fn () =>
wenzelm@33415
   352
          ignore (worker_start ("worker " ^ string_of_int (Unsynchronized.inc next)))) ()
wenzelm@28203
   353
      else ();
wenzelm@28206
   354
wenzelm@33415
   355
    val _ =
wenzelm@33415
   356
      if ! max_active = max_active0 andalso ! max_workers = max_workers0 then ()
wenzelm@33415
   357
      else signal work_available;
wenzelm@33415
   358
wenzelm@33415
   359
wenzelm@33415
   360
    (* canceled groups *)
wenzelm@33415
   361
wenzelm@32225
   362
    val _ =
wenzelm@32225
   363
      if null (! canceled) then ()
wenzelm@32293
   364
      else
wenzelm@32293
   365
       (Multithreading.tracing 1 (fn () =>
wenzelm@32293
   366
          string_of_int (length (! canceled)) ^ " canceled groups");
wenzelm@44299
   367
        Unsynchronized.change canceled (filter_out (null o cancel_now));
wenzelm@32293
   368
        broadcast_work ());
wenzelm@28206
   369
wenzelm@33415
   370
wenzelm@33415
   371
    (* delay loop *)
wenzelm@33415
   372
wenzelm@32295
   373
    val _ = Exn.release (wait_timeout next_round scheduler_event);
wenzelm@28167
   374
wenzelm@33415
   375
wenzelm@33415
   376
    (* shutdown *)
wenzelm@33415
   377
wenzelm@34277
   378
    val _ = if Task_Queue.all_passive (! queue) then do_shutdown := true else ();
wenzelm@32219
   379
    val continue = not (! do_shutdown andalso null (! workers));
wenzelm@32219
   380
    val _ = if continue then () else scheduler := NONE;
wenzelm@33415
   381
wenzelm@32219
   382
    val _ = broadcast scheduler_event;
wenzelm@32295
   383
  in continue end
wenzelm@39232
   384
  handle exn =>
wenzelm@39232
   385
    if Exn.is_interrupt exn then
wenzelm@39232
   386
     (Multithreading.tracing 1 (fn () => "Interrupt");
wenzelm@44341
   387
      List.app cancel_later (cancel_all ());
wenzelm@39232
   388
      broadcast_work (); true)
wenzelm@39232
   389
    else reraise exn;
wenzelm@32295
   390
wenzelm@28206
   391
fun scheduler_loop () =
wenzelm@44173
   392
 (while
wenzelm@33416
   393
    Multithreading.with_attributes
wenzelm@33416
   394
      (Multithreading.sync_interrupts Multithreading.public_interrupts)
wenzelm@33416
   395
      (fn _ => SYNCHRONIZED "scheduler" (fn () => scheduler_next ()))
wenzelm@44173
   396
  do (); last_round := Time.zeroTime);
wenzelm@28156
   397
wenzelm@28203
   398
fun scheduler_active () = (*requires SYNCHRONIZED*)
wenzelm@28203
   399
  (case ! scheduler of NONE => false | SOME thread => Thread.isActive thread);
wenzelm@28203
   400
wenzelm@32228
   401
fun scheduler_check () = (*requires SYNCHRONIZED*)
wenzelm@32228
   402
 (do_shutdown := false;
wenzelm@32248
   403
  if scheduler_active () then ()
wenzelm@37216
   404
  else scheduler := SOME (Simple_Thread.fork false scheduler_loop));
wenzelm@28156
   405
wenzelm@44301
   406
wenzelm@44301
   407
wenzelm@44301
   408
(** futures **)
wenzelm@44301
   409
wenzelm@44301
   410
(* cancel *)
wenzelm@44301
   411
wenzelm@44301
   412
fun cancel_group group = SYNCHRONIZED "cancel_group" (fn () =>
wenzelm@44299
   413
  let
wenzelm@47421
   414
    val _ = if null (cancel_now group) then () else cancel_later group;
wenzelm@47421
   415
    val _ = signal work_available;
wenzelm@47421
   416
    val _ = scheduler_check ();
wenzelm@47404
   417
  in () end);
wenzelm@44299
   418
wenzelm@44301
   419
fun cancel x = cancel_group (Task_Queue.group_of_task (task_of x));
wenzelm@29366
   420
wenzelm@28156
   421
wenzelm@44110
   422
(* future jobs *)
wenzelm@44110
   423
wenzelm@44249
   424
fun assign_result group result raw_res =
wenzelm@44110
   425
  let
wenzelm@44249
   426
    val res =
wenzelm@44249
   427
      (case raw_res of
wenzelm@44249
   428
        Exn.Exn exn => Exn.Exn (#2 (Par_Exn.serial exn))
wenzelm@44249
   429
      | _ => raw_res);
wenzelm@44110
   430
    val _ = Single_Assignment.assign result res
wenzelm@44110
   431
      handle exn as Fail _ =>
wenzelm@44110
   432
        (case Single_Assignment.peek result of
wenzelm@44110
   433
          SOME (Exn.Exn e) => reraise (if Exn.is_interrupt e then e else exn)
wenzelm@44110
   434
        | _ => reraise exn);
wenzelm@44110
   435
    val ok =
wenzelm@44110
   436
      (case the (Single_Assignment.peek result) of
wenzelm@44111
   437
        Exn.Exn exn =>
wenzelm@44111
   438
          (SYNCHRONIZED "cancel" (fn () => Task_Queue.cancel_group group exn); false)
wenzelm@44110
   439
      | Exn.Res _ => true);
wenzelm@44110
   440
  in ok end;
wenzelm@44110
   441
wenzelm@44113
   442
fun future_job group interrupts (e: unit -> 'a) =
wenzelm@44110
   443
  let
wenzelm@44110
   444
    val result = Single_Assignment.var "future" : 'a result;
wenzelm@44110
   445
    val pos = Position.thread_data ();
wenzelm@44110
   446
    fun job ok =
wenzelm@44110
   447
      let
wenzelm@44110
   448
        val res =
wenzelm@44110
   449
          if ok then
wenzelm@44110
   450
            Exn.capture (fn () =>
wenzelm@44113
   451
              Multithreading.with_attributes
wenzelm@44113
   452
                (if interrupts
wenzelm@44113
   453
                 then Multithreading.private_interrupts else Multithreading.no_interrupts)
wenzelm@44295
   454
                (fn _ => Position.setmp_thread_data pos e ())) ()
wenzelm@44110
   455
          else Exn.interrupt_exn;
wenzelm@44110
   456
      in assign_result group result res end;
wenzelm@44110
   457
  in (result, job) end;
wenzelm@44110
   458
wenzelm@44110
   459
wenzelm@29366
   460
(* fork *)
wenzelm@29366
   461
wenzelm@44427
   462
type params = {name: string, group: group option, deps: task list, pri: int, interrupts: bool};
wenzelm@44427
   463
val default_params: params = {name = "", group = NONE, deps = [], pri = 0, interrupts = true};
wenzelm@44113
   464
wenzelm@44427
   465
fun forks ({name, group, deps, pri, interrupts}: params) es =
wenzelm@41674
   466
  if null es then []
wenzelm@41674
   467
  else
wenzelm@41674
   468
    let
wenzelm@41674
   469
      val grp =
wenzelm@41674
   470
        (case group of
wenzelm@41674
   471
          NONE => worker_subgroup ()
wenzelm@41674
   472
        | SOME grp => grp);
wenzelm@41708
   473
      fun enqueue e queue =
wenzelm@41674
   474
        let
wenzelm@44113
   475
          val (result, job) = future_job grp interrupts e;
wenzelm@41708
   476
          val (task, queue') = Task_Queue.enqueue name grp deps pri job queue;
wenzelm@41683
   477
          val future = Future {promised = false, task = task, result = result};
wenzelm@41708
   478
        in (future, queue') end;
wenzelm@41674
   479
    in
wenzelm@41674
   480
      SYNCHRONIZED "enqueue" (fn () =>
wenzelm@41674
   481
        let
wenzelm@41708
   482
          val (futures, queue') = fold_map enqueue es (! queue);
wenzelm@41708
   483
          val _ = queue := queue';
wenzelm@41708
   484
          val minimal = forall (not o Task_Queue.known_task queue') deps;
wenzelm@41674
   485
          val _ = if minimal then signal work_available else ();
wenzelm@41674
   486
          val _ = scheduler_check ();
wenzelm@41674
   487
        in futures end)
wenzelm@41674
   488
    end;
wenzelm@28162
   489
wenzelm@44113
   490
fun fork_pri pri e =
wenzelm@44301
   491
  (singleton o forks) {name = "fork", group = NONE, deps = [], pri = pri, interrupts = true} e;
wenzelm@44113
   492
wenzelm@41672
   493
fun fork e = fork_pri 0 e;
wenzelm@28186
   494
wenzelm@28186
   495
wenzelm@29366
   496
(* join *)
wenzelm@29366
   497
wenzelm@29551
   498
local
wenzelm@29551
   499
wenzelm@32099
   500
fun get_result x =
wenzelm@32099
   501
  (case peek x of
wenzelm@37852
   502
    NONE => Exn.Exn (Fail "Unfinished future")
wenzelm@39232
   503
  | SOME res =>
wenzelm@39232
   504
      if Exn.is_interrupt_exn res then
wenzelm@44247
   505
        (case Task_Queue.group_status (Task_Queue.group_of_task (task_of x)) of
wenzelm@44247
   506
          NONE => res
wenzelm@44247
   507
        | SOME exn => Exn.Exn exn)
wenzelm@39232
   508
      else res);
wenzelm@28186
   509
wenzelm@32095
   510
fun join_next deps = (*requires SYNCHRONIZED*)
wenzelm@41695
   511
  if null deps then NONE
wenzelm@32224
   512
  else
wenzelm@41681
   513
    (case Unsynchronized.change_result queue (Task_Queue.dequeue_deps (Thread.self ()) deps) of
wenzelm@41695
   514
      (NONE, []) => NONE
wenzelm@41695
   515
    | (NONE, deps') =>
wenzelm@41695
   516
        (worker_waiting deps' (fn () => worker_wait true work_finished); join_next deps')
wenzelm@32224
   517
    | (SOME work, deps') => SOME (work, deps'));
wenzelm@32095
   518
wenzelm@32814
   519
fun execute_work NONE = ()
wenzelm@44110
   520
  | execute_work (SOME (work, deps')) =
wenzelm@44110
   521
      (worker_joining (fn () => worker_exec work); join_work deps')
wenzelm@32814
   522
and join_work deps =
wenzelm@43538
   523
  Multithreading.with_attributes Multithreading.no_interrupts
wenzelm@43538
   524
    (fn _ => execute_work (SYNCHRONIZED "join" (fn () => join_next deps)));
wenzelm@32814
   525
wenzelm@29551
   526
in
wenzelm@29551
   527
wenzelm@29366
   528
fun join_results xs =
wenzelm@41679
   529
  let
wenzelm@41679
   530
    val _ =
wenzelm@41679
   531
      if forall is_finished xs then ()
wenzelm@41679
   532
      else if Multithreading.self_critical () then
wenzelm@41679
   533
        error "Cannot join future values within critical section"
wenzelm@41695
   534
      else if is_some (worker_task ()) then join_work (map task_of xs)
wenzelm@41681
   535
      else List.app (ignore o Single_Assignment.await o result_of) xs;
wenzelm@41679
   536
  in map get_result xs end;
wenzelm@28186
   537
wenzelm@29551
   538
end;
wenzelm@29551
   539
wenzelm@28647
   540
fun join_result x = singleton join_results x;
wenzelm@44330
   541
fun joins xs = Par_Exn.release_all (join_results xs);
wenzelm@28647
   542
fun join x = Exn.release (join_result x);
wenzelm@28156
   543
wenzelm@44301
   544
fun join_tasks [] = ()
wenzelm@44301
   545
  | join_tasks tasks =
wenzelm@44301
   546
      (singleton o forks)
wenzelm@44301
   547
        {name = "join_tasks", group = SOME (new_group NONE),
wenzelm@44301
   548
          deps = tasks, pri = 0, interrupts = false} I
wenzelm@44301
   549
      |> join;
wenzelm@44301
   550
wenzelm@29366
   551
wenzelm@44110
   552
(* fast-path versions -- bypassing task queue *)
wenzelm@34277
   553
wenzelm@44294
   554
fun value_result (res: 'a Exn.result) =
wenzelm@34277
   555
  let
wenzelm@45136
   556
    val task = Task_Queue.dummy_task;
wenzelm@41683
   557
    val group = Task_Queue.group_of_task task;
wenzelm@35016
   558
    val result = Single_Assignment.var "value" : 'a result;
wenzelm@44294
   559
    val _ = assign_result group result res;
wenzelm@41683
   560
  in Future {promised = false, task = task, result = result} end;
wenzelm@29366
   561
wenzelm@44294
   562
fun value x = value_result (Exn.Res x);
wenzelm@44294
   563
wenzelm@44330
   564
fun cond_forks args es =
wenzelm@44330
   565
  if Multithreading.enabled () then forks args es
wenzelm@44330
   566
  else map (fn e => value_result (Exn.interruptible_capture e ())) es;
wenzelm@44330
   567
wenzelm@29384
   568
fun map_future f x =
wenzelm@29366
   569
  let
wenzelm@29384
   570
    val task = task_of x;
wenzelm@44300
   571
    val group = new_group (SOME (Task_Queue.group_of_task task));
wenzelm@44113
   572
    val (result, job) = future_job group true (fn () => f (join x));
wenzelm@29384
   573
wenzelm@32246
   574
    val extended = SYNCHRONIZED "extend" (fn () =>
wenzelm@29366
   575
      (case Task_Queue.extend task job (! queue) of
wenzelm@29366
   576
        SOME queue' => (queue := queue'; true)
wenzelm@29366
   577
      | NONE => false));
wenzelm@29366
   578
  in
wenzelm@41683
   579
    if extended then Future {promised = false, task = task, result = result}
wenzelm@41672
   580
    else
wenzelm@44330
   581
      (singleton o cond_forks)
wenzelm@44301
   582
        {name = "map_future", group = SOME group, deps = [task],
wenzelm@44301
   583
          pri = Task_Queue.pri_of_task task, interrupts = true}
wenzelm@41672
   584
        (fn () => f (join x))
wenzelm@29366
   585
  end;
wenzelm@28979
   586
wenzelm@28191
   587
wenzelm@34277
   588
(* promised futures -- fulfilled by external means *)
wenzelm@34277
   589
wenzelm@44298
   590
fun promise_group group abort : 'a future =
wenzelm@34277
   591
  let
wenzelm@35016
   592
    val result = Single_Assignment.var "promise" : 'a result;
wenzelm@44298
   593
    fun assign () = assign_result group result Exn.interrupt_exn
wenzelm@39243
   594
      handle Fail _ => true
wenzelm@39243
   595
        | exn =>
wenzelm@44298
   596
            if Exn.is_interrupt exn
wenzelm@44298
   597
            then raise Fail "Concurrent attempt to fulfill promise"
wenzelm@39243
   598
            else reraise exn;
wenzelm@44298
   599
    fun job () =
wenzelm@44298
   600
      Multithreading.with_attributes Multithreading.no_interrupts
wenzelm@44298
   601
        (fn _ => assign () before abort ());
wenzelm@37854
   602
    val task = SYNCHRONIZED "enqueue_passive" (fn () =>
wenzelm@44298
   603
      Unsynchronized.change_result queue (Task_Queue.enqueue_passive group job));
wenzelm@41683
   604
  in Future {promised = true, task = task, result = result} end;
wenzelm@34277
   605
wenzelm@44298
   606
fun promise abort = promise_group (worker_subgroup ()) abort;
wenzelm@34277
   607
wenzelm@41683
   608
fun fulfill_result (Future {promised, task, result}) res =
wenzelm@39243
   609
  if not promised then raise Fail "Not a promised future"
wenzelm@39243
   610
  else
wenzelm@39243
   611
    let
wenzelm@41683
   612
      val group = Task_Queue.group_of_task task;
wenzelm@39243
   613
      fun job ok = assign_result group result (if ok then res else Exn.interrupt_exn);
wenzelm@39243
   614
      val _ =
wenzelm@39243
   615
        Multithreading.with_attributes Multithreading.no_interrupts (fn _ =>
wenzelm@39243
   616
          let
wenzelm@39243
   617
            val still_passive =
wenzelm@39243
   618
              SYNCHRONIZED "fulfill_result" (fn () =>
wenzelm@39243
   619
                Unsynchronized.change_result queue
wenzelm@39243
   620
                  (Task_Queue.dequeue_passive (Thread.self ()) task));
wenzelm@44110
   621
          in if still_passive then worker_exec (task, [job]) else () end);
wenzelm@41681
   622
      val _ =
wenzelm@41695
   623
        if is_some (Single_Assignment.peek result) then ()
wenzelm@41695
   624
        else worker_waiting [task] (fn () => ignore (Single_Assignment.await result));
wenzelm@39243
   625
    in () end;
wenzelm@34277
   626
wenzelm@43761
   627
fun fulfill x res = fulfill_result x (Exn.Res res);
wenzelm@34277
   628
wenzelm@34277
   629
wenzelm@32228
   630
(* shutdown *)
wenzelm@29366
   631
wenzelm@28203
   632
fun shutdown () =
wenzelm@28276
   633
  if Multithreading.available then
wenzelm@28276
   634
    SYNCHRONIZED "shutdown" (fn () =>
wenzelm@32228
   635
     while scheduler_active () do
wenzelm@34279
   636
      (wait scheduler_event; broadcast_work ()))
wenzelm@28276
   637
  else ();
wenzelm@28203
   638
wenzelm@29366
   639
wenzelm@38236
   640
(* status markup *)
wenzelm@37690
   641
wenzelm@38236
   642
fun status e =
wenzelm@37690
   643
  let
wenzelm@40448
   644
    val task_props =
wenzelm@40448
   645
      (case worker_task () of
wenzelm@40448
   646
        NONE => I
wenzelm@45666
   647
      | SOME task => Markup.properties [(Isabelle_Markup.taskN, Task_Queue.str_of_task task)]);
wenzelm@45666
   648
    val _ = Output.status (Markup.markup_only (task_props Isabelle_Markup.forked));
wenzelm@37690
   649
    val x = e ();  (*sic -- report "joined" only for success*)
wenzelm@45666
   650
    val _ = Output.status (Markup.markup_only (task_props Isabelle_Markup.joined));
wenzelm@37690
   651
  in x end;
wenzelm@37690
   652
wenzelm@37690
   653
wenzelm@47404
   654
(* queue status *)
wenzelm@47404
   655
wenzelm@47404
   656
fun group_tasks group = Task_Queue.group_tasks (! queue) group;
wenzelm@47404
   657
wenzelm@47404
   658
fun queue_status () = Task_Queue.status (! queue);
wenzelm@47404
   659
wenzelm@47404
   660
wenzelm@29366
   661
(*final declarations of this structure!*)
wenzelm@29366
   662
val map = map_future;
wenzelm@29366
   663
wenzelm@28156
   664
end;
wenzelm@28972
   665
wenzelm@28972
   666
type 'a future = 'a Future.future;
wenzelm@28972
   667