src/Pure/Concurrent/future.ML
author wenzelm
Fri, 19 Aug 2011 16:13:26 +0200
changeset 44300 349cc426d929
parent 44299 061599cb6eb0
child 44301 65f60d9ac4bf
permissions -rw-r--r--
tuned signature -- treat structure Task_Queue as private to implementation;
Ignore whitespace changes - Everywhere: Within whitespace: At end of lines:
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
     1
(*  Title:      Pure/Concurrent/future.ML
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
     2
    Author:     Makarius
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
     3
44268
f6a11c1da821 tuned comments;
wenzelm
parents: 44249
diff changeset
     4
Value-oriented parallelism via futures and promises.  See also
32246
d4f7934e9555 misc tuning;
wenzelm
parents: 32230
diff changeset
     5
http://www4.in.tum.de/~wenzelm/papers/parallel-isabelle.pdf
37904
332cd0197d34 tuned comments;
wenzelm
parents: 37865
diff changeset
     6
http://www4.in.tum.de/~wenzelm/papers/parallel-ml.pdf
28201
7ae5cdb7b122 some general notes on future values;
wenzelm
parents: 28197
diff changeset
     7
7ae5cdb7b122 some general notes on future values;
wenzelm
parents: 28197
diff changeset
     8
Notes:
7ae5cdb7b122 some general notes on future values;
wenzelm
parents: 28197
diff changeset
     9
7ae5cdb7b122 some general notes on future values;
wenzelm
parents: 28197
diff changeset
    10
  * Futures are similar to delayed evaluation, i.e. delay/force is
44268
f6a11c1da821 tuned comments;
wenzelm
parents: 44249
diff changeset
    11
    generalized to fork/join.  The idea is to model parallel
f6a11c1da821 tuned comments;
wenzelm
parents: 44249
diff changeset
    12
    value-oriented computations (not communicating processes).
28201
7ae5cdb7b122 some general notes on future values;
wenzelm
parents: 28197
diff changeset
    13
7ae5cdb7b122 some general notes on future values;
wenzelm
parents: 28197
diff changeset
    14
  * Forked futures are evaluated spontaneously by a farm of worker
7ae5cdb7b122 some general notes on future values;
wenzelm
parents: 28197
diff changeset
    15
    threads in the background; join resynchronizes the computation and
7ae5cdb7b122 some general notes on future values;
wenzelm
parents: 28197
diff changeset
    16
    delivers results (values or exceptions).
7ae5cdb7b122 some general notes on future values;
wenzelm
parents: 28197
diff changeset
    17
7ae5cdb7b122 some general notes on future values;
wenzelm
parents: 28197
diff changeset
    18
  * The pool of worker threads is limited, usually in correlation with
7ae5cdb7b122 some general notes on future values;
wenzelm
parents: 28197
diff changeset
    19
    the number of physical cores on the machine.  Note that allocation
44268
f6a11c1da821 tuned comments;
wenzelm
parents: 44249
diff changeset
    20
    of runtime resources may be distorted either if workers yield CPU
f6a11c1da821 tuned comments;
wenzelm
parents: 44249
diff changeset
    21
    time (e.g. via system sleep or wait operations), or if non-worker
28201
7ae5cdb7b122 some general notes on future values;
wenzelm
parents: 28197
diff changeset
    22
    threads contend for significant runtime resources independently.
44268
f6a11c1da821 tuned comments;
wenzelm
parents: 44249
diff changeset
    23
    There is a limited number of replacement worker threads that get
f6a11c1da821 tuned comments;
wenzelm
parents: 44249
diff changeset
    24
    activated in certain explicit wait conditions.
34277
7325a5e3587f added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents: 33416
diff changeset
    25
44268
f6a11c1da821 tuned comments;
wenzelm
parents: 44249
diff changeset
    26
  * Future tasks are organized in groups, which are block-structured.
f6a11c1da821 tuned comments;
wenzelm
parents: 44249
diff changeset
    27
    When forking a new new task, the default is to open an individual
f6a11c1da821 tuned comments;
wenzelm
parents: 44249
diff changeset
    28
    subgroup, unless some common group is specified explicitly.
f6a11c1da821 tuned comments;
wenzelm
parents: 44249
diff changeset
    29
    Failure of one group member causes the immediate peers to be
f6a11c1da821 tuned comments;
wenzelm
parents: 44249
diff changeset
    30
    interrupted eventually (i.e. none by default).  Interrupted tasks
f6a11c1da821 tuned comments;
wenzelm
parents: 44249
diff changeset
    31
    that lack regular result information, will pick up parallel
f6a11c1da821 tuned comments;
wenzelm
parents: 44249
diff changeset
    32
    exceptions from the cumulative group context (as Par_Exn).
f6a11c1da821 tuned comments;
wenzelm
parents: 44249
diff changeset
    33
44299
061599cb6eb0 refined Future.cancel: explicit future allows to join actual cancellation;
wenzelm
parents: 44298
diff changeset
    34
  * Future task groups may be canceled: present and future group
061599cb6eb0 refined Future.cancel: explicit future allows to join actual cancellation;
wenzelm
parents: 44298
diff changeset
    35
    members will be interrupted eventually.
061599cb6eb0 refined Future.cancel: explicit future allows to join actual cancellation;
wenzelm
parents: 44298
diff changeset
    36
44268
f6a11c1da821 tuned comments;
wenzelm
parents: 44249
diff changeset
    37
  * Promised "passive" futures are fulfilled by external means.  There
f6a11c1da821 tuned comments;
wenzelm
parents: 44249
diff changeset
    38
    is no associated evaluation task, but other futures can depend on
f6a11c1da821 tuned comments;
wenzelm
parents: 44249
diff changeset
    39
    them via regular join operations.
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
    40
*)
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
    41
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
    42
signature FUTURE =
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
    43
sig
44300
349cc426d929 tuned signature -- treat structure Task_Queue as private to implementation;
wenzelm
parents: 44299
diff changeset
    44
  type task = Task_Queue.task
349cc426d929 tuned signature -- treat structure Task_Queue as private to implementation;
wenzelm
parents: 44299
diff changeset
    45
  type group = Task_Queue.group
349cc426d929 tuned signature -- treat structure Task_Queue as private to implementation;
wenzelm
parents: 44299
diff changeset
    46
  val new_group: group option -> group
349cc426d929 tuned signature -- treat structure Task_Queue as private to implementation;
wenzelm
parents: 44299
diff changeset
    47
  val worker_task: unit -> task option
349cc426d929 tuned signature -- treat structure Task_Queue as private to implementation;
wenzelm
parents: 44299
diff changeset
    48
  val worker_group: unit -> group option
349cc426d929 tuned signature -- treat structure Task_Queue as private to implementation;
wenzelm
parents: 44299
diff changeset
    49
  val worker_subgroup: unit -> group
28972
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
    50
  type 'a future
44300
349cc426d929 tuned signature -- treat structure Task_Queue as private to implementation;
wenzelm
parents: 44299
diff changeset
    51
  val task_of: 'a future -> task
28972
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
    52
  val peek: 'a future -> 'a Exn.result option
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
    53
  val is_finished: 'a future -> bool
44198
a41ea419de68 explicit check of finished evaluation;
wenzelm
parents: 44173
diff changeset
    54
  val get_finished: 'a future -> 'a
44113
0baa8bbd355a future_job: explicit indication of interrupts;
wenzelm
parents: 44111
diff changeset
    55
  type fork_params =
44300
349cc426d929 tuned signature -- treat structure Task_Queue as private to implementation;
wenzelm
parents: 44299
diff changeset
    56
    {name: string, group: group option, deps: task list, pri: int, interrupts: bool}
44113
0baa8bbd355a future_job: explicit indication of interrupts;
wenzelm
parents: 44111
diff changeset
    57
  val forks: fork_params -> (unit -> 'a) list -> 'a future list
29119
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
    58
  val fork_pri: int -> (unit -> 'a) -> 'a future
32724
aaeeb0ba2035 added fork_deps_pri;
wenzelm
parents: 32644
diff changeset
    59
  val fork: (unit -> 'a) -> 'a future
28972
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
    60
  val join_results: 'a future list -> 'a Exn.result list
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
    61
  val join_result: 'a future -> 'a Exn.result
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
    62
  val join: 'a future -> 'a
44294
a0ddd5760444 clarified Future.cond_forks: more uniform handling of exceptional situations;
wenzelm
parents: 44268
diff changeset
    63
  val value_result: 'a Exn.result -> 'a future
34277
7325a5e3587f added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents: 33416
diff changeset
    64
  val value: 'a -> 'a future
28972
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
    65
  val map: ('a -> 'b) -> 'a future -> 'b future
44300
349cc426d929 tuned signature -- treat structure Task_Queue as private to implementation;
wenzelm
parents: 44299
diff changeset
    66
  val cancel_group: group -> unit future
44299
061599cb6eb0 refined Future.cancel: explicit future allows to join actual cancellation;
wenzelm
parents: 44298
diff changeset
    67
  val cancel: 'a future -> unit future
061599cb6eb0 refined Future.cancel: explicit future allows to join actual cancellation;
wenzelm
parents: 44298
diff changeset
    68
  val interruptible_task: ('a -> 'b) -> 'a -> 'b
44113
0baa8bbd355a future_job: explicit indication of interrupts;
wenzelm
parents: 44111
diff changeset
    69
  val cond_forks: fork_params -> (unit -> 'a) list -> 'a future list
44300
349cc426d929 tuned signature -- treat structure Task_Queue as private to implementation;
wenzelm
parents: 44299
diff changeset
    70
  val promise_group: group -> (unit -> unit) -> 'a future
44298
b8f8488704e2 Future.promise: explicit abort operation (like uninterruptible future job);
wenzelm
parents: 44295
diff changeset
    71
  val promise: (unit -> unit) -> 'a future
34277
7325a5e3587f added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents: 33416
diff changeset
    72
  val fulfill_result: 'a future -> 'a Exn.result -> unit
7325a5e3587f added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents: 33416
diff changeset
    73
  val fulfill: 'a future -> 'a -> unit
28203
88f18387f1c9 shutdown: global join-and-shutdown operation;
wenzelm
parents: 28202
diff changeset
    74
  val shutdown: unit -> unit
38236
d8c7be27e01d explicitly distinguish Output.status (essential feedback) vs. Output.report (useful markup);
wenzelm
parents: 37904
diff changeset
    75
  val status: (unit -> 'a) -> 'a
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
    76
end;
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
    77
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
    78
structure Future: FUTURE =
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
    79
struct
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
    80
28177
8c0335bc9336 inherit group from running thread, or create a new one -- make it harder to re-use canceled groups;
wenzelm
parents: 28170
diff changeset
    81
(** future values **)
8c0335bc9336 inherit group from running thread, or create a new one -- make it harder to re-use canceled groups;
wenzelm
parents: 28170
diff changeset
    82
44300
349cc426d929 tuned signature -- treat structure Task_Queue as private to implementation;
wenzelm
parents: 44299
diff changeset
    83
type task = Task_Queue.task;
349cc426d929 tuned signature -- treat structure Task_Queue as private to implementation;
wenzelm
parents: 44299
diff changeset
    84
type group = Task_Queue.group;
349cc426d929 tuned signature -- treat structure Task_Queue as private to implementation;
wenzelm
parents: 44299
diff changeset
    85
val new_group = Task_Queue.new_group;
349cc426d929 tuned signature -- treat structure Task_Queue as private to implementation;
wenzelm
parents: 44299
diff changeset
    86
349cc426d929 tuned signature -- treat structure Task_Queue as private to implementation;
wenzelm
parents: 44299
diff changeset
    87
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
    88
(* identifiers *)
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
    89
32058
c76fd93b3b99 more abstract Future.is_worker;
wenzelm
parents: 32055
diff changeset
    90
local
44300
349cc426d929 tuned signature -- treat structure Task_Queue as private to implementation;
wenzelm
parents: 44299
diff changeset
    91
  val tag = Universal.tag () : task option Universal.tag;
32058
c76fd93b3b99 more abstract Future.is_worker;
wenzelm
parents: 32055
diff changeset
    92
in
41683
73dde8006820 maintain Task_Queue.group within Task_Queue.task;
wenzelm
parents: 41681
diff changeset
    93
  fun worker_task () = the_default NONE (Thread.getLocal tag);
44110
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
    94
  fun setmp_worker_task task f x = setmp_thread_data tag (worker_task ()) (SOME task) f x;
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
    95
end;
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
    96
41683
73dde8006820 maintain Task_Queue.group within Task_Queue.task;
wenzelm
parents: 41681
diff changeset
    97
val worker_group = Option.map Task_Queue.group_of_task o worker_task;
44300
349cc426d929 tuned signature -- treat structure Task_Queue as private to implementation;
wenzelm
parents: 44299
diff changeset
    98
fun worker_subgroup () = new_group (worker_group ());
34277
7325a5e3587f added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents: 33416
diff changeset
    99
41679
79716cb61bfd refined task timing: joining vs. waiting;
wenzelm
parents: 41674
diff changeset
   100
fun worker_joining e =
79716cb61bfd refined task timing: joining vs. waiting;
wenzelm
parents: 41674
diff changeset
   101
  (case worker_task () of
79716cb61bfd refined task timing: joining vs. waiting;
wenzelm
parents: 41674
diff changeset
   102
    NONE => e ()
79716cb61bfd refined task timing: joining vs. waiting;
wenzelm
parents: 41674
diff changeset
   103
  | SOME task => Task_Queue.joining task e);
79716cb61bfd refined task timing: joining vs. waiting;
wenzelm
parents: 41674
diff changeset
   104
41680
a4c822915eaa more informative task timing: some dependency tracking;
wenzelm
parents: 41679
diff changeset
   105
fun worker_waiting deps e =
41670
74010c6af0a4 added basic task timing;
wenzelm
parents: 40448
diff changeset
   106
  (case worker_task () of
74010c6af0a4 added basic task timing;
wenzelm
parents: 40448
diff changeset
   107
    NONE => e ()
41680
a4c822915eaa more informative task timing: some dependency tracking;
wenzelm
parents: 41679
diff changeset
   108
  | SOME task => Task_Queue.waiting task deps e);
41670
74010c6af0a4 added basic task timing;
wenzelm
parents: 40448
diff changeset
   109
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
   110
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
   111
(* datatype future *)
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
   112
35016
c0f2e49bccfc result: Single_Assignment.var;
wenzelm
parents: 34281
diff changeset
   113
type 'a result = 'a Exn.result Single_Assignment.var;
c0f2e49bccfc result: Single_Assignment.var;
wenzelm
parents: 34281
diff changeset
   114
28972
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
   115
datatype 'a future = Future of
34277
7325a5e3587f added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents: 33416
diff changeset
   116
 {promised: bool,
44300
349cc426d929 tuned signature -- treat structure Task_Queue as private to implementation;
wenzelm
parents: 44299
diff changeset
   117
  task: task,
35016
c0f2e49bccfc result: Single_Assignment.var;
wenzelm
parents: 34281
diff changeset
   118
  result: 'a result};
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
   119
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
   120
fun task_of (Future {task, ...}) = task;
32253
d9def420c84e future result: Synchronized.var;
wenzelm
parents: 32249
diff changeset
   121
fun result_of (Future {result, ...}) = result;
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
   122
35016
c0f2e49bccfc result: Single_Assignment.var;
wenzelm
parents: 34281
diff changeset
   123
fun peek x = Single_Assignment.peek (result_of x);
28558
2a6297b4273c replaced str_of by general peek;
wenzelm
parents: 28548
diff changeset
   124
fun is_finished x = is_some (peek x);
28320
c6aef67f964d added is_finished;
wenzelm
parents: 28304
diff changeset
   125
44198
a41ea419de68 explicit check of finished evaluation;
wenzelm
parents: 44173
diff changeset
   126
fun get_finished x =
a41ea419de68 explicit check of finished evaluation;
wenzelm
parents: 44173
diff changeset
   127
  (case peek x of
a41ea419de68 explicit check of finished evaluation;
wenzelm
parents: 44173
diff changeset
   128
    SOME res => Exn.release res
a41ea419de68 explicit check of finished evaluation;
wenzelm
parents: 44173
diff changeset
   129
  | NONE => raise Fail "Unfinished future evaluation");
a41ea419de68 explicit check of finished evaluation;
wenzelm
parents: 44173
diff changeset
   130
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
   131
28177
8c0335bc9336 inherit group from running thread, or create a new one -- make it harder to re-use canceled groups;
wenzelm
parents: 28170
diff changeset
   132
8c0335bc9336 inherit group from running thread, or create a new one -- make it harder to re-use canceled groups;
wenzelm
parents: 28170
diff changeset
   133
(** scheduling **)
8c0335bc9336 inherit group from running thread, or create a new one -- make it harder to re-use canceled groups;
wenzelm
parents: 28170
diff changeset
   134
8c0335bc9336 inherit group from running thread, or create a new one -- make it harder to re-use canceled groups;
wenzelm
parents: 28170
diff changeset
   135
(* synchronization *)
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   136
32219
9a2566d1fdbd more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents: 32186
diff changeset
   137
val scheduler_event = ConditionVar.conditionVar ();
9a2566d1fdbd more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents: 32186
diff changeset
   138
val work_available = ConditionVar.conditionVar ();
9a2566d1fdbd more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents: 32186
diff changeset
   139
val work_finished = ConditionVar.conditionVar ();
9a2566d1fdbd more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents: 32186
diff changeset
   140
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   141
local
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   142
  val lock = Mutex.mutex ();
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   143
in
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   144
37216
3165bc303f66 modernized some structure names, keeping a few legacy aliases;
wenzelm
parents: 37182
diff changeset
   145
fun SYNCHRONIZED name = Simple_Thread.synchronized name lock;
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   146
32219
9a2566d1fdbd more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents: 32186
diff changeset
   147
fun wait cond = (*requires SYNCHRONIZED*)
32295
400cc493d466 renamed Multithreading.regular_interrupts to Multithreading.public_interrupts;
wenzelm
parents: 32293
diff changeset
   148
  Multithreading.sync_wait NONE NONE cond lock;
28206
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   149
32295
400cc493d466 renamed Multithreading.regular_interrupts to Multithreading.public_interrupts;
wenzelm
parents: 32293
diff changeset
   150
fun wait_timeout timeout cond = (*requires SYNCHRONIZED*)
400cc493d466 renamed Multithreading.regular_interrupts to Multithreading.public_interrupts;
wenzelm
parents: 32293
diff changeset
   151
  Multithreading.sync_wait NONE (SOME (Time.+ (Time.now (), timeout))) cond lock;
28166
43087721a66e moved task, thread_data, group, queue to task_queue.ML;
wenzelm
parents: 28163
diff changeset
   152
32219
9a2566d1fdbd more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents: 32186
diff changeset
   153
fun signal cond = (*requires SYNCHRONIZED*)
9a2566d1fdbd more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents: 32186
diff changeset
   154
  ConditionVar.signal cond;
9a2566d1fdbd more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents: 32186
diff changeset
   155
9a2566d1fdbd more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents: 32186
diff changeset
   156
fun broadcast cond = (*requires SYNCHRONIZED*)
28166
43087721a66e moved task, thread_data, group, queue to task_queue.ML;
wenzelm
parents: 28163
diff changeset
   157
  ConditionVar.broadcast cond;
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   158
32248
0241916a5f06 more precise treatment of scheduler_event: continous pulse (50ms) instead of flooding, which was burning many CPU cycles in spare threads;
wenzelm
parents: 32247
diff changeset
   159
fun broadcast_work () = (*requires SYNCHRONIZED*)
0241916a5f06 more precise treatment of scheduler_event: continous pulse (50ms) instead of flooding, which was burning many CPU cycles in spare threads;
wenzelm
parents: 32247
diff changeset
   160
 (ConditionVar.broadcast work_available;
32225
d5d6f47fb018 cancel: improved reactivity due to more careful broadcasting;
wenzelm
parents: 32224
diff changeset
   161
  ConditionVar.broadcast work_finished);
d5d6f47fb018 cancel: improved reactivity due to more careful broadcasting;
wenzelm
parents: 32224
diff changeset
   162
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   163
end;
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   164
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   165
33410
e351f4c1f18c worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents: 33409
diff changeset
   166
(* global state *)
e351f4c1f18c worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents: 33409
diff changeset
   167
e351f4c1f18c worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents: 33409
diff changeset
   168
val queue = Unsynchronized.ref Task_Queue.empty;
e351f4c1f18c worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents: 33409
diff changeset
   169
val next = Unsynchronized.ref 0;
e351f4c1f18c worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents: 33409
diff changeset
   170
val scheduler = Unsynchronized.ref (NONE: Thread.thread option);
44300
349cc426d929 tuned signature -- treat structure Task_Queue as private to implementation;
wenzelm
parents: 44299
diff changeset
   171
val canceled = Unsynchronized.ref ([]: group list);
33410
e351f4c1f18c worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents: 33409
diff changeset
   172
val do_shutdown = Unsynchronized.ref false;
e351f4c1f18c worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents: 33409
diff changeset
   173
val max_workers = Unsynchronized.ref 0;
e351f4c1f18c worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents: 33409
diff changeset
   174
val max_active = Unsynchronized.ref 0;
33411
a07558eb5029 worker_next: treat wait for work_available as Sleeping, not Waiting;
wenzelm
parents: 33410
diff changeset
   175
val worker_trend = Unsynchronized.ref 0;
33410
e351f4c1f18c worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents: 33409
diff changeset
   176
e351f4c1f18c worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents: 33409
diff changeset
   177
datatype worker_state = Working | Waiting | Sleeping;
e351f4c1f18c worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents: 33409
diff changeset
   178
val workers = Unsynchronized.ref ([]: (Thread.thread * worker_state Unsynchronized.ref) list);
e351f4c1f18c worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents: 33409
diff changeset
   179
e351f4c1f18c worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents: 33409
diff changeset
   180
fun count_workers state = (*requires SYNCHRONIZED*)
e351f4c1f18c worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents: 33409
diff changeset
   181
  fold (fn (_, state_ref) => fn i => if ! state_ref = state then i + 1 else i) (! workers) 0;
e351f4c1f18c worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents: 33409
diff changeset
   182
e351f4c1f18c worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents: 33409
diff changeset
   183
44110
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   184
(* cancellation primitives *)
32099
5382c93108db propagate exceptions within future groups;
wenzelm
parents: 32096
diff changeset
   185
34279
02936e77a07c tasks of canceled groups are considered "ready" -- enables to purge the queue from tasks depending on unfinished promises (also improves general reactivity);
wenzelm
parents: 34277
diff changeset
   186
fun cancel_now group = (*requires SYNCHRONIZED*)
34280
16bf3e9786a3 eliminated cache, which complicates the code without making a real difference (NB: deque_towards often disrupts caching, and daisy-chaining of workers already reduces queue overhead);
wenzelm
parents: 34279
diff changeset
   187
  Task_Queue.cancel (! queue) group;
34279
02936e77a07c tasks of canceled groups are considered "ready" -- enables to purge the queue from tasks depending on unfinished promises (also improves general reactivity);
wenzelm
parents: 34277
diff changeset
   188
02936e77a07c tasks of canceled groups are considered "ready" -- enables to purge the queue from tasks depending on unfinished promises (also improves general reactivity);
wenzelm
parents: 34277
diff changeset
   189
fun cancel_later group = (*requires SYNCHRONIZED*)
32738
15bb09ca0378 explicit indication of Unsynchronized.ref;
wenzelm
parents: 32724
diff changeset
   190
 (Unsynchronized.change canceled (insert Task_Queue.eq_group group);
15bb09ca0378 explicit indication of Unsynchronized.ref;
wenzelm
parents: 32724
diff changeset
   191
  broadcast scheduler_event);
29341
6bb007a0f9f2 more reactive scheduler: reduced loop timeout, propagate broadcast interrupt via TaskQueue.cancel_all;
wenzelm
parents: 29119
diff changeset
   192
44110
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   193
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   194
(* worker threads *)
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   195
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   196
fun worker_exec (task, jobs) =
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
   197
  let
41683
73dde8006820 maintain Task_Queue.group within Task_Queue.task;
wenzelm
parents: 41681
diff changeset
   198
    val group = Task_Queue.group_of_task task;
32102
81d03a29980c added worker_group;
wenzelm
parents: 32099
diff changeset
   199
    val valid = not (Task_Queue.is_canceled group);
41670
74010c6af0a4 added basic task timing;
wenzelm
parents: 40448
diff changeset
   200
    val ok =
74010c6af0a4 added basic task timing;
wenzelm
parents: 40448
diff changeset
   201
      Task_Queue.running task (fn () =>
41683
73dde8006820 maintain Task_Queue.group within Task_Queue.task;
wenzelm
parents: 41681
diff changeset
   202
        setmp_worker_task task (fn () =>
41670
74010c6af0a4 added basic task timing;
wenzelm
parents: 40448
diff changeset
   203
          fold (fn job => fn ok => job valid andalso ok) jobs true) ());
41776
3bd83302a3c3 less verbose tracing;
wenzelm
parents: 41737
diff changeset
   204
    val _ = Multithreading.tracing 2 (fn () =>
41670
74010c6af0a4 added basic task timing;
wenzelm
parents: 40448
diff changeset
   205
      let
43951
804783d6ee26 more detailed tracing;
wenzelm
parents: 43761
diff changeset
   206
        val s = Task_Queue.str_of_task_groups task;
41670
74010c6af0a4 added basic task timing;
wenzelm
parents: 40448
diff changeset
   207
        fun micros time = string_of_int (Time.toNanoseconds time div 1000);
41680
a4c822915eaa more informative task timing: some dependency tracking;
wenzelm
parents: 41679
diff changeset
   208
        val (run, wait, deps) = Task_Queue.timing_of_task task;
a4c822915eaa more informative task timing: some dependency tracking;
wenzelm
parents: 41679
diff changeset
   209
      in "TASK " ^ s ^ " " ^ micros run ^ " " ^ micros wait ^ " (" ^ commas deps ^ ")" end);
32246
d4f7934e9555 misc tuning;
wenzelm
parents: 32230
diff changeset
   210
    val _ = SYNCHRONIZED "finish" (fn () =>
32219
9a2566d1fdbd more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents: 32186
diff changeset
   211
      let
32738
15bb09ca0378 explicit indication of Unsynchronized.ref;
wenzelm
parents: 32724
diff changeset
   212
        val maximal = Unsynchronized.change_result queue (Task_Queue.finish task);
44295
e43f0ea90c9a more focused use of Multithreading.interrupted: retain interrupts within task group boundary, without loss of information;
wenzelm
parents: 44294
diff changeset
   213
        val test = Exn.capture Multithreading.interrupted ();
32219
9a2566d1fdbd more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents: 32186
diff changeset
   214
        val _ =
44295
e43f0ea90c9a more focused use of Multithreading.interrupted: retain interrupts within task group boundary, without loss of information;
wenzelm
parents: 44294
diff changeset
   215
          if ok andalso not (Exn.is_interrupt_exn test) then ()
44299
061599cb6eb0 refined Future.cancel: explicit future allows to join actual cancellation;
wenzelm
parents: 44298
diff changeset
   216
          else if null (cancel_now group) then ()
34279
02936e77a07c tasks of canceled groups are considered "ready" -- enables to purge the queue from tasks depending on unfinished promises (also improves general reactivity);
wenzelm
parents: 34277
diff changeset
   217
          else cancel_later group;
32219
9a2566d1fdbd more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents: 32186
diff changeset
   218
        val _ = broadcast work_finished;
33413
cb409680dda8 avoid broadcast work_available, use daisy-chained signal instead;
wenzelm
parents: 33411
diff changeset
   219
        val _ = if maximal then () else signal work_available;
32219
9a2566d1fdbd more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents: 32186
diff changeset
   220
      in () end);
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
   221
  in () end;
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
   222
33410
e351f4c1f18c worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents: 33409
diff changeset
   223
fun worker_wait active cond = (*requires SYNCHRONIZED*)
33406
1ddcb8472bd2 slightly leaner and more direct control of worker activity etc.;
wenzelm
parents: 33061
diff changeset
   224
  let
33410
e351f4c1f18c worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents: 33409
diff changeset
   225
    val state =
e351f4c1f18c worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents: 33409
diff changeset
   226
      (case AList.lookup Thread.equal (! workers) (Thread.self ()) of
e351f4c1f18c worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents: 33409
diff changeset
   227
        SOME state => state
e351f4c1f18c worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents: 33409
diff changeset
   228
      | NONE => raise Fail "Unregistered worker thread");
e351f4c1f18c worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents: 33409
diff changeset
   229
    val _ = state := (if active then Waiting else Sleeping);
33406
1ddcb8472bd2 slightly leaner and more direct control of worker activity etc.;
wenzelm
parents: 33061
diff changeset
   230
    val _ = wait cond;
33410
e351f4c1f18c worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents: 33409
diff changeset
   231
    val _ = state := Working;
33406
1ddcb8472bd2 slightly leaner and more direct control of worker activity etc.;
wenzelm
parents: 33061
diff changeset
   232
  in () end;
28162
55772e4e95e0 tuned check_cache;
wenzelm
parents: 28156
diff changeset
   233
33415
352fe8e9162d worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents: 33413
diff changeset
   234
fun worker_next () = (*requires SYNCHRONIZED*)
33406
1ddcb8472bd2 slightly leaner and more direct control of worker activity etc.;
wenzelm
parents: 33061
diff changeset
   235
  if length (! workers) > ! max_workers then
1ddcb8472bd2 slightly leaner and more direct control of worker activity etc.;
wenzelm
parents: 33061
diff changeset
   236
    (Unsynchronized.change workers (AList.delete Thread.equal (Thread.self ()));
33415
352fe8e9162d worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents: 33413
diff changeset
   237
     signal work_available;
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
   238
     NONE)
33410
e351f4c1f18c worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents: 33409
diff changeset
   239
  else if count_workers Working > ! max_active then
33415
352fe8e9162d worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents: 33413
diff changeset
   240
    (worker_wait false work_available; worker_next ())
28166
43087721a66e moved task, thread_data, group, queue to task_queue.ML;
wenzelm
parents: 28163
diff changeset
   241
  else
32738
15bb09ca0378 explicit indication of Unsynchronized.ref;
wenzelm
parents: 32724
diff changeset
   242
    (case Unsynchronized.change_result queue (Task_Queue.dequeue (Thread.self ())) of
33415
352fe8e9162d worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents: 33413
diff changeset
   243
      NONE => (worker_wait false work_available; worker_next ())
33413
cb409680dda8 avoid broadcast work_available, use daisy-chained signal instead;
wenzelm
parents: 33411
diff changeset
   244
    | some => (signal work_available; some));
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   245
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
   246
fun worker_loop name =
33415
352fe8e9162d worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents: 33413
diff changeset
   247
  (case SYNCHRONIZED name (fn () => worker_next ()) of
29119
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
   248
    NONE => ()
44295
e43f0ea90c9a more focused use of Multithreading.interrupted: retain interrupts within task group boundary, without loss of information;
wenzelm
parents: 44294
diff changeset
   249
  | SOME work => (worker_exec work; worker_loop name));
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   250
33407
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   251
fun worker_start name = (*requires SYNCHRONIZED*)
37216
3165bc303f66 modernized some structure names, keeping a few legacy aliases;
wenzelm
parents: 37182
diff changeset
   252
  Unsynchronized.change workers (cons (Simple_Thread.fork false (fn () => worker_loop name),
33410
e351f4c1f18c worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents: 33409
diff changeset
   253
    Unsynchronized.ref Working));
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   254
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   255
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   256
(* scheduler *)
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   257
33407
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   258
val status_ticks = Unsynchronized.ref 0;
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   259
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   260
val last_round = Unsynchronized.ref Time.zeroTime;
40301
bf39a257b3d3 simplified some time constants;
wenzelm
parents: 39243
diff changeset
   261
val next_round = seconds 0.05;
32226
23511e4da055 tuned tracing;
wenzelm
parents: 32225
diff changeset
   262
28206
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   263
fun scheduler_next () = (*requires SYNCHRONIZED*)
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   264
  let
33407
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   265
    val now = Time.now ();
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   266
    val tick = Time.<= (Time.+ (! last_round, next_round), now);
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   267
    val _ = if tick then last_round := now else ();
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   268
33415
352fe8e9162d worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents: 33413
diff changeset
   269
352fe8e9162d worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents: 33413
diff changeset
   270
    (* queue and worker status *)
352fe8e9162d worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents: 33413
diff changeset
   271
32226
23511e4da055 tuned tracing;
wenzelm
parents: 32225
diff changeset
   272
    val _ =
33407
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   273
      if tick then Unsynchronized.change status_ticks (fn i => (i + 1) mod 10) else ();
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   274
    val _ =
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   275
      if tick andalso ! status_ticks = 0 then
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   276
        Multithreading.tracing 1 (fn () =>
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   277
          let
34277
7325a5e3587f added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents: 33416
diff changeset
   278
            val {ready, pending, running, passive} = Task_Queue.status (! queue);
33407
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   279
            val total = length (! workers);
33410
e351f4c1f18c worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents: 33409
diff changeset
   280
            val active = count_workers Working;
e351f4c1f18c worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents: 33409
diff changeset
   281
            val waiting = count_workers Waiting;
33407
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   282
          in
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   283
            "SCHEDULE " ^ Time.toString now ^ ": " ^
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   284
              string_of_int ready ^ " ready, " ^
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   285
              string_of_int pending ^ " pending, " ^
34277
7325a5e3587f added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents: 33416
diff changeset
   286
              string_of_int running ^ " running, " ^
7325a5e3587f added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents: 33416
diff changeset
   287
              string_of_int passive ^ " passive; " ^
33407
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   288
              string_of_int total ^ " workers, " ^
33410
e351f4c1f18c worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents: 33409
diff changeset
   289
              string_of_int active ^ " active, " ^
e351f4c1f18c worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents: 33409
diff changeset
   290
              string_of_int waiting ^ " waiting "
33407
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   291
          end)
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   292
      else ();
32053
257eac3946e9 scheduler: tuned tracing (status);
wenzelm
parents: 31631
diff changeset
   293
28191
9e5f556409c6 future: allow explicit group;
wenzelm
parents: 28186
diff changeset
   294
    val _ =
32219
9a2566d1fdbd more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents: 32186
diff changeset
   295
      if forall (Thread.isActive o #1) (! workers) then ()
32095
ad4be204fdfe tuned tracing;
wenzelm
parents: 32058
diff changeset
   296
      else
33409
wenzelm
parents: 33408
diff changeset
   297
        let
37682
ad5489a6be48 tuned white space;
wenzelm
parents: 37216
diff changeset
   298
          val (alive, dead) = List.partition (Thread.isActive o #1) (! workers);
33409
wenzelm
parents: 33408
diff changeset
   299
          val _ = workers := alive;
wenzelm
parents: 33408
diff changeset
   300
        in
wenzelm
parents: 33408
diff changeset
   301
          Multithreading.tracing 0 (fn () =>
wenzelm
parents: 33408
diff changeset
   302
            "SCHEDULE: disposed " ^ string_of_int (length dead) ^ " dead worker threads")
wenzelm
parents: 33408
diff changeset
   303
        end;
28191
9e5f556409c6 future: allow explicit group;
wenzelm
parents: 28186
diff changeset
   304
33415
352fe8e9162d worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents: 33413
diff changeset
   305
352fe8e9162d worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents: 33413
diff changeset
   306
    (* worker pool adjustments *)
352fe8e9162d worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents: 33413
diff changeset
   307
352fe8e9162d worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents: 33413
diff changeset
   308
    val max_active0 = ! max_active;
352fe8e9162d worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents: 33413
diff changeset
   309
    val max_workers0 = ! max_workers;
352fe8e9162d worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents: 33413
diff changeset
   310
28206
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   311
    val m = if ! do_shutdown then 0 else Multithreading.max_threads_value ();
33406
1ddcb8472bd2 slightly leaner and more direct control of worker activity etc.;
wenzelm
parents: 33061
diff changeset
   312
    val _ = max_active := m;
1ddcb8472bd2 slightly leaner and more direct control of worker activity etc.;
wenzelm
parents: 33061
diff changeset
   313
33411
a07558eb5029 worker_next: treat wait for work_available as Sleeping, not Waiting;
wenzelm
parents: 33410
diff changeset
   314
    val mm =
a07558eb5029 worker_next: treat wait for work_available as Sleeping, not Waiting;
wenzelm
parents: 33410
diff changeset
   315
      if ! do_shutdown then 0
a07558eb5029 worker_next: treat wait for work_available as Sleeping, not Waiting;
wenzelm
parents: 33410
diff changeset
   316
      else if m = 9999 then 1
33413
cb409680dda8 avoid broadcast work_available, use daisy-chained signal instead;
wenzelm
parents: 33411
diff changeset
   317
      else Int.min (Int.max (count_workers Working + 2 * count_workers Waiting, m), 4 * m);
33411
a07558eb5029 worker_next: treat wait for work_available as Sleeping, not Waiting;
wenzelm
parents: 33410
diff changeset
   318
    val _ =
a07558eb5029 worker_next: treat wait for work_available as Sleeping, not Waiting;
wenzelm
parents: 33410
diff changeset
   319
      if tick andalso mm > ! max_workers then
a07558eb5029 worker_next: treat wait for work_available as Sleeping, not Waiting;
wenzelm
parents: 33410
diff changeset
   320
        Unsynchronized.change worker_trend (fn w => if w < 0 then 0 else w + 1)
a07558eb5029 worker_next: treat wait for work_available as Sleeping, not Waiting;
wenzelm
parents: 33410
diff changeset
   321
      else if tick andalso mm < ! max_workers then
a07558eb5029 worker_next: treat wait for work_available as Sleeping, not Waiting;
wenzelm
parents: 33410
diff changeset
   322
        Unsynchronized.change worker_trend (fn w => if w > 0 then 0 else w - 1)
a07558eb5029 worker_next: treat wait for work_available as Sleeping, not Waiting;
wenzelm
parents: 33410
diff changeset
   323
      else ();
a07558eb5029 worker_next: treat wait for work_available as Sleeping, not Waiting;
wenzelm
parents: 33410
diff changeset
   324
    val _ =
33415
352fe8e9162d worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents: 33413
diff changeset
   325
      if mm = 0 orelse ! worker_trend > 50 orelse ! worker_trend < ~50 then
352fe8e9162d worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents: 33413
diff changeset
   326
        max_workers := mm
44173
aaaa13e297dc immediate fork of initial workers -- avoid 5 ticks (250ms) for adaptive scheme (a07558eb5029);
wenzelm
parents: 44115
diff changeset
   327
      else if ! worker_trend > 5 andalso ! max_workers < 2 * m orelse ! max_workers = 0 then
33415
352fe8e9162d worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents: 33413
diff changeset
   328
        max_workers := Int.min (mm, 2 * m)
33411
a07558eb5029 worker_next: treat wait for work_available as Sleeping, not Waiting;
wenzelm
parents: 33410
diff changeset
   329
      else ();
33406
1ddcb8472bd2 slightly leaner and more direct control of worker activity etc.;
wenzelm
parents: 33061
diff changeset
   330
33407
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   331
    val missing = ! max_workers - length (! workers);
28203
88f18387f1c9 shutdown: global join-and-shutdown operation;
wenzelm
parents: 28202
diff changeset
   332
    val _ =
33407
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   333
      if missing > 0 then
33415
352fe8e9162d worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents: 33413
diff changeset
   334
        funpow missing (fn () =>
352fe8e9162d worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents: 33413
diff changeset
   335
          ignore (worker_start ("worker " ^ string_of_int (Unsynchronized.inc next)))) ()
28203
88f18387f1c9 shutdown: global join-and-shutdown operation;
wenzelm
parents: 28202
diff changeset
   336
      else ();
28206
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   337
33415
352fe8e9162d worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents: 33413
diff changeset
   338
    val _ =
352fe8e9162d worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents: 33413
diff changeset
   339
      if ! max_active = max_active0 andalso ! max_workers = max_workers0 then ()
352fe8e9162d worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents: 33413
diff changeset
   340
      else signal work_available;
352fe8e9162d worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents: 33413
diff changeset
   341
352fe8e9162d worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents: 33413
diff changeset
   342
352fe8e9162d worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents: 33413
diff changeset
   343
    (* canceled groups *)
352fe8e9162d worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents: 33413
diff changeset
   344
32225
d5d6f47fb018 cancel: improved reactivity due to more careful broadcasting;
wenzelm
parents: 32224
diff changeset
   345
    val _ =
d5d6f47fb018 cancel: improved reactivity due to more careful broadcasting;
wenzelm
parents: 32224
diff changeset
   346
      if null (! canceled) then ()
32293
e0b8da3fae4d tuned tracing;
wenzelm
parents: 32286
diff changeset
   347
      else
e0b8da3fae4d tuned tracing;
wenzelm
parents: 32286
diff changeset
   348
       (Multithreading.tracing 1 (fn () =>
e0b8da3fae4d tuned tracing;
wenzelm
parents: 32286
diff changeset
   349
          string_of_int (length (! canceled)) ^ " canceled groups");
44299
061599cb6eb0 refined Future.cancel: explicit future allows to join actual cancellation;
wenzelm
parents: 44298
diff changeset
   350
        Unsynchronized.change canceled (filter_out (null o cancel_now));
32293
e0b8da3fae4d tuned tracing;
wenzelm
parents: 32286
diff changeset
   351
        broadcast_work ());
28206
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   352
33415
352fe8e9162d worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents: 33413
diff changeset
   353
352fe8e9162d worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents: 33413
diff changeset
   354
    (* delay loop *)
352fe8e9162d worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents: 33413
diff changeset
   355
32295
400cc493d466 renamed Multithreading.regular_interrupts to Multithreading.public_interrupts;
wenzelm
parents: 32293
diff changeset
   356
    val _ = Exn.release (wait_timeout next_round scheduler_event);
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
   357
33415
352fe8e9162d worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents: 33413
diff changeset
   358
352fe8e9162d worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents: 33413
diff changeset
   359
    (* shutdown *)
352fe8e9162d worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents: 33413
diff changeset
   360
34277
7325a5e3587f added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents: 33416
diff changeset
   361
    val _ = if Task_Queue.all_passive (! queue) then do_shutdown := true else ();
32219
9a2566d1fdbd more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents: 32186
diff changeset
   362
    val continue = not (! do_shutdown andalso null (! workers));
9a2566d1fdbd more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents: 32186
diff changeset
   363
    val _ = if continue then () else scheduler := NONE;
33415
352fe8e9162d worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents: 33413
diff changeset
   364
32219
9a2566d1fdbd more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents: 32186
diff changeset
   365
    val _ = broadcast scheduler_event;
32295
400cc493d466 renamed Multithreading.regular_interrupts to Multithreading.public_interrupts;
wenzelm
parents: 32293
diff changeset
   366
  in continue end
39232
69c6d3e87660 more abstract treatment of interrupts in structure Exn -- hardly ever need to mention Interrupt literally;
wenzelm
parents: 38236
diff changeset
   367
  handle exn =>
69c6d3e87660 more abstract treatment of interrupts in structure Exn -- hardly ever need to mention Interrupt literally;
wenzelm
parents: 38236
diff changeset
   368
    if Exn.is_interrupt exn then
69c6d3e87660 more abstract treatment of interrupts in structure Exn -- hardly ever need to mention Interrupt literally;
wenzelm
parents: 38236
diff changeset
   369
     (Multithreading.tracing 1 (fn () => "Interrupt");
69c6d3e87660 more abstract treatment of interrupts in structure Exn -- hardly ever need to mention Interrupt literally;
wenzelm
parents: 38236
diff changeset
   370
      List.app cancel_later (Task_Queue.cancel_all (! queue));
69c6d3e87660 more abstract treatment of interrupts in structure Exn -- hardly ever need to mention Interrupt literally;
wenzelm
parents: 38236
diff changeset
   371
      broadcast_work (); true)
69c6d3e87660 more abstract treatment of interrupts in structure Exn -- hardly ever need to mention Interrupt literally;
wenzelm
parents: 38236
diff changeset
   372
    else reraise exn;
32295
400cc493d466 renamed Multithreading.regular_interrupts to Multithreading.public_interrupts;
wenzelm
parents: 32293
diff changeset
   373
28206
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   374
fun scheduler_loop () =
44173
aaaa13e297dc immediate fork of initial workers -- avoid 5 ticks (250ms) for adaptive scheme (a07558eb5029);
wenzelm
parents: 44115
diff changeset
   375
 (while
33416
13d00799fe49 scheduler: clarified interrupt attributes and handling;
wenzelm
parents: 33415
diff changeset
   376
    Multithreading.with_attributes
13d00799fe49 scheduler: clarified interrupt attributes and handling;
wenzelm
parents: 33415
diff changeset
   377
      (Multithreading.sync_interrupts Multithreading.public_interrupts)
13d00799fe49 scheduler: clarified interrupt attributes and handling;
wenzelm
parents: 33415
diff changeset
   378
      (fn _ => SYNCHRONIZED "scheduler" (fn () => scheduler_next ()))
44173
aaaa13e297dc immediate fork of initial workers -- avoid 5 ticks (250ms) for adaptive scheme (a07558eb5029);
wenzelm
parents: 44115
diff changeset
   379
  do (); last_round := Time.zeroTime);
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   380
28203
88f18387f1c9 shutdown: global join-and-shutdown operation;
wenzelm
parents: 28202
diff changeset
   381
fun scheduler_active () = (*requires SYNCHRONIZED*)
88f18387f1c9 shutdown: global join-and-shutdown operation;
wenzelm
parents: 28202
diff changeset
   382
  (case ! scheduler of NONE => false | SOME thread => Thread.isActive thread);
88f18387f1c9 shutdown: global join-and-shutdown operation;
wenzelm
parents: 28202
diff changeset
   383
32228
7622c03141b0 scheduler: shutdown spontaneously (after some delay) if queue is empty;
wenzelm
parents: 32227
diff changeset
   384
fun scheduler_check () = (*requires SYNCHRONIZED*)
7622c03141b0 scheduler: shutdown spontaneously (after some delay) if queue is empty;
wenzelm
parents: 32227
diff changeset
   385
 (do_shutdown := false;
32248
0241916a5f06 more precise treatment of scheduler_event: continous pulse (50ms) instead of flooding, which was burning many CPU cycles in spare threads;
wenzelm
parents: 32247
diff changeset
   386
  if scheduler_active () then ()
37216
3165bc303f66 modernized some structure names, keeping a few legacy aliases;
wenzelm
parents: 37182
diff changeset
   387
  else scheduler := SOME (Simple_Thread.fork false scheduler_loop));
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   388
44299
061599cb6eb0 refined Future.cancel: explicit future allows to join actual cancellation;
wenzelm
parents: 44298
diff changeset
   389
fun scheduler_cancel group = SYNCHRONIZED "scheduler_cancel" (fn () =>
061599cb6eb0 refined Future.cancel: explicit future allows to join actual cancellation;
wenzelm
parents: 44298
diff changeset
   390
  let
061599cb6eb0 refined Future.cancel: explicit future allows to join actual cancellation;
wenzelm
parents: 44298
diff changeset
   391
    val running = cancel_now group;
061599cb6eb0 refined Future.cancel: explicit future allows to join actual cancellation;
wenzelm
parents: 44298
diff changeset
   392
    val _ =
061599cb6eb0 refined Future.cancel: explicit future allows to join actual cancellation;
wenzelm
parents: 44298
diff changeset
   393
      if null running then ()
061599cb6eb0 refined Future.cancel: explicit future allows to join actual cancellation;
wenzelm
parents: 44298
diff changeset
   394
      else (cancel_later group; signal work_available; scheduler_check ());
061599cb6eb0 refined Future.cancel: explicit future allows to join actual cancellation;
wenzelm
parents: 44298
diff changeset
   395
  in running end);
061599cb6eb0 refined Future.cancel: explicit future allows to join actual cancellation;
wenzelm
parents: 44298
diff changeset
   396
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   397
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   398
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   399
(** futures **)
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   400
44110
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   401
(* future jobs *)
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   402
44249
64620f1d6f87 identify parallel exceptions where they emerge first -- to achieve unique results within evaluation graph;
wenzelm
parents: 44247
diff changeset
   403
fun assign_result group result raw_res =
44110
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   404
  let
44249
64620f1d6f87 identify parallel exceptions where they emerge first -- to achieve unique results within evaluation graph;
wenzelm
parents: 44247
diff changeset
   405
    val res =
64620f1d6f87 identify parallel exceptions where they emerge first -- to achieve unique results within evaluation graph;
wenzelm
parents: 44247
diff changeset
   406
      (case raw_res of
64620f1d6f87 identify parallel exceptions where they emerge first -- to achieve unique results within evaluation graph;
wenzelm
parents: 44247
diff changeset
   407
        Exn.Exn exn => Exn.Exn (#2 (Par_Exn.serial exn))
64620f1d6f87 identify parallel exceptions where they emerge first -- to achieve unique results within evaluation graph;
wenzelm
parents: 44247
diff changeset
   408
      | _ => raw_res);
44110
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   409
    val _ = Single_Assignment.assign result res
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   410
      handle exn as Fail _ =>
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   411
        (case Single_Assignment.peek result of
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   412
          SOME (Exn.Exn e) => reraise (if Exn.is_interrupt e then e else exn)
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   413
        | _ => reraise exn);
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   414
    val ok =
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   415
      (case the (Single_Assignment.peek result) of
44111
2d16c693d536 synchronized cancel and flushing of Multithreading.interrupted state, to ensure that interrupts stay within task boundaries;
wenzelm
parents: 44110
diff changeset
   416
        Exn.Exn exn =>
2d16c693d536 synchronized cancel and flushing of Multithreading.interrupted state, to ensure that interrupts stay within task boundaries;
wenzelm
parents: 44110
diff changeset
   417
          (SYNCHRONIZED "cancel" (fn () => Task_Queue.cancel_group group exn); false)
44110
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   418
      | Exn.Res _ => true);
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   419
  in ok end;
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   420
44113
0baa8bbd355a future_job: explicit indication of interrupts;
wenzelm
parents: 44111
diff changeset
   421
fun future_job group interrupts (e: unit -> 'a) =
44110
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   422
  let
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   423
    val result = Single_Assignment.var "future" : 'a result;
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   424
    val pos = Position.thread_data ();
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   425
    fun job ok =
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   426
      let
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   427
        val res =
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   428
          if ok then
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   429
            Exn.capture (fn () =>
44113
0baa8bbd355a future_job: explicit indication of interrupts;
wenzelm
parents: 44111
diff changeset
   430
              Multithreading.with_attributes
0baa8bbd355a future_job: explicit indication of interrupts;
wenzelm
parents: 44111
diff changeset
   431
                (if interrupts
0baa8bbd355a future_job: explicit indication of interrupts;
wenzelm
parents: 44111
diff changeset
   432
                 then Multithreading.private_interrupts else Multithreading.no_interrupts)
44295
e43f0ea90c9a more focused use of Multithreading.interrupted: retain interrupts within task group boundary, without loss of information;
wenzelm
parents: 44294
diff changeset
   433
                (fn _ => Position.setmp_thread_data pos e ())) ()
44110
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   434
          else Exn.interrupt_exn;
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   435
      in assign_result group result res end;
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   436
  in (result, job) end;
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   437
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   438
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   439
(* fork *)
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   440
44113
0baa8bbd355a future_job: explicit indication of interrupts;
wenzelm
parents: 44111
diff changeset
   441
type fork_params =
44300
349cc426d929 tuned signature -- treat structure Task_Queue as private to implementation;
wenzelm
parents: 44299
diff changeset
   442
  {name: string, group: group option, deps: task list, pri: int, interrupts: bool};
44113
0baa8bbd355a future_job: explicit indication of interrupts;
wenzelm
parents: 44111
diff changeset
   443
0baa8bbd355a future_job: explicit indication of interrupts;
wenzelm
parents: 44111
diff changeset
   444
fun forks ({name, group, deps, pri, interrupts}: fork_params) es =
41674
7da257539a8d tuned signature;
wenzelm
parents: 41673
diff changeset
   445
  if null es then []
7da257539a8d tuned signature;
wenzelm
parents: 41673
diff changeset
   446
  else
7da257539a8d tuned signature;
wenzelm
parents: 41673
diff changeset
   447
    let
7da257539a8d tuned signature;
wenzelm
parents: 41673
diff changeset
   448
      val grp =
7da257539a8d tuned signature;
wenzelm
parents: 41673
diff changeset
   449
        (case group of
7da257539a8d tuned signature;
wenzelm
parents: 41673
diff changeset
   450
          NONE => worker_subgroup ()
7da257539a8d tuned signature;
wenzelm
parents: 41673
diff changeset
   451
        | SOME grp => grp);
41708
d2e6b1132df2 tuned signature;
wenzelm
parents: 41695
diff changeset
   452
      fun enqueue e queue =
41674
7da257539a8d tuned signature;
wenzelm
parents: 41673
diff changeset
   453
        let
44113
0baa8bbd355a future_job: explicit indication of interrupts;
wenzelm
parents: 44111
diff changeset
   454
          val (result, job) = future_job grp interrupts e;
41708
d2e6b1132df2 tuned signature;
wenzelm
parents: 41695
diff changeset
   455
          val (task, queue') = Task_Queue.enqueue name grp deps pri job queue;
41683
73dde8006820 maintain Task_Queue.group within Task_Queue.task;
wenzelm
parents: 41681
diff changeset
   456
          val future = Future {promised = false, task = task, result = result};
41708
d2e6b1132df2 tuned signature;
wenzelm
parents: 41695
diff changeset
   457
        in (future, queue') end;
41674
7da257539a8d tuned signature;
wenzelm
parents: 41673
diff changeset
   458
    in
7da257539a8d tuned signature;
wenzelm
parents: 41673
diff changeset
   459
      SYNCHRONIZED "enqueue" (fn () =>
7da257539a8d tuned signature;
wenzelm
parents: 41673
diff changeset
   460
        let
41708
d2e6b1132df2 tuned signature;
wenzelm
parents: 41695
diff changeset
   461
          val (futures, queue') = fold_map enqueue es (! queue);
d2e6b1132df2 tuned signature;
wenzelm
parents: 41695
diff changeset
   462
          val _ = queue := queue';
d2e6b1132df2 tuned signature;
wenzelm
parents: 41695
diff changeset
   463
          val minimal = forall (not o Task_Queue.known_task queue') deps;
41674
7da257539a8d tuned signature;
wenzelm
parents: 41673
diff changeset
   464
          val _ = if minimal then signal work_available else ();
7da257539a8d tuned signature;
wenzelm
parents: 41673
diff changeset
   465
          val _ = scheduler_check ();
7da257539a8d tuned signature;
wenzelm
parents: 41673
diff changeset
   466
        in futures end)
7da257539a8d tuned signature;
wenzelm
parents: 41673
diff changeset
   467
    end;
28162
55772e4e95e0 tuned check_cache;
wenzelm
parents: 28156
diff changeset
   468
44113
0baa8bbd355a future_job: explicit indication of interrupts;
wenzelm
parents: 44111
diff changeset
   469
fun fork_pri pri e =
0baa8bbd355a future_job: explicit indication of interrupts;
wenzelm
parents: 44111
diff changeset
   470
  singleton (forks {name = "", group = NONE, deps = [], pri = pri, interrupts = true}) e;
0baa8bbd355a future_job: explicit indication of interrupts;
wenzelm
parents: 44111
diff changeset
   471
41672
2f70b1ddd09f more direct Future.bulk, which potentially reduces overhead for Par_List;
wenzelm
parents: 41670
diff changeset
   472
fun fork e = fork_pri 0 e;
28186
6a8417f36837 cancel: check_scheduler;
wenzelm
parents: 28177
diff changeset
   473
6a8417f36837 cancel: check_scheduler;
wenzelm
parents: 28177
diff changeset
   474
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   475
(* join *)
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   476
29551
95e469919c3e join_results: when dependencies are resulved (but not finished yet),
wenzelm
parents: 29431
diff changeset
   477
local
95e469919c3e join_results: when dependencies are resulved (but not finished yet),
wenzelm
parents: 29431
diff changeset
   478
32099
5382c93108db propagate exceptions within future groups;
wenzelm
parents: 32096
diff changeset
   479
fun get_result x =
5382c93108db propagate exceptions within future groups;
wenzelm
parents: 32096
diff changeset
   480
  (case peek x of
37852
a902f158b4fc eliminated old-style sys_error/SYS_ERROR in favour of exception Fail -- after careful checking that there is no overlap with existing handling of that;
wenzelm
parents: 37690
diff changeset
   481
    NONE => Exn.Exn (Fail "Unfinished future")
39232
69c6d3e87660 more abstract treatment of interrupts in structure Exn -- hardly ever need to mention Interrupt literally;
wenzelm
parents: 38236
diff changeset
   482
  | SOME res =>
69c6d3e87660 more abstract treatment of interrupts in structure Exn -- hardly ever need to mention Interrupt literally;
wenzelm
parents: 38236
diff changeset
   483
      if Exn.is_interrupt_exn res then
44247
270366301bd7 more systematic handling of parallel exceptions;
wenzelm
parents: 44198
diff changeset
   484
        (case Task_Queue.group_status (Task_Queue.group_of_task (task_of x)) of
270366301bd7 more systematic handling of parallel exceptions;
wenzelm
parents: 44198
diff changeset
   485
          NONE => res
270366301bd7 more systematic handling of parallel exceptions;
wenzelm
parents: 44198
diff changeset
   486
        | SOME exn => Exn.Exn exn)
39232
69c6d3e87660 more abstract treatment of interrupts in structure Exn -- hardly ever need to mention Interrupt literally;
wenzelm
parents: 38236
diff changeset
   487
      else res);
28186
6a8417f36837 cancel: check_scheduler;
wenzelm
parents: 28177
diff changeset
   488
32095
ad4be204fdfe tuned tracing;
wenzelm
parents: 32058
diff changeset
   489
fun join_next deps = (*requires SYNCHRONIZED*)
41695
afdbec23b92b eliminated slightly odd abstract type Task_Queue.deps;
wenzelm
parents: 41683
diff changeset
   490
  if null deps then NONE
32224
a46f5e9b1718 dequeue_towards: always return active tasks;
wenzelm
parents: 32222
diff changeset
   491
  else
41681
b5d7b15166bf Future.join_results: discontinued post-hoc recording of dynamic dependencies;
wenzelm
parents: 41680
diff changeset
   492
    (case Unsynchronized.change_result queue (Task_Queue.dequeue_deps (Thread.self ()) deps) of
41695
afdbec23b92b eliminated slightly odd abstract type Task_Queue.deps;
wenzelm
parents: 41683
diff changeset
   493
      (NONE, []) => NONE
afdbec23b92b eliminated slightly odd abstract type Task_Queue.deps;
wenzelm
parents: 41683
diff changeset
   494
    | (NONE, deps') =>
afdbec23b92b eliminated slightly odd abstract type Task_Queue.deps;
wenzelm
parents: 41683
diff changeset
   495
        (worker_waiting deps' (fn () => worker_wait true work_finished); join_next deps')
32224
a46f5e9b1718 dequeue_towards: always return active tasks;
wenzelm
parents: 32222
diff changeset
   496
    | (SOME work, deps') => SOME (work, deps'));
32095
ad4be204fdfe tuned tracing;
wenzelm
parents: 32058
diff changeset
   497
32814
81897d30b97f added Task_Queue.depend (again) -- light-weight version for transitive graph;
wenzelm
parents: 32738
diff changeset
   498
fun execute_work NONE = ()
44110
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   499
  | execute_work (SOME (work, deps')) =
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   500
      (worker_joining (fn () => worker_exec work); join_work deps')
32814
81897d30b97f added Task_Queue.depend (again) -- light-weight version for transitive graph;
wenzelm
parents: 32738
diff changeset
   501
and join_work deps =
43538
de5c79682b56 more robust join_results: join_work needs to be uninterruptible, otherwise the task being dequeued by join_next might be never executed/finished!
wenzelm
parents: 42128
diff changeset
   502
  Multithreading.with_attributes Multithreading.no_interrupts
de5c79682b56 more robust join_results: join_work needs to be uninterruptible, otherwise the task being dequeued by join_next might be never executed/finished!
wenzelm
parents: 42128
diff changeset
   503
    (fn _ => execute_work (SYNCHRONIZED "join" (fn () => join_next deps)));
32814
81897d30b97f added Task_Queue.depend (again) -- light-weight version for transitive graph;
wenzelm
parents: 32738
diff changeset
   504
29551
95e469919c3e join_results: when dependencies are resulved (but not finished yet),
wenzelm
parents: 29431
diff changeset
   505
in
95e469919c3e join_results: when dependencies are resulved (but not finished yet),
wenzelm
parents: 29431
diff changeset
   506
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   507
fun join_results xs =
41679
79716cb61bfd refined task timing: joining vs. waiting;
wenzelm
parents: 41674
diff changeset
   508
  let
79716cb61bfd refined task timing: joining vs. waiting;
wenzelm
parents: 41674
diff changeset
   509
    val _ =
79716cb61bfd refined task timing: joining vs. waiting;
wenzelm
parents: 41674
diff changeset
   510
      if forall is_finished xs then ()
79716cb61bfd refined task timing: joining vs. waiting;
wenzelm
parents: 41674
diff changeset
   511
      else if Multithreading.self_critical () then
79716cb61bfd refined task timing: joining vs. waiting;
wenzelm
parents: 41674
diff changeset
   512
        error "Cannot join future values within critical section"
41695
afdbec23b92b eliminated slightly odd abstract type Task_Queue.deps;
wenzelm
parents: 41683
diff changeset
   513
      else if is_some (worker_task ()) then join_work (map task_of xs)
41681
b5d7b15166bf Future.join_results: discontinued post-hoc recording of dynamic dependencies;
wenzelm
parents: 41680
diff changeset
   514
      else List.app (ignore o Single_Assignment.await o result_of) xs;
41679
79716cb61bfd refined task timing: joining vs. waiting;
wenzelm
parents: 41674
diff changeset
   515
  in map get_result xs end;
28186
6a8417f36837 cancel: check_scheduler;
wenzelm
parents: 28177
diff changeset
   516
29551
95e469919c3e join_results: when dependencies are resulved (but not finished yet),
wenzelm
parents: 29431
diff changeset
   517
end;
95e469919c3e join_results: when dependencies are resulved (but not finished yet),
wenzelm
parents: 29431
diff changeset
   518
28647
8068cdc84e7e join_results: allow CRITICAL join of finished futures;
wenzelm
parents: 28645
diff changeset
   519
fun join_result x = singleton join_results x;
8068cdc84e7e join_results: allow CRITICAL join of finished futures;
wenzelm
parents: 28645
diff changeset
   520
fun join x = Exn.release (join_result x);
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   521
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   522
44110
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   523
(* fast-path versions -- bypassing task queue *)
34277
7325a5e3587f added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents: 33416
diff changeset
   524
44294
a0ddd5760444 clarified Future.cond_forks: more uniform handling of exceptional situations;
wenzelm
parents: 44268
diff changeset
   525
fun value_result (res: 'a Exn.result) =
34277
7325a5e3587f added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents: 33416
diff changeset
   526
  let
41683
73dde8006820 maintain Task_Queue.group within Task_Queue.task;
wenzelm
parents: 41681
diff changeset
   527
    val task = Task_Queue.dummy_task ();
73dde8006820 maintain Task_Queue.group within Task_Queue.task;
wenzelm
parents: 41681
diff changeset
   528
    val group = Task_Queue.group_of_task task;
35016
c0f2e49bccfc result: Single_Assignment.var;
wenzelm
parents: 34281
diff changeset
   529
    val result = Single_Assignment.var "value" : 'a result;
44294
a0ddd5760444 clarified Future.cond_forks: more uniform handling of exceptional situations;
wenzelm
parents: 44268
diff changeset
   530
    val _ = assign_result group result res;
41683
73dde8006820 maintain Task_Queue.group within Task_Queue.task;
wenzelm
parents: 41681
diff changeset
   531
  in Future {promised = false, task = task, result = result} end;
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   532
44294
a0ddd5760444 clarified Future.cond_forks: more uniform handling of exceptional situations;
wenzelm
parents: 44268
diff changeset
   533
fun value x = value_result (Exn.Res x);
a0ddd5760444 clarified Future.cond_forks: more uniform handling of exceptional situations;
wenzelm
parents: 44268
diff changeset
   534
29384
a3c7e9ae9b71 more robust propagation of errors through bulk jobs;
wenzelm
parents: 29366
diff changeset
   535
fun map_future f x =
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   536
  let
29384
a3c7e9ae9b71 more robust propagation of errors through bulk jobs;
wenzelm
parents: 29366
diff changeset
   537
    val task = task_of x;
44300
349cc426d929 tuned signature -- treat structure Task_Queue as private to implementation;
wenzelm
parents: 44299
diff changeset
   538
    val group = new_group (SOME (Task_Queue.group_of_task task));
44113
0baa8bbd355a future_job: explicit indication of interrupts;
wenzelm
parents: 44111
diff changeset
   539
    val (result, job) = future_job group true (fn () => f (join x));
29384
a3c7e9ae9b71 more robust propagation of errors through bulk jobs;
wenzelm
parents: 29366
diff changeset
   540
32246
d4f7934e9555 misc tuning;
wenzelm
parents: 32230
diff changeset
   541
    val extended = SYNCHRONIZED "extend" (fn () =>
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   542
      (case Task_Queue.extend task job (! queue) of
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   543
        SOME queue' => (queue := queue'; true)
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   544
      | NONE => false));
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   545
  in
41683
73dde8006820 maintain Task_Queue.group within Task_Queue.task;
wenzelm
parents: 41681
diff changeset
   546
    if extended then Future {promised = false, task = task, result = result}
41672
2f70b1ddd09f more direct Future.bulk, which potentially reduces overhead for Par_List;
wenzelm
parents: 41670
diff changeset
   547
    else
41673
1c191a39549f support named tasks, for improved tracing;
wenzelm
parents: 41672
diff changeset
   548
      singleton
44113
0baa8bbd355a future_job: explicit indication of interrupts;
wenzelm
parents: 44111
diff changeset
   549
        (forks {name = "Future.map", group = SOME group, deps = [task],
0baa8bbd355a future_job: explicit indication of interrupts;
wenzelm
parents: 44111
diff changeset
   550
          pri = Task_Queue.pri_of_task task, interrupts = true})
41672
2f70b1ddd09f more direct Future.bulk, which potentially reduces overhead for Par_List;
wenzelm
parents: 41670
diff changeset
   551
        (fn () => f (join x))
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   552
  end;
28979
3ce619d8d432 fork/map: no inheritance of group (structure is nested, not parallel);
wenzelm
parents: 28972
diff changeset
   553
42128
eb84d28961a9 added Future.cond_forks convenience;
wenzelm
parents: 41776
diff changeset
   554
fun cond_forks args es =
eb84d28961a9 added Future.cond_forks convenience;
wenzelm
parents: 41776
diff changeset
   555
  if Multithreading.enabled () then forks args es
44294
a0ddd5760444 clarified Future.cond_forks: more uniform handling of exceptional situations;
wenzelm
parents: 44268
diff changeset
   556
  else map (fn e => value_result (Exn.interruptible_capture e ())) es;
42128
eb84d28961a9 added Future.cond_forks convenience;
wenzelm
parents: 41776
diff changeset
   557
28191
9e5f556409c6 future: allow explicit group;
wenzelm
parents: 28186
diff changeset
   558
44299
061599cb6eb0 refined Future.cancel: explicit future allows to join actual cancellation;
wenzelm
parents: 44298
diff changeset
   559
(* cancel *)
061599cb6eb0 refined Future.cancel: explicit future allows to join actual cancellation;
wenzelm
parents: 44298
diff changeset
   560
061599cb6eb0 refined Future.cancel: explicit future allows to join actual cancellation;
wenzelm
parents: 44298
diff changeset
   561
fun cancel_group group =
061599cb6eb0 refined Future.cancel: explicit future allows to join actual cancellation;
wenzelm
parents: 44298
diff changeset
   562
  (case scheduler_cancel group of
061599cb6eb0 refined Future.cancel: explicit future allows to join actual cancellation;
wenzelm
parents: 44298
diff changeset
   563
    [] => value ()
061599cb6eb0 refined Future.cancel: explicit future allows to join actual cancellation;
wenzelm
parents: 44298
diff changeset
   564
  | running =>
061599cb6eb0 refined Future.cancel: explicit future allows to join actual cancellation;
wenzelm
parents: 44298
diff changeset
   565
      singleton
44300
349cc426d929 tuned signature -- treat structure Task_Queue as private to implementation;
wenzelm
parents: 44299
diff changeset
   566
        (forks {name = "cancel_group", group = SOME (new_group NONE),
44299
061599cb6eb0 refined Future.cancel: explicit future allows to join actual cancellation;
wenzelm
parents: 44298
diff changeset
   567
          deps = running, pri = 0, interrupts = false}) I);
061599cb6eb0 refined Future.cancel: explicit future allows to join actual cancellation;
wenzelm
parents: 44298
diff changeset
   568
061599cb6eb0 refined Future.cancel: explicit future allows to join actual cancellation;
wenzelm
parents: 44298
diff changeset
   569
fun cancel x = cancel_group (Task_Queue.group_of_task (task_of x));
061599cb6eb0 refined Future.cancel: explicit future allows to join actual cancellation;
wenzelm
parents: 44298
diff changeset
   570
061599cb6eb0 refined Future.cancel: explicit future allows to join actual cancellation;
wenzelm
parents: 44298
diff changeset
   571
fun interruptible_task f x =
061599cb6eb0 refined Future.cancel: explicit future allows to join actual cancellation;
wenzelm
parents: 44298
diff changeset
   572
  (if Multithreading.available then
061599cb6eb0 refined Future.cancel: explicit future allows to join actual cancellation;
wenzelm
parents: 44298
diff changeset
   573
    Multithreading.with_attributes
061599cb6eb0 refined Future.cancel: explicit future allows to join actual cancellation;
wenzelm
parents: 44298
diff changeset
   574
      (if is_some (worker_task ())
061599cb6eb0 refined Future.cancel: explicit future allows to join actual cancellation;
wenzelm
parents: 44298
diff changeset
   575
       then Multithreading.private_interrupts
061599cb6eb0 refined Future.cancel: explicit future allows to join actual cancellation;
wenzelm
parents: 44298
diff changeset
   576
       else Multithreading.public_interrupts)
061599cb6eb0 refined Future.cancel: explicit future allows to join actual cancellation;
wenzelm
parents: 44298
diff changeset
   577
      (fn _ => f x)
061599cb6eb0 refined Future.cancel: explicit future allows to join actual cancellation;
wenzelm
parents: 44298
diff changeset
   578
   else interruptible f x)
061599cb6eb0 refined Future.cancel: explicit future allows to join actual cancellation;
wenzelm
parents: 44298
diff changeset
   579
  before Multithreading.interrupted ();
061599cb6eb0 refined Future.cancel: explicit future allows to join actual cancellation;
wenzelm
parents: 44298
diff changeset
   580
061599cb6eb0 refined Future.cancel: explicit future allows to join actual cancellation;
wenzelm
parents: 44298
diff changeset
   581
34277
7325a5e3587f added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents: 33416
diff changeset
   582
(* promised futures -- fulfilled by external means *)
7325a5e3587f added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents: 33416
diff changeset
   583
44298
b8f8488704e2 Future.promise: explicit abort operation (like uninterruptible future job);
wenzelm
parents: 44295
diff changeset
   584
fun promise_group group abort : 'a future =
34277
7325a5e3587f added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents: 33416
diff changeset
   585
  let
35016
c0f2e49bccfc result: Single_Assignment.var;
wenzelm
parents: 34281
diff changeset
   586
    val result = Single_Assignment.var "promise" : 'a result;
44298
b8f8488704e2 Future.promise: explicit abort operation (like uninterruptible future job);
wenzelm
parents: 44295
diff changeset
   587
    fun assign () = assign_result group result Exn.interrupt_exn
39243
307e3d07d19f Future.promise: more robust treatment of concurrent abort vs. fulfill (amending 047c96f41455);
wenzelm
parents: 39232
diff changeset
   588
      handle Fail _ => true
307e3d07d19f Future.promise: more robust treatment of concurrent abort vs. fulfill (amending 047c96f41455);
wenzelm
parents: 39232
diff changeset
   589
        | exn =>
44298
b8f8488704e2 Future.promise: explicit abort operation (like uninterruptible future job);
wenzelm
parents: 44295
diff changeset
   590
            if Exn.is_interrupt exn
b8f8488704e2 Future.promise: explicit abort operation (like uninterruptible future job);
wenzelm
parents: 44295
diff changeset
   591
            then raise Fail "Concurrent attempt to fulfill promise"
39243
307e3d07d19f Future.promise: more robust treatment of concurrent abort vs. fulfill (amending 047c96f41455);
wenzelm
parents: 39232
diff changeset
   592
            else reraise exn;
44298
b8f8488704e2 Future.promise: explicit abort operation (like uninterruptible future job);
wenzelm
parents: 44295
diff changeset
   593
    fun job () =
b8f8488704e2 Future.promise: explicit abort operation (like uninterruptible future job);
wenzelm
parents: 44295
diff changeset
   594
      Multithreading.with_attributes Multithreading.no_interrupts
b8f8488704e2 Future.promise: explicit abort operation (like uninterruptible future job);
wenzelm
parents: 44295
diff changeset
   595
        (fn _ => assign () before abort ());
37854
047c96f41455 back to more strict dependencies, even for canceled groups (reverting parts of 02936e77a07c);
wenzelm
parents: 37852
diff changeset
   596
    val task = SYNCHRONIZED "enqueue_passive" (fn () =>
44298
b8f8488704e2 Future.promise: explicit abort operation (like uninterruptible future job);
wenzelm
parents: 44295
diff changeset
   597
      Unsynchronized.change_result queue (Task_Queue.enqueue_passive group job));
41683
73dde8006820 maintain Task_Queue.group within Task_Queue.task;
wenzelm
parents: 41681
diff changeset
   598
  in Future {promised = true, task = task, result = result} end;
34277
7325a5e3587f added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents: 33416
diff changeset
   599
44298
b8f8488704e2 Future.promise: explicit abort operation (like uninterruptible future job);
wenzelm
parents: 44295
diff changeset
   600
fun promise abort = promise_group (worker_subgroup ()) abort;
34277
7325a5e3587f added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents: 33416
diff changeset
   601
41683
73dde8006820 maintain Task_Queue.group within Task_Queue.task;
wenzelm
parents: 41681
diff changeset
   602
fun fulfill_result (Future {promised, task, result}) res =
39243
307e3d07d19f Future.promise: more robust treatment of concurrent abort vs. fulfill (amending 047c96f41455);
wenzelm
parents: 39232
diff changeset
   603
  if not promised then raise Fail "Not a promised future"
307e3d07d19f Future.promise: more robust treatment of concurrent abort vs. fulfill (amending 047c96f41455);
wenzelm
parents: 39232
diff changeset
   604
  else
307e3d07d19f Future.promise: more robust treatment of concurrent abort vs. fulfill (amending 047c96f41455);
wenzelm
parents: 39232
diff changeset
   605
    let
41683
73dde8006820 maintain Task_Queue.group within Task_Queue.task;
wenzelm
parents: 41681
diff changeset
   606
      val group = Task_Queue.group_of_task task;
39243
307e3d07d19f Future.promise: more robust treatment of concurrent abort vs. fulfill (amending 047c96f41455);
wenzelm
parents: 39232
diff changeset
   607
      fun job ok = assign_result group result (if ok then res else Exn.interrupt_exn);
307e3d07d19f Future.promise: more robust treatment of concurrent abort vs. fulfill (amending 047c96f41455);
wenzelm
parents: 39232
diff changeset
   608
      val _ =
307e3d07d19f Future.promise: more robust treatment of concurrent abort vs. fulfill (amending 047c96f41455);
wenzelm
parents: 39232
diff changeset
   609
        Multithreading.with_attributes Multithreading.no_interrupts (fn _ =>
307e3d07d19f Future.promise: more robust treatment of concurrent abort vs. fulfill (amending 047c96f41455);
wenzelm
parents: 39232
diff changeset
   610
          let
307e3d07d19f Future.promise: more robust treatment of concurrent abort vs. fulfill (amending 047c96f41455);
wenzelm
parents: 39232
diff changeset
   611
            val still_passive =
307e3d07d19f Future.promise: more robust treatment of concurrent abort vs. fulfill (amending 047c96f41455);
wenzelm
parents: 39232
diff changeset
   612
              SYNCHRONIZED "fulfill_result" (fn () =>
307e3d07d19f Future.promise: more robust treatment of concurrent abort vs. fulfill (amending 047c96f41455);
wenzelm
parents: 39232
diff changeset
   613
                Unsynchronized.change_result queue
307e3d07d19f Future.promise: more robust treatment of concurrent abort vs. fulfill (amending 047c96f41455);
wenzelm
parents: 39232
diff changeset
   614
                  (Task_Queue.dequeue_passive (Thread.self ()) task));
44110
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   615
          in if still_passive then worker_exec (task, [job]) else () end);
41681
b5d7b15166bf Future.join_results: discontinued post-hoc recording of dynamic dependencies;
wenzelm
parents: 41680
diff changeset
   616
      val _ =
41695
afdbec23b92b eliminated slightly odd abstract type Task_Queue.deps;
wenzelm
parents: 41683
diff changeset
   617
        if is_some (Single_Assignment.peek result) then ()
afdbec23b92b eliminated slightly odd abstract type Task_Queue.deps;
wenzelm
parents: 41683
diff changeset
   618
        else worker_waiting [task] (fn () => ignore (Single_Assignment.await result));
39243
307e3d07d19f Future.promise: more robust treatment of concurrent abort vs. fulfill (amending 047c96f41455);
wenzelm
parents: 39232
diff changeset
   619
    in () end;
34277
7325a5e3587f added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents: 33416
diff changeset
   620
43761
e72ba84ae58f tuned signature -- corresponding to Scala version;
wenzelm
parents: 43665
diff changeset
   621
fun fulfill x res = fulfill_result x (Exn.Res res);
34277
7325a5e3587f added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents: 33416
diff changeset
   622
7325a5e3587f added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents: 33416
diff changeset
   623
32228
7622c03141b0 scheduler: shutdown spontaneously (after some delay) if queue is empty;
wenzelm
parents: 32227
diff changeset
   624
(* shutdown *)
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   625
28203
88f18387f1c9 shutdown: global join-and-shutdown operation;
wenzelm
parents: 28202
diff changeset
   626
fun shutdown () =
28276
fbc707811203 shutdown only if Multithreading.available;
wenzelm
parents: 28242
diff changeset
   627
  if Multithreading.available then
fbc707811203 shutdown only if Multithreading.available;
wenzelm
parents: 28242
diff changeset
   628
    SYNCHRONIZED "shutdown" (fn () =>
32228
7622c03141b0 scheduler: shutdown spontaneously (after some delay) if queue is empty;
wenzelm
parents: 32227
diff changeset
   629
     while scheduler_active () do
34279
02936e77a07c tasks of canceled groups are considered "ready" -- enables to purge the queue from tasks depending on unfinished promises (also improves general reactivity);
wenzelm
parents: 34277
diff changeset
   630
      (wait scheduler_event; broadcast_work ()))
28276
fbc707811203 shutdown only if Multithreading.available;
wenzelm
parents: 28242
diff changeset
   631
  else ();
28203
88f18387f1c9 shutdown: global join-and-shutdown operation;
wenzelm
parents: 28202
diff changeset
   632
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   633
38236
d8c7be27e01d explicitly distinguish Output.status (essential feedback) vs. Output.report (useful markup);
wenzelm
parents: 37904
diff changeset
   634
(* status markup *)
37690
b16231572c61 general Future.report -- also for Toplevel.async_state;
wenzelm
parents: 37682
diff changeset
   635
38236
d8c7be27e01d explicitly distinguish Output.status (essential feedback) vs. Output.report (useful markup);
wenzelm
parents: 37904
diff changeset
   636
fun status e =
37690
b16231572c61 general Future.report -- also for Toplevel.async_state;
wenzelm
parents: 37682
diff changeset
   637
  let
40448
f9347a30d1b2 explicitly identify forked/joined tasks;
wenzelm
parents: 40301
diff changeset
   638
    val task_props =
f9347a30d1b2 explicitly identify forked/joined tasks;
wenzelm
parents: 40301
diff changeset
   639
      (case worker_task () of
f9347a30d1b2 explicitly identify forked/joined tasks;
wenzelm
parents: 40301
diff changeset
   640
        NONE => I
f9347a30d1b2 explicitly identify forked/joined tasks;
wenzelm
parents: 40301
diff changeset
   641
      | SOME task => Markup.properties [(Markup.taskN, Task_Queue.str_of_task task)]);
43665
573d1272f36d tuned signature;
wenzelm
parents: 43538
diff changeset
   642
    val _ = Output.status (Markup.markup_only (task_props Markup.forked));
37690
b16231572c61 general Future.report -- also for Toplevel.async_state;
wenzelm
parents: 37682
diff changeset
   643
    val x = e ();  (*sic -- report "joined" only for success*)
43665
573d1272f36d tuned signature;
wenzelm
parents: 43538
diff changeset
   644
    val _ = Output.status (Markup.markup_only (task_props Markup.joined));
37690
b16231572c61 general Future.report -- also for Toplevel.async_state;
wenzelm
parents: 37682
diff changeset
   645
  in x end;
b16231572c61 general Future.report -- also for Toplevel.async_state;
wenzelm
parents: 37682
diff changeset
   646
b16231572c61 general Future.report -- also for Toplevel.async_state;
wenzelm
parents: 37682
diff changeset
   647
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   648
(*final declarations of this structure!*)
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   649
val map = map_future;
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   650
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   651
end;
28972
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
   652
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
   653
type 'a future = 'a Future.future;
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
   654