src/Pure/Concurrent/future.ML
author haftmann
Thu Oct 22 13:48:06 2009 +0200 (2009-10-22)
changeset 33063 4d462963a7db
parent 32814 81897d30b97f
child 33061 e3e61133e0fc
permissions -rw-r--r--
map_range (and map_index) combinator
wenzelm@28156
     1
(*  Title:      Pure/Concurrent/future.ML
wenzelm@28156
     2
    Author:     Makarius
wenzelm@28156
     3
wenzelm@32246
     4
Future values, see also
wenzelm@32246
     5
http://www4.in.tum.de/~wenzelm/papers/parallel-isabelle.pdf
wenzelm@28201
     6
wenzelm@28201
     7
Notes:
wenzelm@28201
     8
wenzelm@28201
     9
  * Futures are similar to delayed evaluation, i.e. delay/force is
wenzelm@28201
    10
    generalized to fork/join (and variants).  The idea is to model
wenzelm@28201
    11
    parallel value-oriented computations, but *not* communicating
wenzelm@28201
    12
    processes.
wenzelm@28201
    13
wenzelm@28201
    14
  * Futures are grouped; failure of one group member causes the whole
wenzelm@32220
    15
    group to be interrupted eventually.  Groups are block-structured.
wenzelm@28201
    16
wenzelm@28201
    17
  * Forked futures are evaluated spontaneously by a farm of worker
wenzelm@28201
    18
    threads in the background; join resynchronizes the computation and
wenzelm@28201
    19
    delivers results (values or exceptions).
wenzelm@28201
    20
wenzelm@28201
    21
  * The pool of worker threads is limited, usually in correlation with
wenzelm@28201
    22
    the number of physical cores on the machine.  Note that allocation
wenzelm@28201
    23
    of runtime resources is distorted either if workers yield CPU time
wenzelm@28201
    24
    (e.g. via system sleep or wait operations), or if non-worker
wenzelm@28201
    25
    threads contend for significant runtime resources independently.
wenzelm@28156
    26
*)
wenzelm@28156
    27
wenzelm@28156
    28
signature FUTURE =
wenzelm@28156
    29
sig
wenzelm@29119
    30
  type task = Task_Queue.task
wenzelm@29119
    31
  type group = Task_Queue.group
wenzelm@32058
    32
  val is_worker: unit -> bool
wenzelm@32814
    33
  val worker_task: unit -> Task_Queue.task option
wenzelm@32102
    34
  val worker_group: unit -> Task_Queue.group option
wenzelm@28972
    35
  type 'a future
wenzelm@28972
    36
  val task_of: 'a future -> task
wenzelm@28972
    37
  val group_of: 'a future -> group
wenzelm@28972
    38
  val peek: 'a future -> 'a Exn.result option
wenzelm@28972
    39
  val is_finished: 'a future -> bool
wenzelm@28997
    40
  val value: 'a -> 'a future
wenzelm@28979
    41
  val fork_group: group -> (unit -> 'a) -> 'a future
wenzelm@32724
    42
  val fork_deps_pri: 'b future list -> int -> (unit -> 'a) -> 'a future
wenzelm@28979
    43
  val fork_deps: 'b future list -> (unit -> 'a) -> 'a future
wenzelm@29119
    44
  val fork_pri: int -> (unit -> 'a) -> 'a future
wenzelm@32724
    45
  val fork: (unit -> 'a) -> 'a future
wenzelm@28972
    46
  val join_results: 'a future list -> 'a Exn.result list
wenzelm@28972
    47
  val join_result: 'a future -> 'a Exn.result
wenzelm@28972
    48
  val join: 'a future -> 'a
wenzelm@28972
    49
  val map: ('a -> 'b) -> 'a future -> 'b future
wenzelm@30618
    50
  val interruptible_task: ('a -> 'b) -> 'a -> 'b
wenzelm@29431
    51
  val cancel_group: group -> unit
wenzelm@28972
    52
  val cancel: 'a future -> unit
wenzelm@28203
    53
  val shutdown: unit -> unit
wenzelm@28156
    54
end;
wenzelm@28156
    55
wenzelm@28156
    56
structure Future: FUTURE =
wenzelm@28156
    57
struct
wenzelm@28156
    58
wenzelm@28177
    59
(** future values **)
wenzelm@28177
    60
wenzelm@28167
    61
(* identifiers *)
wenzelm@28167
    62
wenzelm@29119
    63
type task = Task_Queue.task;
wenzelm@29119
    64
type group = Task_Queue.group;
wenzelm@28167
    65
wenzelm@32058
    66
local
wenzelm@32058
    67
  val tag = Universal.tag () : (string * task * group) option Universal.tag;
wenzelm@32058
    68
in
wenzelm@28177
    69
  fun thread_data () = the_default NONE (Thread.getLocal tag);
wenzelm@32058
    70
  fun setmp_thread_data data f x =
wenzelm@32058
    71
    Library.setmp_thread_data tag (thread_data ()) (SOME data) f x;
wenzelm@28167
    72
end;
wenzelm@28167
    73
wenzelm@32058
    74
val is_worker = is_some o thread_data;
wenzelm@32814
    75
val worker_task = Option.map #2 o thread_data;
wenzelm@32102
    76
val worker_group = Option.map #3 o thread_data;
wenzelm@32058
    77
wenzelm@28167
    78
wenzelm@28167
    79
(* datatype future *)
wenzelm@28167
    80
wenzelm@28972
    81
datatype 'a future = Future of
wenzelm@28167
    82
 {task: task,
wenzelm@28177
    83
  group: group,
wenzelm@32253
    84
  result: 'a Exn.result option Synchronized.var};
wenzelm@28167
    85
wenzelm@28167
    86
fun task_of (Future {task, ...}) = task;
wenzelm@28167
    87
fun group_of (Future {group, ...}) = group;
wenzelm@32253
    88
fun result_of (Future {result, ...}) = result;
wenzelm@28167
    89
wenzelm@32592
    90
fun peek x = Synchronized.value (result_of x);
wenzelm@28558
    91
fun is_finished x = is_some (peek x);
wenzelm@28320
    92
wenzelm@28997
    93
fun value x = Future
wenzelm@29119
    94
 {task = Task_Queue.new_task 0,
wenzelm@32102
    95
  group = Task_Queue.new_group NONE,
wenzelm@32253
    96
  result = Synchronized.var "future" (SOME (Exn.Result x))};
wenzelm@28997
    97
wenzelm@28167
    98
wenzelm@28177
    99
wenzelm@28177
   100
(** scheduling **)
wenzelm@28177
   101
wenzelm@28177
   102
(* global state *)
wenzelm@28177
   103
wenzelm@32738
   104
val queue = Unsynchronized.ref Task_Queue.empty;
wenzelm@32738
   105
val next = Unsynchronized.ref 0;
wenzelm@32738
   106
val workers = Unsynchronized.ref ([]: (Thread.thread * bool) list);
wenzelm@32738
   107
val scheduler = Unsynchronized.ref (NONE: Thread.thread option);
wenzelm@32738
   108
val excessive = Unsynchronized.ref 0;
wenzelm@32738
   109
val canceled = Unsynchronized.ref ([]: Task_Queue.group list);
wenzelm@32738
   110
val do_shutdown = Unsynchronized.ref false;
wenzelm@28177
   111
wenzelm@28177
   112
wenzelm@28177
   113
(* synchronization *)
wenzelm@28156
   114
wenzelm@32219
   115
val scheduler_event = ConditionVar.conditionVar ();
wenzelm@32219
   116
val work_available = ConditionVar.conditionVar ();
wenzelm@32219
   117
val work_finished = ConditionVar.conditionVar ();
wenzelm@32219
   118
wenzelm@28156
   119
local
wenzelm@28156
   120
  val lock = Mutex.mutex ();
wenzelm@28156
   121
in
wenzelm@28156
   122
wenzelm@28575
   123
fun SYNCHRONIZED name = SimpleThread.synchronized name lock;
wenzelm@28156
   124
wenzelm@32219
   125
fun wait cond = (*requires SYNCHRONIZED*)
wenzelm@32295
   126
  Multithreading.sync_wait NONE NONE cond lock;
wenzelm@28206
   127
wenzelm@32295
   128
fun wait_timeout timeout cond = (*requires SYNCHRONIZED*)
wenzelm@32295
   129
  Multithreading.sync_wait NONE (SOME (Time.+ (Time.now (), timeout))) cond lock;
wenzelm@28166
   130
wenzelm@32219
   131
fun signal cond = (*requires SYNCHRONIZED*)
wenzelm@32219
   132
  ConditionVar.signal cond;
wenzelm@32219
   133
wenzelm@32219
   134
fun broadcast cond = (*requires SYNCHRONIZED*)
wenzelm@28166
   135
  ConditionVar.broadcast cond;
wenzelm@28156
   136
wenzelm@32248
   137
fun broadcast_work () = (*requires SYNCHRONIZED*)
wenzelm@32248
   138
 (ConditionVar.broadcast work_available;
wenzelm@32225
   139
  ConditionVar.broadcast work_finished);
wenzelm@32225
   140
wenzelm@28156
   141
end;
wenzelm@28156
   142
wenzelm@28156
   143
wenzelm@32099
   144
(* execute future jobs *)
wenzelm@32099
   145
wenzelm@32099
   146
fun future_job group (e: unit -> 'a) =
wenzelm@32099
   147
  let
wenzelm@32253
   148
    val result = Synchronized.var "future" (NONE: 'a Exn.result option);
wenzelm@32107
   149
    fun job ok =
wenzelm@32107
   150
      let
wenzelm@32107
   151
        val res =
wenzelm@32107
   152
          if ok then
wenzelm@32230
   153
            Exn.capture (fn () =>
wenzelm@32295
   154
              Multithreading.with_attributes Multithreading.private_interrupts (fn _ => e ())) ()
wenzelm@32107
   155
          else Exn.Exn Exn.Interrupt;
wenzelm@32295
   156
        val _ = Synchronized.change result
wenzelm@32295
   157
          (fn NONE => SOME res
wenzelm@32295
   158
            | SOME _ => raise Fail "Duplicate assignment of future value");
wenzelm@32107
   159
      in
wenzelm@32107
   160
        (case res of
wenzelm@32107
   161
          Exn.Exn exn => (Task_Queue.cancel_group group exn; false)
wenzelm@32107
   162
        | Exn.Result _ => true)
wenzelm@32107
   163
      end;
wenzelm@32099
   164
  in (result, job) end;
wenzelm@28156
   165
wenzelm@29341
   166
fun do_cancel group = (*requires SYNCHRONIZED*)
wenzelm@32738
   167
 (Unsynchronized.change canceled (insert Task_Queue.eq_group group);
wenzelm@32738
   168
  broadcast scheduler_event);
wenzelm@29341
   169
wenzelm@29366
   170
fun execute name (task, group, jobs) =
wenzelm@28167
   171
  let
wenzelm@32102
   172
    val valid = not (Task_Queue.is_canceled group);
wenzelm@32058
   173
    val ok = setmp_thread_data (name, task, group) (fn () =>
wenzelm@29384
   174
      fold (fn job => fn ok => job valid andalso ok) jobs true) ();
wenzelm@32246
   175
    val _ = SYNCHRONIZED "finish" (fn () =>
wenzelm@32219
   176
      let
wenzelm@32738
   177
        val maximal = Unsynchronized.change_result queue (Task_Queue.finish task);
wenzelm@32219
   178
        val _ =
wenzelm@32219
   179
          if ok then ()
wenzelm@32219
   180
          else if Task_Queue.cancel (! queue) group then ()
wenzelm@32219
   181
          else do_cancel group;
wenzelm@32219
   182
        val _ = broadcast work_finished;
wenzelm@32219
   183
        val _ = if maximal then () else broadcast work_available;
wenzelm@32219
   184
      in () end);
wenzelm@28167
   185
  in () end;
wenzelm@28167
   186
wenzelm@28167
   187
wenzelm@32246
   188
(* worker activity *)
wenzelm@32246
   189
wenzelm@32246
   190
fun count_active () = (*requires SYNCHRONIZED*)
wenzelm@32246
   191
  fold (fn (_, active) => fn i => if active then i + 1 else i) (! workers) 0;
wenzelm@32246
   192
wenzelm@32246
   193
fun change_active active = (*requires SYNCHRONIZED*)
wenzelm@32738
   194
  Unsynchronized.change workers
wenzelm@32738
   195
    (AList.update Thread.equal (Thread.self (), active));
wenzelm@32246
   196
wenzelm@32246
   197
wenzelm@28167
   198
(* worker threads *)
wenzelm@28167
   199
wenzelm@32219
   200
fun worker_wait cond = (*requires SYNCHRONIZED*)
wenzelm@32286
   201
  (change_active false; wait cond; change_active true);
wenzelm@28162
   202
wenzelm@29119
   203
fun worker_next () = (*requires SYNCHRONIZED*)
wenzelm@28167
   204
  if ! excessive > 0 then
wenzelm@32738
   205
    (Unsynchronized.dec excessive;
wenzelm@32738
   206
     Unsynchronized.change workers
wenzelm@32738
   207
      (filter_out (fn (thread, _) => Thread.equal (thread, Thread.self ())));
wenzelm@32219
   208
     broadcast scheduler_event;
wenzelm@28167
   209
     NONE)
wenzelm@32246
   210
  else if count_active () > Multithreading.max_threads_value () then
wenzelm@32246
   211
    (worker_wait scheduler_event; worker_next ())
wenzelm@28166
   212
  else
wenzelm@32738
   213
    (case Unsynchronized.change_result queue (Task_Queue.dequeue (Thread.self ())) of
wenzelm@32219
   214
      NONE => (worker_wait work_available; worker_next ())
wenzelm@28166
   215
    | some => some);
wenzelm@28156
   216
wenzelm@28167
   217
fun worker_loop name =
wenzelm@32107
   218
  (case SYNCHRONIZED name (fn () => worker_next ()) of
wenzelm@29119
   219
    NONE => ()
wenzelm@28167
   220
  | SOME work => (execute name work; worker_loop name));
wenzelm@28156
   221
wenzelm@28167
   222
fun worker_start name = (*requires SYNCHRONIZED*)
wenzelm@32738
   223
  Unsynchronized.change workers (cons (SimpleThread.fork false (fn () =>
wenzelm@32248
   224
     (broadcast scheduler_event; worker_loop name)), true));
wenzelm@28156
   225
wenzelm@28156
   226
wenzelm@28156
   227
(* scheduler *)
wenzelm@28156
   228
wenzelm@32738
   229
val last_status = Unsynchronized.ref Time.zeroTime;
wenzelm@32248
   230
val next_status = Time.fromMilliseconds 500;
wenzelm@32248
   231
val next_round = Time.fromMilliseconds 50;
wenzelm@32226
   232
wenzelm@28206
   233
fun scheduler_next () = (*requires SYNCHRONIZED*)
wenzelm@28156
   234
  let
wenzelm@32226
   235
    (*queue and worker status*)
wenzelm@32226
   236
    val _ =
wenzelm@32226
   237
      let val now = Time.now () in
wenzelm@32226
   238
        if Time.> (Time.+ (! last_status, next_status), now) then ()
wenzelm@32226
   239
        else
wenzelm@32226
   240
         (last_status := now; Multithreading.tracing 1 (fn () =>
wenzelm@32226
   241
            let
wenzelm@32226
   242
              val {ready, pending, running} = Task_Queue.status (! queue);
wenzelm@32226
   243
              val total = length (! workers);
wenzelm@32246
   244
              val active = count_active ();
wenzelm@32226
   245
            in
wenzelm@32617
   246
              "SCHEDULE " ^ Time.toString now ^ ": " ^
wenzelm@32226
   247
                string_of_int ready ^ " ready, " ^
wenzelm@32226
   248
                string_of_int pending ^ " pending, " ^
wenzelm@32226
   249
                string_of_int running ^ " running; " ^
wenzelm@32226
   250
                string_of_int total ^ " workers, " ^
wenzelm@32226
   251
                string_of_int active ^ " active"
wenzelm@32226
   252
            end))
wenzelm@32226
   253
      end;
wenzelm@32053
   254
wenzelm@28206
   255
    (*worker threads*)
wenzelm@28191
   256
    val _ =
wenzelm@32219
   257
      if forall (Thread.isActive o #1) (! workers) then ()
wenzelm@32095
   258
      else
wenzelm@32219
   259
        (case List.partition (Thread.isActive o #1) (! workers) of
wenzelm@32095
   260
          (_, []) => ()
wenzelm@32220
   261
        | (alive, dead) =>
wenzelm@32220
   262
            (workers := alive; Multithreading.tracing 0 (fn () =>
wenzelm@32220
   263
              "SCHEDULE: disposed " ^ string_of_int (length dead) ^ " dead worker threads")));
wenzelm@28191
   264
wenzelm@28206
   265
    val m = if ! do_shutdown then 0 else Multithreading.max_threads_value ();
wenzelm@32644
   266
    val mm = if m = 9999 then 1 else m * 2;
wenzelm@32219
   267
    val l = length (! workers);
wenzelm@32095
   268
    val _ = excessive := l - mm;
wenzelm@28203
   269
    val _ =
wenzelm@32095
   270
      if mm > l then
wenzelm@32738
   271
        funpow (mm - l) (fn () =>
wenzelm@32738
   272
          worker_start ("worker " ^ string_of_int (Unsynchronized.inc next))) ()
wenzelm@28203
   273
      else ();
wenzelm@28206
   274
wenzelm@28206
   275
    (*canceled groups*)
wenzelm@32225
   276
    val _ =
wenzelm@32225
   277
      if null (! canceled) then ()
wenzelm@32293
   278
      else
wenzelm@32293
   279
       (Multithreading.tracing 1 (fn () =>
wenzelm@32293
   280
          string_of_int (length (! canceled)) ^ " canceled groups");
wenzelm@32738
   281
        Unsynchronized.change canceled (filter_out (Task_Queue.cancel (! queue)));
wenzelm@32293
   282
        broadcast_work ());
wenzelm@28206
   283
wenzelm@32228
   284
    (*delay loop*)
wenzelm@32295
   285
    val _ = Exn.release (wait_timeout next_round scheduler_event);
wenzelm@28167
   286
wenzelm@32219
   287
    (*shutdown*)
wenzelm@32228
   288
    val _ = if Task_Queue.is_empty (! queue) then do_shutdown := true else ();
wenzelm@32219
   289
    val continue = not (! do_shutdown andalso null (! workers));
wenzelm@32219
   290
    val _ = if continue then () else scheduler := NONE;
wenzelm@32219
   291
    val _ = broadcast scheduler_event;
wenzelm@32295
   292
  in continue end
wenzelm@32295
   293
  handle Exn.Interrupt =>
wenzelm@32295
   294
   (Multithreading.tracing 1 (fn () => "Interrupt");
wenzelm@32296
   295
    uninterruptible (fn _ => fn () => List.app do_cancel (Task_Queue.cancel_all (! queue))) ();
wenzelm@32295
   296
    scheduler_next ());
wenzelm@32295
   297
wenzelm@28206
   298
fun scheduler_loop () =
wenzelm@32295
   299
  Multithreading.with_attributes
wenzelm@32295
   300
    (Multithreading.sync_interrupts Multithreading.public_interrupts)
wenzelm@32295
   301
    (fn _ => while SYNCHRONIZED "scheduler" (fn () => scheduler_next ()) do ());
wenzelm@28156
   302
wenzelm@28203
   303
fun scheduler_active () = (*requires SYNCHRONIZED*)
wenzelm@28203
   304
  (case ! scheduler of NONE => false | SOME thread => Thread.isActive thread);
wenzelm@28203
   305
wenzelm@32228
   306
fun scheduler_check () = (*requires SYNCHRONIZED*)
wenzelm@32228
   307
 (do_shutdown := false;
wenzelm@32248
   308
  if scheduler_active () then ()
wenzelm@32248
   309
  else scheduler := SOME (SimpleThread.fork false scheduler_loop));
wenzelm@28156
   310
wenzelm@28156
   311
wenzelm@29366
   312
wenzelm@29366
   313
(** futures **)
wenzelm@28156
   314
wenzelm@29366
   315
(* fork *)
wenzelm@29366
   316
wenzelm@29366
   317
fun fork_future opt_group deps pri e =
wenzelm@29366
   318
  let
wenzelm@32102
   319
    val group =
wenzelm@32102
   320
      (case opt_group of
wenzelm@32102
   321
        SOME group => group
wenzelm@32102
   322
      | NONE => Task_Queue.new_group (worker_group ()));
wenzelm@29366
   323
    val (result, job) = future_job group e;
wenzelm@32246
   324
    val task = SYNCHRONIZED "enqueue" (fn () =>
wenzelm@32219
   325
      let
wenzelm@32738
   326
        val (task, minimal) =
wenzelm@32738
   327
          Unsynchronized.change_result queue (Task_Queue.enqueue group deps pri job);
wenzelm@32219
   328
        val _ = if minimal then signal work_available else ();
wenzelm@32228
   329
        val _ = scheduler_check ();
wenzelm@32219
   330
      in task end);
wenzelm@28166
   331
  in Future {task = task, group = group, result = result} end;
wenzelm@28162
   332
wenzelm@29366
   333
fun fork_group group e = fork_future (SOME group) [] 0 e;
wenzelm@32724
   334
fun fork_deps_pri deps pri e = fork_future NONE (map task_of deps) pri e;
wenzelm@32724
   335
fun fork_deps deps e = fork_deps_pri deps 0 e;
wenzelm@32724
   336
fun fork_pri pri e = fork_deps_pri [] pri e;
wenzelm@32724
   337
fun fork e = fork_deps [] e;
wenzelm@28186
   338
wenzelm@28186
   339
wenzelm@29366
   340
(* join *)
wenzelm@29366
   341
wenzelm@29551
   342
local
wenzelm@29551
   343
wenzelm@32099
   344
fun get_result x =
wenzelm@32099
   345
  (case peek x of
wenzelm@32102
   346
    NONE => Exn.Exn (SYS_ERROR "unfinished future")
wenzelm@32102
   347
  | SOME (Exn.Exn Exn.Interrupt) =>
wenzelm@32102
   348
      Exn.Exn (Exn.EXCEPTIONS (Exn.flatten_list (Task_Queue.group_status (group_of x))))
wenzelm@32102
   349
  | SOME res => res);
wenzelm@28186
   350
wenzelm@32224
   351
fun join_wait x =
wenzelm@32814
   352
  Synchronized.guarded_access (result_of x)
wenzelm@32814
   353
    (fn NONE => NONE | some => SOME ((), some));
wenzelm@32224
   354
wenzelm@32095
   355
fun join_next deps = (*requires SYNCHRONIZED*)
wenzelm@32224
   356
  if null deps then NONE
wenzelm@32224
   357
  else
wenzelm@32738
   358
    (case Unsynchronized.change_result queue (Task_Queue.dequeue_towards (Thread.self ()) deps) of
wenzelm@32224
   359
      (NONE, []) => NONE
wenzelm@32224
   360
    | (NONE, deps') => (worker_wait work_finished; join_next deps')
wenzelm@32224
   361
    | (SOME work, deps') => SOME (work, deps'));
wenzelm@32095
   362
wenzelm@32814
   363
fun execute_work NONE = ()
wenzelm@32814
   364
  | execute_work (SOME (work, deps')) = (execute "join" work; join_work deps')
wenzelm@32814
   365
and join_work deps =
wenzelm@32814
   366
  execute_work (SYNCHRONIZED "join" (fn () => join_next deps));
wenzelm@32814
   367
wenzelm@32814
   368
fun join_depend task deps =
wenzelm@32814
   369
  execute_work (SYNCHRONIZED "join" (fn () =>
wenzelm@32814
   370
    (Unsynchronized.change queue (Task_Queue.depend task deps); join_next deps)));
wenzelm@29551
   371
wenzelm@29551
   372
in
wenzelm@29551
   373
wenzelm@29366
   374
fun join_results xs =
wenzelm@29366
   375
  if forall is_finished xs then map get_result xs
wenzelm@32246
   376
  else if Multithreading.self_critical () then
wenzelm@32246
   377
    error "Cannot join future values within critical section"
wenzelm@32814
   378
  else
wenzelm@32814
   379
    (case worker_task () of
wenzelm@32814
   380
      SOME task => join_depend task (map task_of xs)
wenzelm@32814
   381
    | NONE => List.app join_wait xs;
wenzelm@32814
   382
    map get_result xs);
wenzelm@28186
   383
wenzelm@29551
   384
end;
wenzelm@29551
   385
wenzelm@28647
   386
fun join_result x = singleton join_results x;
wenzelm@28647
   387
fun join x = Exn.release (join_result x);
wenzelm@28156
   388
wenzelm@29366
   389
wenzelm@29366
   390
(* map *)
wenzelm@29366
   391
wenzelm@29384
   392
fun map_future f x =
wenzelm@29366
   393
  let
wenzelm@29384
   394
    val task = task_of x;
wenzelm@32102
   395
    val group = Task_Queue.new_group (SOME (group_of x));
wenzelm@29384
   396
    val (result, job) = future_job group (fn () => f (join x));
wenzelm@29384
   397
wenzelm@32246
   398
    val extended = SYNCHRONIZED "extend" (fn () =>
wenzelm@29366
   399
      (case Task_Queue.extend task job (! queue) of
wenzelm@29366
   400
        SOME queue' => (queue := queue'; true)
wenzelm@29366
   401
      | NONE => false));
wenzelm@29366
   402
  in
wenzelm@29384
   403
    if extended then Future {task = task, group = group, result = result}
wenzelm@32099
   404
    else fork_future (SOME group) [task] (Task_Queue.pri_of_task task) (fn () => f (join x))
wenzelm@29366
   405
  end;
wenzelm@28979
   406
wenzelm@28191
   407
wenzelm@29431
   408
(* cancellation *)
wenzelm@28202
   409
wenzelm@30618
   410
fun interruptible_task f x =
wenzelm@30618
   411
  if Multithreading.available then
wenzelm@30618
   412
    Multithreading.with_attributes
wenzelm@32058
   413
      (if is_worker ()
wenzelm@32295
   414
       then Multithreading.private_interrupts
wenzelm@32295
   415
       else Multithreading.public_interrupts)
wenzelm@32295
   416
      (fn _ => f x)
wenzelm@30618
   417
  else interruptible f x;
wenzelm@30618
   418
wenzelm@32228
   419
(*cancel: present and future group members will be interrupted eventually*)
wenzelm@32228
   420
fun cancel_group group = SYNCHRONIZED "cancel" (fn () => do_cancel group);
wenzelm@29431
   421
fun cancel x = cancel_group (group_of x);
wenzelm@28206
   422
wenzelm@29366
   423
wenzelm@32228
   424
(* shutdown *)
wenzelm@29366
   425
wenzelm@28203
   426
fun shutdown () =
wenzelm@28276
   427
  if Multithreading.available then
wenzelm@28276
   428
    SYNCHRONIZED "shutdown" (fn () =>
wenzelm@32228
   429
     while scheduler_active () do
wenzelm@32248
   430
      (wait scheduler_event; broadcast_work ()))
wenzelm@28276
   431
  else ();
wenzelm@28203
   432
wenzelm@29366
   433
wenzelm@29366
   434
(*final declarations of this structure!*)
wenzelm@29366
   435
val map = map_future;
wenzelm@29366
   436
wenzelm@28156
   437
end;
wenzelm@28972
   438
wenzelm@28972
   439
type 'a future = 'a Future.future;
wenzelm@28972
   440