src/Pure/Concurrent/future.ML
author wenzelm
Fri, 18 Mar 2016 16:26:35 +0100
changeset 62663 bea354f6ff21
parent 62505 9e2a65912111
child 62819 d3ff367a16a0
permissions -rw-r--r--
clarified modules; tuned signature;
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
57350
fc4d65afdf13 more on "Futures";
wenzelm
parents: 56333
diff changeset
     4
Value-oriented parallel execution via futures and promises.
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
     5
*)
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
     6
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
     7
signature FUTURE =
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
     8
sig
44300
349cc426d929 tuned signature -- treat structure Task_Queue as private to implementation;
wenzelm
parents: 44299
diff changeset
     9
  type task = Task_Queue.task
349cc426d929 tuned signature -- treat structure Task_Queue as private to implementation;
wenzelm
parents: 44299
diff changeset
    10
  type group = Task_Queue.group
349cc426d929 tuned signature -- treat structure Task_Queue as private to implementation;
wenzelm
parents: 44299
diff changeset
    11
  val new_group: group option -> group
349cc426d929 tuned signature -- treat structure Task_Queue as private to implementation;
wenzelm
parents: 44299
diff changeset
    12
  val worker_task: unit -> task option
349cc426d929 tuned signature -- treat structure Task_Queue as private to implementation;
wenzelm
parents: 44299
diff changeset
    13
  val worker_group: unit -> group option
52603
a44e9a1d5d8b tuned signature;
wenzelm
parents: 52558
diff changeset
    14
  val the_worker_group: unit -> group
44300
349cc426d929 tuned signature -- treat structure Task_Queue as private to implementation;
wenzelm
parents: 44299
diff changeset
    15
  val worker_subgroup: unit -> group
28972
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
    16
  type 'a future
44300
349cc426d929 tuned signature -- treat structure Task_Queue as private to implementation;
wenzelm
parents: 44299
diff changeset
    17
  val task_of: 'a future -> task
28972
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
    18
  val peek: 'a future -> 'a Exn.result option
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
    19
  val is_finished: 'a future -> bool
50280
0eb9b5d09f31 more uniform ML statistics;
wenzelm
parents: 50255
diff changeset
    20
  val ML_statistics: bool Unsynchronized.ref
44301
65f60d9ac4bf tuned signature (again);
wenzelm
parents: 44300
diff changeset
    21
  val interruptible_task: ('a -> 'b) -> 'a -> 'b
47404
e6e5750f1311 simplified Future.cancel/cancel_group (again) -- running threads only;
wenzelm
parents: 45666
diff changeset
    22
  val cancel_group: group -> unit
e6e5750f1311 simplified Future.cancel/cancel_group (again) -- running threads only;
wenzelm
parents: 45666
diff changeset
    23
  val cancel: 'a future -> unit
56333
38f1422ef473 support bulk messages consisting of small string segments, which are more healthy to the Poly/ML RTS and might prevent spurious GC crashes such as MTGCProcessMarkPointers::ScanAddressesInObject;
wenzelm
parents: 54671
diff changeset
    24
  val error_message: Position.T -> (serial * string) * string option -> unit
50914
fe4714886d92 identify future results more carefully, to avoid odd duplication of error messages, notably from forked goals;
wenzelm
parents: 50911
diff changeset
    25
  val identify_result: Position.T -> 'a Exn.result -> 'a Exn.result
44427
c4a86d72a5cc tuned signature;
wenzelm
parents: 44387
diff changeset
    26
  type params = {name: string, group: group option, deps: task list, pri: int, interrupts: bool}
c4a86d72a5cc tuned signature;
wenzelm
parents: 44387
diff changeset
    27
  val default_params: params
c4a86d72a5cc tuned signature;
wenzelm
parents: 44387
diff changeset
    28
  val forks: params -> (unit -> 'a) list -> 'a future list
32724
aaeeb0ba2035 added fork_deps_pri;
wenzelm
parents: 32644
diff changeset
    29
  val fork: (unit -> 'a) -> 'a future
28972
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
    30
  val join_results: 'a future list -> 'a Exn.result list
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
    31
  val join_result: 'a future -> 'a Exn.result
44330
b28e091f683a added Future.joins convenience;
wenzelm
parents: 44301
diff changeset
    32
  val joins: 'a future list -> 'a list
28972
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
    33
  val join: 'a future -> 'a
54369
7bf7b2903fb9 tuned signature;
wenzelm
parents: 53190
diff changeset
    34
  val join_tasks: task list -> unit
54649
99b9249b3e05 more official task context via Task_Queue.enroll, which is required to participate in group cancellation (e.g. to terminate command exec);
wenzelm
parents: 54637
diff changeset
    35
  val task_context: string -> group -> ('a -> 'b) -> 'a -> 'b
44294
a0ddd5760444 clarified Future.cond_forks: more uniform handling of exceptional situations;
wenzelm
parents: 44268
diff changeset
    36
  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
    37
  val value: 'a -> 'a future
44427
c4a86d72a5cc tuned signature;
wenzelm
parents: 44387
diff changeset
    38
  val cond_forks: params -> (unit -> 'a) list -> 'a future list
28972
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
    39
  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
    40
  val promise_group: group -> (unit -> unit) -> 'a future
44298
b8f8488704e2 Future.promise: explicit abort operation (like uninterruptible future job);
wenzelm
parents: 44295
diff changeset
    41
  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
    42
  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
    43
  val fulfill: 'a future -> 'a -> unit
54369
7bf7b2903fb9 tuned signature;
wenzelm
parents: 53190
diff changeset
    44
  val group_snapshot: group -> task list
28203
88f18387f1c9 shutdown: global join-and-shutdown operation;
wenzelm
parents: 28202
diff changeset
    45
  val shutdown: unit -> unit
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
    46
end;
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
    47
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
    48
structure Future: FUTURE =
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
    49
struct
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
    50
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
    51
(** 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
    52
44300
349cc426d929 tuned signature -- treat structure Task_Queue as private to implementation;
wenzelm
parents: 44299
diff changeset
    53
type task = Task_Queue.task;
349cc426d929 tuned signature -- treat structure Task_Queue as private to implementation;
wenzelm
parents: 44299
diff changeset
    54
type group = Task_Queue.group;
349cc426d929 tuned signature -- treat structure Task_Queue as private to implementation;
wenzelm
parents: 44299
diff changeset
    55
val new_group = Task_Queue.new_group;
349cc426d929 tuned signature -- treat structure Task_Queue as private to implementation;
wenzelm
parents: 44299
diff changeset
    56
349cc426d929 tuned signature -- treat structure Task_Queue as private to implementation;
wenzelm
parents: 44299
diff changeset
    57
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
    58
(* identifiers *)
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
    59
32058
c76fd93b3b99 more abstract Future.is_worker;
wenzelm
parents: 32055
diff changeset
    60
local
44300
349cc426d929 tuned signature -- treat structure Task_Queue as private to implementation;
wenzelm
parents: 44299
diff changeset
    61
  val tag = Universal.tag () : task option Universal.tag;
32058
c76fd93b3b99 more abstract Future.is_worker;
wenzelm
parents: 32055
diff changeset
    62
in
41683
73dde8006820 maintain Task_Queue.group within Task_Queue.task;
wenzelm
parents: 41681
diff changeset
    63
  fun worker_task () = the_default NONE (Thread.getLocal tag);
44110
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
    64
  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
    65
end;
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
    66
41683
73dde8006820 maintain Task_Queue.group within Task_Queue.task;
wenzelm
parents: 41681
diff changeset
    67
val worker_group = Option.map Task_Queue.group_of_task o worker_task;
52603
a44e9a1d5d8b tuned signature;
wenzelm
parents: 52558
diff changeset
    68
a44e9a1d5d8b tuned signature;
wenzelm
parents: 52558
diff changeset
    69
fun the_worker_group () =
a44e9a1d5d8b tuned signature;
wenzelm
parents: 52558
diff changeset
    70
  (case worker_group () of
a44e9a1d5d8b tuned signature;
wenzelm
parents: 52558
diff changeset
    71
    SOME group => group
a44e9a1d5d8b tuned signature;
wenzelm
parents: 52558
diff changeset
    72
  | NONE => raise Fail "Missing worker thread context");
a44e9a1d5d8b tuned signature;
wenzelm
parents: 52558
diff changeset
    73
44300
349cc426d929 tuned signature -- treat structure Task_Queue as private to implementation;
wenzelm
parents: 44299
diff changeset
    74
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
    75
41679
79716cb61bfd refined task timing: joining vs. waiting;
wenzelm
parents: 41674
diff changeset
    76
fun worker_joining e =
79716cb61bfd refined task timing: joining vs. waiting;
wenzelm
parents: 41674
diff changeset
    77
  (case worker_task () of
79716cb61bfd refined task timing: joining vs. waiting;
wenzelm
parents: 41674
diff changeset
    78
    NONE => e ()
79716cb61bfd refined task timing: joining vs. waiting;
wenzelm
parents: 41674
diff changeset
    79
  | SOME task => Task_Queue.joining task e);
79716cb61bfd refined task timing: joining vs. waiting;
wenzelm
parents: 41674
diff changeset
    80
41680
a4c822915eaa more informative task timing: some dependency tracking;
wenzelm
parents: 41679
diff changeset
    81
fun worker_waiting deps e =
41670
74010c6af0a4 added basic task timing;
wenzelm
parents: 40448
diff changeset
    82
  (case worker_task () of
74010c6af0a4 added basic task timing;
wenzelm
parents: 40448
diff changeset
    83
    NONE => e ()
41680
a4c822915eaa more informative task timing: some dependency tracking;
wenzelm
parents: 41679
diff changeset
    84
  | SOME task => Task_Queue.waiting task deps e);
41670
74010c6af0a4 added basic task timing;
wenzelm
parents: 40448
diff changeset
    85
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
    86
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
    87
(* datatype future *)
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
    88
35016
c0f2e49bccfc result: Single_Assignment.var;
wenzelm
parents: 34281
diff changeset
    89
type 'a result = 'a Exn.result Single_Assignment.var;
c0f2e49bccfc result: Single_Assignment.var;
wenzelm
parents: 34281
diff changeset
    90
28972
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
    91
datatype 'a future = Future of
34277
7325a5e3587f added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents: 33416
diff changeset
    92
 {promised: bool,
44300
349cc426d929 tuned signature -- treat structure Task_Queue as private to implementation;
wenzelm
parents: 44299
diff changeset
    93
  task: task,
35016
c0f2e49bccfc result: Single_Assignment.var;
wenzelm
parents: 34281
diff changeset
    94
  result: 'a result};
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
    95
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
    96
fun task_of (Future {task, ...}) = task;
32253
d9def420c84e future result: Synchronized.var;
wenzelm
parents: 32249
diff changeset
    97
fun result_of (Future {result, ...}) = result;
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
    98
35016
c0f2e49bccfc result: Single_Assignment.var;
wenzelm
parents: 34281
diff changeset
    99
fun peek x = Single_Assignment.peek (result_of x);
28558
2a6297b4273c replaced str_of by general peek;
wenzelm
parents: 28548
diff changeset
   100
fun is_finished x = is_some (peek x);
28320
c6aef67f964d added is_finished;
wenzelm
parents: 28304
diff changeset
   101
62663
bea354f6ff21 clarified modules;
wenzelm
parents: 62505
diff changeset
   102
val _ =
bea354f6ff21 clarified modules;
wenzelm
parents: 62505
diff changeset
   103
  PolyML.addPrettyPrinter (fn depth => fn pretty => fn x =>
bea354f6ff21 clarified modules;
wenzelm
parents: 62505
diff changeset
   104
    (case peek x of
bea354f6ff21 clarified modules;
wenzelm
parents: 62505
diff changeset
   105
      NONE => PolyML.PrettyString "<future>"
bea354f6ff21 clarified modules;
wenzelm
parents: 62505
diff changeset
   106
    | SOME (Exn.Exn _) => PolyML.PrettyString "<failed>"
bea354f6ff21 clarified modules;
wenzelm
parents: 62505
diff changeset
   107
    | SOME (Exn.Res y) => pretty (y, depth)));
bea354f6ff21 clarified modules;
wenzelm
parents: 62505
diff changeset
   108
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
   109
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
   110
8c0335bc9336 inherit group from running thread, or create a new one -- make it harder to re-use canceled groups;
wenzelm
parents: 28170
diff changeset
   111
(** 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
   112
8c0335bc9336 inherit group from running thread, or create a new one -- make it harder to re-use canceled groups;
wenzelm
parents: 28170
diff changeset
   113
(* synchronization *)
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   114
32219
9a2566d1fdbd more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents: 32186
diff changeset
   115
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
   116
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
   117
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
   118
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   119
local
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   120
  val lock = Mutex.mutex ();
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   121
in
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   122
59054
61b723761dff load simple_thread.ML later, such that it benefits from redefined print_exception_trace;
wenzelm
parents: 57350
diff changeset
   123
fun SYNCHRONIZED name = Multithreading.synchronized name lock;
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   124
32219
9a2566d1fdbd more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents: 32186
diff changeset
   125
fun wait cond = (*requires SYNCHRONIZED*)
32295
400cc493d466 renamed Multithreading.regular_interrupts to Multithreading.public_interrupts;
wenzelm
parents: 32293
diff changeset
   126
  Multithreading.sync_wait NONE NONE cond lock;
28206
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   127
32295
400cc493d466 renamed Multithreading.regular_interrupts to Multithreading.public_interrupts;
wenzelm
parents: 32293
diff changeset
   128
fun wait_timeout timeout cond = (*requires SYNCHRONIZED*)
400cc493d466 renamed Multithreading.regular_interrupts to Multithreading.public_interrupts;
wenzelm
parents: 32293
diff changeset
   129
  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
   130
32219
9a2566d1fdbd more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents: 32186
diff changeset
   131
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
   132
  ConditionVar.signal cond;
9a2566d1fdbd more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents: 32186
diff changeset
   133
9a2566d1fdbd more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents: 32186
diff changeset
   134
fun broadcast cond = (*requires SYNCHRONIZED*)
28166
43087721a66e moved task, thread_data, group, queue to task_queue.ML;
wenzelm
parents: 28163
diff changeset
   135
  ConditionVar.broadcast cond;
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   136
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   137
end;
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   138
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   139
33410
e351f4c1f18c worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents: 33409
diff changeset
   140
(* global state *)
e351f4c1f18c worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents: 33409
diff changeset
   141
e351f4c1f18c worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents: 33409
diff changeset
   142
val queue = Unsynchronized.ref Task_Queue.empty;
e351f4c1f18c worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents: 33409
diff changeset
   143
val next = Unsynchronized.ref 0;
e351f4c1f18c worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents: 33409
diff changeset
   144
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
   145
val canceled = Unsynchronized.ref ([]: group list);
33410
e351f4c1f18c worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents: 33409
diff changeset
   146
val do_shutdown = Unsynchronized.ref false;
e351f4c1f18c worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents: 33409
diff changeset
   147
val max_workers = Unsynchronized.ref 0;
e351f4c1f18c worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents: 33409
diff changeset
   148
val max_active = Unsynchronized.ref 0;
e351f4c1f18c worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents: 33409
diff changeset
   149
50280
0eb9b5d09f31 more uniform ML statistics;
wenzelm
parents: 50255
diff changeset
   150
val status_ticks = Unsynchronized.ref 0;
0eb9b5d09f31 more uniform ML statistics;
wenzelm
parents: 50255
diff changeset
   151
val last_round = Unsynchronized.ref Time.zeroTime;
0eb9b5d09f31 more uniform ML statistics;
wenzelm
parents: 50255
diff changeset
   152
val next_round = seconds 0.05;
0eb9b5d09f31 more uniform ML statistics;
wenzelm
parents: 50255
diff changeset
   153
33410
e351f4c1f18c worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents: 33409
diff changeset
   154
datatype worker_state = Working | Waiting | Sleeping;
e351f4c1f18c worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents: 33409
diff changeset
   155
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
   156
e351f4c1f18c worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents: 33409
diff changeset
   157
fun count_workers state = (*requires SYNCHRONIZED*)
e351f4c1f18c worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents: 33409
diff changeset
   158
  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
   159
e351f4c1f18c worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents: 33409
diff changeset
   160
50280
0eb9b5d09f31 more uniform ML statistics;
wenzelm
parents: 50255
diff changeset
   161
0eb9b5d09f31 more uniform ML statistics;
wenzelm
parents: 50255
diff changeset
   162
(* status *)
0eb9b5d09f31 more uniform ML statistics;
wenzelm
parents: 50255
diff changeset
   163
0eb9b5d09f31 more uniform ML statistics;
wenzelm
parents: 50255
diff changeset
   164
val ML_statistics = Unsynchronized.ref false;
0eb9b5d09f31 more uniform ML statistics;
wenzelm
parents: 50255
diff changeset
   165
0eb9b5d09f31 more uniform ML statistics;
wenzelm
parents: 50255
diff changeset
   166
fun report_status () = (*requires SYNCHRONIZED*)
0eb9b5d09f31 more uniform ML statistics;
wenzelm
parents: 50255
diff changeset
   167
  if ! ML_statistics then
0eb9b5d09f31 more uniform ML statistics;
wenzelm
parents: 50255
diff changeset
   168
    let
60610
f52b4b0c10c4 improved scheduling for urgent tasks, using farm of replacement threads (may lead to factor 2 overloading, but CPUs are usually hyperthreaded);
wenzelm
parents: 59468
diff changeset
   169
      val {ready, pending, running, passive, urgent} = Task_Queue.status (! queue);
50280
0eb9b5d09f31 more uniform ML statistics;
wenzelm
parents: 50255
diff changeset
   170
      val total = length (! workers);
0eb9b5d09f31 more uniform ML statistics;
wenzelm
parents: 50255
diff changeset
   171
      val active = count_workers Working;
0eb9b5d09f31 more uniform ML statistics;
wenzelm
parents: 50255
diff changeset
   172
      val waiting = count_workers Waiting;
0eb9b5d09f31 more uniform ML statistics;
wenzelm
parents: 50255
diff changeset
   173
      val stats =
51990
cc66addbba6d more uniform Markup.print_real;
wenzelm
parents: 51661
diff changeset
   174
       [("now", Markup.print_real (Time.toReal (Time.now ()))),
50280
0eb9b5d09f31 more uniform ML statistics;
wenzelm
parents: 50255
diff changeset
   175
        ("tasks_ready", Markup.print_int ready),
0eb9b5d09f31 more uniform ML statistics;
wenzelm
parents: 50255
diff changeset
   176
        ("tasks_pending", Markup.print_int pending),
0eb9b5d09f31 more uniform ML statistics;
wenzelm
parents: 50255
diff changeset
   177
        ("tasks_running", Markup.print_int running),
0eb9b5d09f31 more uniform ML statistics;
wenzelm
parents: 50255
diff changeset
   178
        ("tasks_passive", Markup.print_int passive),
60610
f52b4b0c10c4 improved scheduling for urgent tasks, using farm of replacement threads (may lead to factor 2 overloading, but CPUs are usually hyperthreaded);
wenzelm
parents: 59468
diff changeset
   179
        ("tasks_urgent", Markup.print_int urgent),
50280
0eb9b5d09f31 more uniform ML statistics;
wenzelm
parents: 50255
diff changeset
   180
        ("workers_total", Markup.print_int total),
0eb9b5d09f31 more uniform ML statistics;
wenzelm
parents: 50255
diff changeset
   181
        ("workers_active", Markup.print_int active),
0eb9b5d09f31 more uniform ML statistics;
wenzelm
parents: 50255
diff changeset
   182
        ("workers_waiting", Markup.print_int waiting)] @
0eb9b5d09f31 more uniform ML statistics;
wenzelm
parents: 50255
diff changeset
   183
        ML_Statistics.get ();
56333
38f1422ef473 support bulk messages consisting of small string segments, which are more healthy to the Poly/ML RTS and might prevent spurious GC crashes such as MTGCProcessMarkPointers::ScanAddressesInObject;
wenzelm
parents: 54671
diff changeset
   184
    in Output.try_protocol_message (Markup.ML_statistics :: stats) [] end
50280
0eb9b5d09f31 more uniform ML statistics;
wenzelm
parents: 50255
diff changeset
   185
  else ();
0eb9b5d09f31 more uniform ML statistics;
wenzelm
parents: 50255
diff changeset
   186
0eb9b5d09f31 more uniform ML statistics;
wenzelm
parents: 50255
diff changeset
   187
44110
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   188
(* cancellation primitives *)
32099
5382c93108db propagate exceptions within future groups;
wenzelm
parents: 32096
diff changeset
   189
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
   190
fun cancel_now group = (*requires SYNCHRONIZED*)
44341
a93d25fb14fc purely functional task_queue.ML -- moved actual interrupt_unsynchronized to future.ML;
wenzelm
parents: 44330
diff changeset
   191
  let
47404
e6e5750f1311 simplified Future.cancel/cancel_group (again) -- running threads only;
wenzelm
parents: 45666
diff changeset
   192
    val running = Task_Queue.cancel (! queue) group;
49894
69bfd86cc711 more robust cancel_now: avoid shooting yourself in the foot;
wenzelm
parents: 49009
diff changeset
   193
    val _ = running |> List.app (fn thread =>
61556
0d4ee4168e41 clarified modules;
wenzelm
parents: 61077
diff changeset
   194
      if Standard_Thread.is_self thread then ()
0d4ee4168e41 clarified modules;
wenzelm
parents: 61077
diff changeset
   195
      else Standard_Thread.interrupt_unsynchronized thread);
47404
e6e5750f1311 simplified Future.cancel/cancel_group (again) -- running threads only;
wenzelm
parents: 45666
diff changeset
   196
  in running end;
44341
a93d25fb14fc purely functional task_queue.ML -- moved actual interrupt_unsynchronized to future.ML;
wenzelm
parents: 44330
diff changeset
   197
a93d25fb14fc purely functional task_queue.ML -- moved actual interrupt_unsynchronized to future.ML;
wenzelm
parents: 44330
diff changeset
   198
fun cancel_all () = (*requires SYNCHRONIZED*)
a93d25fb14fc purely functional task_queue.ML -- moved actual interrupt_unsynchronized to future.ML;
wenzelm
parents: 44330
diff changeset
   199
  let
a93d25fb14fc purely functional task_queue.ML -- moved actual interrupt_unsynchronized to future.ML;
wenzelm
parents: 44330
diff changeset
   200
    val (groups, threads) = Task_Queue.cancel_all (! queue);
61556
0d4ee4168e41 clarified modules;
wenzelm
parents: 61077
diff changeset
   201
    val _ = List.app Standard_Thread.interrupt_unsynchronized threads;
44341
a93d25fb14fc purely functional task_queue.ML -- moved actual interrupt_unsynchronized to future.ML;
wenzelm
parents: 44330
diff changeset
   202
  in groups end;
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
   203
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
   204
fun cancel_later group = (*requires SYNCHRONIZED*)
32738
15bb09ca0378 explicit indication of Unsynchronized.ref;
wenzelm
parents: 32724
diff changeset
   205
 (Unsynchronized.change canceled (insert Task_Queue.eq_group group);
15bb09ca0378 explicit indication of Unsynchronized.ref;
wenzelm
parents: 32724
diff changeset
   206
  broadcast scheduler_event);
29341
6bb007a0f9f2 more reactive scheduler: reduced loop timeout, propagate broadcast interrupt via TaskQueue.cancel_all;
wenzelm
parents: 29119
diff changeset
   207
44301
65f60d9ac4bf tuned signature (again);
wenzelm
parents: 44300
diff changeset
   208
fun interruptible_task f x =
62359
6709e51d5c11 unconditional Multithreading;
wenzelm
parents: 61556
diff changeset
   209
  Multithreading.with_attributes
6709e51d5c11 unconditional Multithreading;
wenzelm
parents: 61556
diff changeset
   210
    (if is_some (worker_task ())
6709e51d5c11 unconditional Multithreading;
wenzelm
parents: 61556
diff changeset
   211
     then Multithreading.private_interrupts
6709e51d5c11 unconditional Multithreading;
wenzelm
parents: 61556
diff changeset
   212
     else Multithreading.public_interrupts)
6709e51d5c11 unconditional Multithreading;
wenzelm
parents: 61556
diff changeset
   213
    (fn _ => f x)
44301
65f60d9ac4bf tuned signature (again);
wenzelm
parents: 44300
diff changeset
   214
  before Multithreading.interrupted ();
65f60d9ac4bf tuned signature (again);
wenzelm
parents: 44300
diff changeset
   215
65f60d9ac4bf tuned signature (again);
wenzelm
parents: 44300
diff changeset
   216
44110
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   217
(* worker threads *)
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   218
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   219
fun worker_exec (task, jobs) =
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
   220
  let
41683
73dde8006820 maintain Task_Queue.group within Task_Queue.task;
wenzelm
parents: 41681
diff changeset
   221
    val group = Task_Queue.group_of_task task;
32102
81d03a29980c added worker_group;
wenzelm
parents: 32099
diff changeset
   222
    val valid = not (Task_Queue.is_canceled group);
41670
74010c6af0a4 added basic task timing;
wenzelm
parents: 40448
diff changeset
   223
    val ok =
74010c6af0a4 added basic task timing;
wenzelm
parents: 40448
diff changeset
   224
      Task_Queue.running task (fn () =>
41683
73dde8006820 maintain Task_Queue.group within Task_Queue.task;
wenzelm
parents: 41681
diff changeset
   225
        setmp_worker_task task (fn () =>
41670
74010c6af0a4 added basic task timing;
wenzelm
parents: 40448
diff changeset
   226
          fold (fn job => fn ok => job valid andalso ok) jobs true) ());
50975
73ec6ad6700e more systematic task statistics;
wenzelm
parents: 50974
diff changeset
   227
    val _ =
73ec6ad6700e more systematic task statistics;
wenzelm
parents: 50974
diff changeset
   228
      if ! Multithreading.trace >= 2 then
56333
38f1422ef473 support bulk messages consisting of small string segments, which are more healthy to the Poly/ML RTS and might prevent spurious GC crashes such as MTGCProcessMarkPointers::ScanAddressesInObject;
wenzelm
parents: 54671
diff changeset
   229
        Output.try_protocol_message (Markup.task_statistics :: Task_Queue.task_statistics task) []
50975
73ec6ad6700e more systematic task statistics;
wenzelm
parents: 50974
diff changeset
   230
      else ();
32246
d4f7934e9555 misc tuning;
wenzelm
parents: 32230
diff changeset
   231
    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
   232
      let
32738
15bb09ca0378 explicit indication of Unsynchronized.ref;
wenzelm
parents: 32724
diff changeset
   233
        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
   234
        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
   235
        val _ =
44295
e43f0ea90c9a more focused use of Multithreading.interrupted: retain interrupts within task group boundary, without loss of information;
wenzelm
parents: 44294
diff changeset
   236
          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
   237
          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
   238
          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
   239
        val _ = broadcast work_finished;
33413
cb409680dda8 avoid broadcast work_available, use daisy-chained signal instead;
wenzelm
parents: 33411
diff changeset
   240
        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
   241
      in () end);
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
   242
  in () end;
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
   243
59465
c21b65a6834b clarified worker_wait;
wenzelm
parents: 59340
diff changeset
   244
fun worker_wait worker_state cond = (*requires SYNCHRONIZED*)
52558
271663ddf289 allow worker guest threads, which participate actively in future joins, but are outside thread accounting;
wenzelm
parents: 51990
diff changeset
   245
  (case AList.lookup Thread.equal (! workers) (Thread.self ()) of
59465
c21b65a6834b clarified worker_wait;
wenzelm
parents: 59340
diff changeset
   246
    SOME state => Unsynchronized.setmp state worker_state wait cond
c21b65a6834b clarified worker_wait;
wenzelm
parents: 59340
diff changeset
   247
  | NONE => wait cond);
28162
55772e4e95e0 tuned check_cache;
wenzelm
parents: 28156
diff changeset
   248
33415
352fe8e9162d worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents: 33413
diff changeset
   249
fun worker_next () = (*requires SYNCHRONIZED*)
33406
1ddcb8472bd2 slightly leaner and more direct control of worker activity etc.;
wenzelm
parents: 33061
diff changeset
   250
  if length (! workers) > ! max_workers then
1ddcb8472bd2 slightly leaner and more direct control of worker activity etc.;
wenzelm
parents: 33061
diff changeset
   251
    (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
   252
     signal work_available;
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
   253
     NONE)
28166
43087721a66e moved task, thread_data, group, queue to task_queue.ML;
wenzelm
parents: 28163
diff changeset
   254
  else
60610
f52b4b0c10c4 improved scheduling for urgent tasks, using farm of replacement threads (may lead to factor 2 overloading, but CPUs are usually hyperthreaded);
wenzelm
parents: 59468
diff changeset
   255
    let val urgent_only = count_workers Working > ! max_active in
f52b4b0c10c4 improved scheduling for urgent tasks, using farm of replacement threads (may lead to factor 2 overloading, but CPUs are usually hyperthreaded);
wenzelm
parents: 59468
diff changeset
   256
      (case Unsynchronized.change_result queue (Task_Queue.dequeue (Thread.self ()) urgent_only) of
f52b4b0c10c4 improved scheduling for urgent tasks, using farm of replacement threads (may lead to factor 2 overloading, but CPUs are usually hyperthreaded);
wenzelm
parents: 59468
diff changeset
   257
        NONE => (worker_wait Sleeping work_available; worker_next ())
f52b4b0c10c4 improved scheduling for urgent tasks, using farm of replacement threads (may lead to factor 2 overloading, but CPUs are usually hyperthreaded);
wenzelm
parents: 59468
diff changeset
   258
      | some => (signal work_available; some))
f52b4b0c10c4 improved scheduling for urgent tasks, using farm of replacement threads (may lead to factor 2 overloading, but CPUs are usually hyperthreaded);
wenzelm
parents: 59468
diff changeset
   259
    end;
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   260
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
   261
fun worker_loop name =
33415
352fe8e9162d worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents: 33413
diff changeset
   262
  (case SYNCHRONIZED name (fn () => worker_next ()) of
29119
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
   263
    NONE => ()
44295
e43f0ea90c9a more focused use of Multithreading.interrupted: retain interrupts within task group boundary, without loss of information;
wenzelm
parents: 44294
diff changeset
   264
  | SOME work => (worker_exec work; worker_loop name));
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   265
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
   266
fun worker_start name = (*requires SYNCHRONIZED*)
59468
fe6651760643 explicit threads_stack_limit (for recent Poly/ML SVN versions), which leads to soft interrupt instead of exhaustion of virtual memory, which is particularly relevant for the bigger address space of x86_64;
wenzelm
parents: 59467
diff changeset
   267
  let
fe6651760643 explicit threads_stack_limit (for recent Poly/ML SVN versions), which leads to soft interrupt instead of exhaustion of virtual memory, which is particularly relevant for the bigger address space of x86_64;
wenzelm
parents: 59467
diff changeset
   268
    val threads_stack_limit =
fe6651760643 explicit threads_stack_limit (for recent Poly/ML SVN versions), which leads to soft interrupt instead of exhaustion of virtual memory, which is particularly relevant for the bigger address space of x86_64;
wenzelm
parents: 59467
diff changeset
   269
      Real.floor (Options.default_real "threads_stack_limit" * 1024.0 * 1024.0 * 1024.0);
fe6651760643 explicit threads_stack_limit (for recent Poly/ML SVN versions), which leads to soft interrupt instead of exhaustion of virtual memory, which is particularly relevant for the bigger address space of x86_64;
wenzelm
parents: 59467
diff changeset
   270
    val stack_limit = if threads_stack_limit <= 0 then NONE else SOME threads_stack_limit;
fe6651760643 explicit threads_stack_limit (for recent Poly/ML SVN versions), which leads to soft interrupt instead of exhaustion of virtual memory, which is particularly relevant for the bigger address space of x86_64;
wenzelm
parents: 59467
diff changeset
   271
    val worker =
61556
0d4ee4168e41 clarified modules;
wenzelm
parents: 61077
diff changeset
   272
      Standard_Thread.fork {name = "worker", stack_limit = stack_limit, interrupts = false}
59468
fe6651760643 explicit threads_stack_limit (for recent Poly/ML SVN versions), which leads to soft interrupt instead of exhaustion of virtual memory, which is particularly relevant for the bigger address space of x86_64;
wenzelm
parents: 59467
diff changeset
   273
        (fn () => worker_loop name);
fe6651760643 explicit threads_stack_limit (for recent Poly/ML SVN versions), which leads to soft interrupt instead of exhaustion of virtual memory, which is particularly relevant for the bigger address space of x86_64;
wenzelm
parents: 59467
diff changeset
   274
  in Unsynchronized.change workers (cons (worker, Unsynchronized.ref Working)) end
59338
2ea1bf517842 discontinued worker_trend: prefer constant number of active + reserve threads;
wenzelm
parents: 59330
diff changeset
   275
  handle Fail msg => Multithreading.tracing 0 (fn () => "SCHEDULER: " ^ msg);
59330
cb3a4caf206d permissive worker_start: failure to fork thread is deferred to later attempt to provide missing threads, without crashing scheduler;
wenzelm
parents: 59180
diff changeset
   276
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   277
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   278
(* scheduler *)
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   279
28206
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   280
fun scheduler_next () = (*requires SYNCHRONIZED*)
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   281
  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
   282
    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
   283
    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
   284
    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
   285
33415
352fe8e9162d worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents: 33413
diff changeset
   286
50280
0eb9b5d09f31 more uniform ML statistics;
wenzelm
parents: 50255
diff changeset
   287
    (* runtime status *)
33415
352fe8e9162d worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents: 33413
diff changeset
   288
32226
23511e4da055 tuned tracing;
wenzelm
parents: 32225
diff changeset
   289
    val _ =
51046
26a0984191b3 report status more frequently on demand;
wenzelm
parents: 50983
diff changeset
   290
      if tick then Unsynchronized.change status_ticks (fn i => i + 1) else ();
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
    val _ =
51046
26a0984191b3 report status more frequently on demand;
wenzelm
parents: 50983
diff changeset
   292
      if tick andalso ! status_ticks mod (if ! Multithreading.trace >= 1 then 2 else 10) = 0
26a0984191b3 report status more frequently on demand;
wenzelm
parents: 50983
diff changeset
   293
      then report_status () else ();
32053
257eac3946e9 scheduler: tuned tracing (status);
wenzelm
parents: 31631
diff changeset
   294
28191
9e5f556409c6 future: allow explicit group;
wenzelm
parents: 28186
diff changeset
   295
    val _ =
59338
2ea1bf517842 discontinued worker_trend: prefer constant number of active + reserve threads;
wenzelm
parents: 59330
diff changeset
   296
      if not tick orelse forall (Thread.isActive o #1) (! workers) then ()
32095
ad4be204fdfe tuned tracing;
wenzelm
parents: 32058
diff changeset
   297
      else
33409
wenzelm
parents: 33408
diff changeset
   298
        let
37682
ad5489a6be48 tuned white space;
wenzelm
parents: 37216
diff changeset
   299
          val (alive, dead) = List.partition (Thread.isActive o #1) (! workers);
33409
wenzelm
parents: 33408
diff changeset
   300
          val _ = workers := alive;
wenzelm
parents: 33408
diff changeset
   301
        in
wenzelm
parents: 33408
diff changeset
   302
          Multithreading.tracing 0 (fn () =>
51279
f4a2fa9286e9 tuned messages;
wenzelm
parents: 51046
diff changeset
   303
            "SCHEDULER: disposed " ^ string_of_int (length dead) ^ " dead worker threads")
33409
wenzelm
parents: 33408
diff changeset
   304
        end;
28191
9e5f556409c6 future: allow explicit group;
wenzelm
parents: 28186
diff changeset
   305
33415
352fe8e9162d worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents: 33413
diff changeset
   306
352fe8e9162d worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents: 33413
diff changeset
   307
    (* worker pool adjustments *)
352fe8e9162d worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents: 33413
diff changeset
   308
352fe8e9162d worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents: 33413
diff changeset
   309
    val max_active0 = ! max_active;
352fe8e9162d worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents: 33413
diff changeset
   310
    val max_workers0 = ! max_workers;
352fe8e9162d worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents: 33413
diff changeset
   311
59340
734bb148503e explicit shutdown of scheduler and worker thread farm, assuming Session.shutdown() before saving heap image;
wenzelm
parents: 59338
diff changeset
   312
    val m =
734bb148503e explicit shutdown of scheduler and worker thread farm, assuming Session.shutdown() before saving heap image;
wenzelm
parents: 59338
diff changeset
   313
      if ! do_shutdown andalso Task_Queue.all_passive (! queue) then 0
734bb148503e explicit shutdown of scheduler and worker thread farm, assuming Session.shutdown() before saving heap image;
wenzelm
parents: 59338
diff changeset
   314
      else Multithreading.max_threads_value ();
33406
1ddcb8472bd2 slightly leaner and more direct control of worker activity etc.;
wenzelm
parents: 33061
diff changeset
   315
    val _ = max_active := m;
59338
2ea1bf517842 discontinued worker_trend: prefer constant number of active + reserve threads;
wenzelm
parents: 59330
diff changeset
   316
    val _ = max_workers := 2 * m;
33406
1ddcb8472bd2 slightly leaner and more direct control of worker activity etc.;
wenzelm
parents: 33061
diff changeset
   317
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
   318
    val missing = ! max_workers - length (! workers);
28203
88f18387f1c9 shutdown: global join-and-shutdown operation;
wenzelm
parents: 28202
diff changeset
   319
    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
   320
      if missing > 0 then
33415
352fe8e9162d worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents: 33413
diff changeset
   321
        funpow missing (fn () =>
352fe8e9162d worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents: 33413
diff changeset
   322
          ignore (worker_start ("worker " ^ string_of_int (Unsynchronized.inc next)))) ()
28203
88f18387f1c9 shutdown: global join-and-shutdown operation;
wenzelm
parents: 28202
diff changeset
   323
      else ();
28206
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   324
33415
352fe8e9162d worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents: 33413
diff changeset
   325
    val _ =
352fe8e9162d worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents: 33413
diff changeset
   326
      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
   327
      else signal work_available;
352fe8e9162d worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents: 33413
diff changeset
   328
352fe8e9162d worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents: 33413
diff changeset
   329
352fe8e9162d worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents: 33413
diff changeset
   330
    (* canceled groups *)
352fe8e9162d worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents: 33413
diff changeset
   331
32225
d5d6f47fb018 cancel: improved reactivity due to more careful broadcasting;
wenzelm
parents: 32224
diff changeset
   332
    val _ =
d5d6f47fb018 cancel: improved reactivity due to more careful broadcasting;
wenzelm
parents: 32224
diff changeset
   333
      if null (! canceled) then ()
32293
e0b8da3fae4d tuned tracing;
wenzelm
parents: 32286
diff changeset
   334
      else
e0b8da3fae4d tuned tracing;
wenzelm
parents: 32286
diff changeset
   335
       (Multithreading.tracing 1 (fn () =>
e0b8da3fae4d tuned tracing;
wenzelm
parents: 32286
diff changeset
   336
          string_of_int (length (! canceled)) ^ " canceled groups");
44299
061599cb6eb0 refined Future.cancel: explicit future allows to join actual cancellation;
wenzelm
parents: 44298
diff changeset
   337
        Unsynchronized.change canceled (filter_out (null o cancel_now));
51281
c05f7e1dd602 signal work_available should be sufficient to initiate daisy-chained workers, and lead to separate broadcast work_finished eventually -- NB: broadcasting all worker threads tends to burn parallel CPU cycles;
wenzelm
parents: 51280
diff changeset
   338
        signal work_available);
28206
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   339
33415
352fe8e9162d worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents: 33413
diff changeset
   340
352fe8e9162d worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents: 33413
diff changeset
   341
    (* delay loop *)
352fe8e9162d worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents: 33413
diff changeset
   342
32295
400cc493d466 renamed Multithreading.regular_interrupts to Multithreading.public_interrupts;
wenzelm
parents: 32293
diff changeset
   343
    val _ = Exn.release (wait_timeout next_round scheduler_event);
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
   344
33415
352fe8e9162d worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents: 33413
diff changeset
   345
352fe8e9162d worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents: 33413
diff changeset
   346
    (* shutdown *)
352fe8e9162d worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents: 33413
diff changeset
   347
32219
9a2566d1fdbd more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents: 32186
diff changeset
   348
    val continue = not (! do_shutdown andalso null (! workers));
50429
f8cd5e53653b final report_status within SYNCHRONIZED part of scheduler loop: required for sanity of data;
wenzelm
parents: 50280
diff changeset
   349
    val _ = if continue then () else (report_status (); scheduler := NONE);
33415
352fe8e9162d worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents: 33413
diff changeset
   350
32219
9a2566d1fdbd more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents: 32186
diff changeset
   351
    val _ = broadcast scheduler_event;
32295
400cc493d466 renamed Multithreading.regular_interrupts to Multithreading.public_interrupts;
wenzelm
parents: 32293
diff changeset
   352
  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
   353
  handle exn =>
69c6d3e87660 more abstract treatment of interrupts in structure Exn -- hardly ever need to mention Interrupt literally;
wenzelm
parents: 38236
diff changeset
   354
    if Exn.is_interrupt exn then
51279
f4a2fa9286e9 tuned messages;
wenzelm
parents: 51046
diff changeset
   355
     (Multithreading.tracing 1 (fn () => "SCHEDULER: Interrupt");
44341
a93d25fb14fc purely functional task_queue.ML -- moved actual interrupt_unsynchronized to future.ML;
wenzelm
parents: 44330
diff changeset
   356
      List.app cancel_later (cancel_all ());
51281
c05f7e1dd602 signal work_available should be sufficient to initiate daisy-chained workers, and lead to separate broadcast work_finished eventually -- NB: broadcasting all worker threads tends to burn parallel CPU cycles;
wenzelm
parents: 51280
diff changeset
   357
      signal work_available; true)
62505
9e2a65912111 clarified modules;
wenzelm
parents: 62359
diff changeset
   358
    else Exn.reraise exn;
32295
400cc493d466 renamed Multithreading.regular_interrupts to Multithreading.public_interrupts;
wenzelm
parents: 32293
diff changeset
   359
28206
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   360
fun scheduler_loop () =
44173
aaaa13e297dc immediate fork of initial workers -- avoid 5 ticks (250ms) for adaptive scheme (a07558eb5029);
wenzelm
parents: 44115
diff changeset
   361
 (while
33416
13d00799fe49 scheduler: clarified interrupt attributes and handling;
wenzelm
parents: 33415
diff changeset
   362
    Multithreading.with_attributes
13d00799fe49 scheduler: clarified interrupt attributes and handling;
wenzelm
parents: 33415
diff changeset
   363
      (Multithreading.sync_interrupts Multithreading.public_interrupts)
13d00799fe49 scheduler: clarified interrupt attributes and handling;
wenzelm
parents: 33415
diff changeset
   364
      (fn _ => SYNCHRONIZED "scheduler" (fn () => scheduler_next ()))
50429
f8cd5e53653b final report_status within SYNCHRONIZED part of scheduler loop: required for sanity of data;
wenzelm
parents: 50280
diff changeset
   365
  do (); last_round := Time.zeroTime);
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   366
28203
88f18387f1c9 shutdown: global join-and-shutdown operation;
wenzelm
parents: 28202
diff changeset
   367
fun scheduler_active () = (*requires SYNCHRONIZED*)
88f18387f1c9 shutdown: global join-and-shutdown operation;
wenzelm
parents: 28202
diff changeset
   368
  (case ! scheduler of NONE => false | SOME thread => Thread.isActive thread);
88f18387f1c9 shutdown: global join-and-shutdown operation;
wenzelm
parents: 28202
diff changeset
   369
32228
7622c03141b0 scheduler: shutdown spontaneously (after some delay) if queue is empty;
wenzelm
parents: 32227
diff changeset
   370
fun scheduler_check () = (*requires SYNCHRONIZED*)
7622c03141b0 scheduler: shutdown spontaneously (after some delay) if queue is empty;
wenzelm
parents: 32227
diff changeset
   371
 (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
   372
  if scheduler_active () then ()
59468
fe6651760643 explicit threads_stack_limit (for recent Poly/ML SVN versions), which leads to soft interrupt instead of exhaustion of virtual memory, which is particularly relevant for the bigger address space of x86_64;
wenzelm
parents: 59467
diff changeset
   373
  else
60764
b610ba36e02c more explicit thread identification;
wenzelm
parents: 60610
diff changeset
   374
    scheduler :=
61556
0d4ee4168e41 clarified modules;
wenzelm
parents: 61077
diff changeset
   375
      SOME (Standard_Thread.fork {name = "scheduler", stack_limit = NONE, interrupts = false}
60764
b610ba36e02c more explicit thread identification;
wenzelm
parents: 60610
diff changeset
   376
        scheduler_loop));
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   377
44301
65f60d9ac4bf tuned signature (again);
wenzelm
parents: 44300
diff changeset
   378
65f60d9ac4bf tuned signature (again);
wenzelm
parents: 44300
diff changeset
   379
65f60d9ac4bf tuned signature (again);
wenzelm
parents: 44300
diff changeset
   380
(** futures **)
65f60d9ac4bf tuned signature (again);
wenzelm
parents: 44300
diff changeset
   381
65f60d9ac4bf tuned signature (again);
wenzelm
parents: 44300
diff changeset
   382
(* cancel *)
65f60d9ac4bf tuned signature (again);
wenzelm
parents: 44300
diff changeset
   383
49906
06a3570b0f0a more official Future.terminate;
wenzelm
parents: 49894
diff changeset
   384
fun cancel_group_unsynchronized group = (*requires SYNCHRONIZED*)
44299
061599cb6eb0 refined Future.cancel: explicit future allows to join actual cancellation;
wenzelm
parents: 44298
diff changeset
   385
  let
47421
9624408d8827 always signal after cancel_group: passive tasks may have become active;
wenzelm
parents: 47404
diff changeset
   386
    val _ = if null (cancel_now group) then () else cancel_later group;
9624408d8827 always signal after cancel_group: passive tasks may have become active;
wenzelm
parents: 47404
diff changeset
   387
    val _ = signal work_available;
9624408d8827 always signal after cancel_group: passive tasks may have become active;
wenzelm
parents: 47404
diff changeset
   388
    val _ = scheduler_check ();
49906
06a3570b0f0a more official Future.terminate;
wenzelm
parents: 49894
diff changeset
   389
  in () end;
06a3570b0f0a more official Future.terminate;
wenzelm
parents: 49894
diff changeset
   390
06a3570b0f0a more official Future.terminate;
wenzelm
parents: 49894
diff changeset
   391
fun cancel_group group =
06a3570b0f0a more official Future.terminate;
wenzelm
parents: 49894
diff changeset
   392
  SYNCHRONIZED "cancel_group" (fn () => cancel_group_unsynchronized group);
44299
061599cb6eb0 refined Future.cancel: explicit future allows to join actual cancellation;
wenzelm
parents: 44298
diff changeset
   393
44301
65f60d9ac4bf tuned signature (again);
wenzelm
parents: 44300
diff changeset
   394
fun cancel x = cancel_group (Task_Queue.group_of_task (task_of x));
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   395
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   396
50914
fe4714886d92 identify future results more carefully, to avoid odd duplication of error messages, notably from forked goals;
wenzelm
parents: 50911
diff changeset
   397
(* results *)
fe4714886d92 identify future results more carefully, to avoid odd duplication of error messages, notably from forked goals;
wenzelm
parents: 50911
diff changeset
   398
56333
38f1422ef473 support bulk messages consisting of small string segments, which are more healthy to the Poly/ML RTS and might prevent spurious GC crashes such as MTGCProcessMarkPointers::ScanAddressesInObject;
wenzelm
parents: 54671
diff changeset
   399
fun error_message pos ((serial, msg), exec_id) =
50916
fd902b616b48 proper runtime position (cf. fe4714886d92 and Toplevel.error_msg) -- to make error messages actually appear in the document;
wenzelm
parents: 50914
diff changeset
   400
  Position.setmp_thread_data pos (fn () =>
50931
a7484a7b6c8a clarified Future.error_msg: slightly more robust id check, actually suppress displaced messages;
wenzelm
parents: 50916
diff changeset
   401
    let val id = Position.get_id pos in
a7484a7b6c8a clarified Future.error_msg: slightly more robust id check, actually suppress displaced messages;
wenzelm
parents: 50916
diff changeset
   402
      if is_none id orelse is_none exec_id orelse id = exec_id
54387
890e983cb07b tuned signature;
wenzelm
parents: 54369
diff changeset
   403
      then Output.error_message' (serial, msg) else ()
50931
a7484a7b6c8a clarified Future.error_msg: slightly more robust id check, actually suppress displaced messages;
wenzelm
parents: 50916
diff changeset
   404
    end) ();
44110
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   405
50914
fe4714886d92 identify future results more carefully, to avoid odd duplication of error messages, notably from forked goals;
wenzelm
parents: 50911
diff changeset
   406
fun identify_result pos res =
61077
06cca32aa519 thread context for exceptions from forks, e.g. relevant when printing errors;
wenzelm
parents: 60911
diff changeset
   407
  res |> Exn.map_exn (fn exn =>
06cca32aa519 thread context for exceptions from forks, e.g. relevant when printing errors;
wenzelm
parents: 60911
diff changeset
   408
    let val exec_id =
06cca32aa519 thread context for exceptions from forks, e.g. relevant when printing errors;
wenzelm
parents: 60911
diff changeset
   409
      (case Position.get_id pos of
06cca32aa519 thread context for exceptions from forks, e.g. relevant when printing errors;
wenzelm
parents: 60911
diff changeset
   410
        NONE => []
06cca32aa519 thread context for exceptions from forks, e.g. relevant when printing errors;
wenzelm
parents: 60911
diff changeset
   411
      | SOME id => [(Markup.exec_idN, id)])
06cca32aa519 thread context for exceptions from forks, e.g. relevant when printing errors;
wenzelm
parents: 60911
diff changeset
   412
    in Par_Exn.identify exec_id exn end);
50914
fe4714886d92 identify future results more carefully, to avoid odd duplication of error messages, notably from forked goals;
wenzelm
parents: 50911
diff changeset
   413
fe4714886d92 identify future results more carefully, to avoid odd duplication of error messages, notably from forked goals;
wenzelm
parents: 50911
diff changeset
   414
fun assign_result group result res =
44110
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   415
  let
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   416
    val _ = Single_Assignment.assign result res
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   417
      handle exn as Fail _ =>
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   418
        (case Single_Assignment.peek result of
62505
9e2a65912111 clarified modules;
wenzelm
parents: 62359
diff changeset
   419
          SOME (Exn.Exn e) => Exn.reraise (if Exn.is_interrupt e then e else exn)
9e2a65912111 clarified modules;
wenzelm
parents: 62359
diff changeset
   420
        | _ => Exn.reraise exn);
44110
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   421
    val ok =
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   422
      (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
   423
        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
   424
          (SYNCHRONIZED "cancel" (fn () => Task_Queue.cancel_group group exn); false)
44110
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   425
      | Exn.Res _ => true);
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   426
  in ok end;
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   427
50914
fe4714886d92 identify future results more carefully, to avoid odd duplication of error messages, notably from forked goals;
wenzelm
parents: 50911
diff changeset
   428
fe4714886d92 identify future results more carefully, to avoid odd duplication of error messages, notably from forked goals;
wenzelm
parents: 50911
diff changeset
   429
(* future jobs *)
fe4714886d92 identify future results more carefully, to avoid odd duplication of error messages, notably from forked goals;
wenzelm
parents: 50911
diff changeset
   430
54649
99b9249b3e05 more official task context via Task_Queue.enroll, which is required to participate in group cancellation (e.g. to terminate command exec);
wenzelm
parents: 54637
diff changeset
   431
fun future_job group atts (e: unit -> 'a) =
44110
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   432
  let
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   433
    val result = Single_Assignment.var "future" : 'a result;
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   434
    val pos = Position.thread_data ();
60911
858694df711b default ML context for forks, e.g. relevant for debugging and toplevel pretty-printing;
wenzelm
parents: 60764
diff changeset
   435
    val context = Context.thread_data ();
44110
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   436
    fun job ok =
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   437
      let
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   438
        val res =
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   439
          if ok then
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   440
            Exn.capture (fn () =>
60911
858694df711b default ML context for forks, e.g. relevant for debugging and toplevel pretty-printing;
wenzelm
parents: 60764
diff changeset
   441
              Multithreading.with_attributes atts (fn _ =>
858694df711b default ML context for forks, e.g. relevant for debugging and toplevel pretty-printing;
wenzelm
parents: 60764
diff changeset
   442
                (Position.setmp_thread_data pos o Context.setmp_thread_data context) e ())) ()
44110
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   443
          else Exn.interrupt_exn;
50914
fe4714886d92 identify future results more carefully, to avoid odd duplication of error messages, notably from forked goals;
wenzelm
parents: 50911
diff changeset
   444
      in assign_result group result (identify_result pos res) end;
44110
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   445
  in (result, job) end;
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   446
058520fa03a8 tuned source structure;
wenzelm
parents: 43952
diff changeset
   447
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   448
(* fork *)
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   449
44427
c4a86d72a5cc tuned signature;
wenzelm
parents: 44387
diff changeset
   450
type params = {name: string, group: group option, deps: task list, pri: int, interrupts: bool};
c4a86d72a5cc tuned signature;
wenzelm
parents: 44387
diff changeset
   451
val default_params: params = {name = "", group = NONE, deps = [], pri = 0, interrupts = true};
44113
0baa8bbd355a future_job: explicit indication of interrupts;
wenzelm
parents: 44111
diff changeset
   452
44427
c4a86d72a5cc tuned signature;
wenzelm
parents: 44387
diff changeset
   453
fun forks ({name, group, deps, pri, interrupts}: params) es =
41674
7da257539a8d tuned signature;
wenzelm
parents: 41673
diff changeset
   454
  if null es then []
7da257539a8d tuned signature;
wenzelm
parents: 41673
diff changeset
   455
  else
7da257539a8d tuned signature;
wenzelm
parents: 41673
diff changeset
   456
    let
7da257539a8d tuned signature;
wenzelm
parents: 41673
diff changeset
   457
      val grp =
7da257539a8d tuned signature;
wenzelm
parents: 41673
diff changeset
   458
        (case group of
7da257539a8d tuned signature;
wenzelm
parents: 41673
diff changeset
   459
          NONE => worker_subgroup ()
7da257539a8d tuned signature;
wenzelm
parents: 41673
diff changeset
   460
        | SOME grp => grp);
41708
d2e6b1132df2 tuned signature;
wenzelm
parents: 41695
diff changeset
   461
      fun enqueue e queue =
41674
7da257539a8d tuned signature;
wenzelm
parents: 41673
diff changeset
   462
        let
54649
99b9249b3e05 more official task context via Task_Queue.enroll, which is required to participate in group cancellation (e.g. to terminate command exec);
wenzelm
parents: 54637
diff changeset
   463
          val atts =
99b9249b3e05 more official task context via Task_Queue.enroll, which is required to participate in group cancellation (e.g. to terminate command exec);
wenzelm
parents: 54637
diff changeset
   464
            if interrupts
99b9249b3e05 more official task context via Task_Queue.enroll, which is required to participate in group cancellation (e.g. to terminate command exec);
wenzelm
parents: 54637
diff changeset
   465
            then Multithreading.private_interrupts
99b9249b3e05 more official task context via Task_Queue.enroll, which is required to participate in group cancellation (e.g. to terminate command exec);
wenzelm
parents: 54637
diff changeset
   466
            else Multithreading.no_interrupts;
99b9249b3e05 more official task context via Task_Queue.enroll, which is required to participate in group cancellation (e.g. to terminate command exec);
wenzelm
parents: 54637
diff changeset
   467
          val (result, job) = future_job grp atts e;
41708
d2e6b1132df2 tuned signature;
wenzelm
parents: 41695
diff changeset
   468
          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
   469
          val future = Future {promised = false, task = task, result = result};
41708
d2e6b1132df2 tuned signature;
wenzelm
parents: 41695
diff changeset
   470
        in (future, queue') end;
41674
7da257539a8d tuned signature;
wenzelm
parents: 41673
diff changeset
   471
    in
7da257539a8d tuned signature;
wenzelm
parents: 41673
diff changeset
   472
      SYNCHRONIZED "enqueue" (fn () =>
7da257539a8d tuned signature;
wenzelm
parents: 41673
diff changeset
   473
        let
41708
d2e6b1132df2 tuned signature;
wenzelm
parents: 41695
diff changeset
   474
          val (futures, queue') = fold_map enqueue es (! queue);
d2e6b1132df2 tuned signature;
wenzelm
parents: 41695
diff changeset
   475
          val _ = queue := queue';
d2e6b1132df2 tuned signature;
wenzelm
parents: 41695
diff changeset
   476
          val minimal = forall (not o Task_Queue.known_task queue') deps;
41674
7da257539a8d tuned signature;
wenzelm
parents: 41673
diff changeset
   477
          val _ = if minimal then signal work_available else ();
7da257539a8d tuned signature;
wenzelm
parents: 41673
diff changeset
   478
          val _ = scheduler_check ();
7da257539a8d tuned signature;
wenzelm
parents: 41673
diff changeset
   479
        in futures end)
7da257539a8d tuned signature;
wenzelm
parents: 41673
diff changeset
   480
    end;
28162
55772e4e95e0 tuned check_cache;
wenzelm
parents: 28156
diff changeset
   481
50983
1290afb88f90 tuned signature;
wenzelm
parents: 50975
diff changeset
   482
fun fork e =
1290afb88f90 tuned signature;
wenzelm
parents: 50975
diff changeset
   483
  (singleton o forks) {name = "fork", group = NONE, deps = [], pri = 0, interrupts = true} e;
28186
6a8417f36837 cancel: check_scheduler;
wenzelm
parents: 28177
diff changeset
   484
6a8417f36837 cancel: check_scheduler;
wenzelm
parents: 28177
diff changeset
   485
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   486
(* join *)
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   487
32099
5382c93108db propagate exceptions within future groups;
wenzelm
parents: 32096
diff changeset
   488
fun get_result x =
5382c93108db propagate exceptions within future groups;
wenzelm
parents: 32096
diff changeset
   489
  (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
   490
    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
   491
  | SOME res =>
69c6d3e87660 more abstract treatment of interrupts in structure Exn -- hardly ever need to mention Interrupt literally;
wenzelm
parents: 38236
diff changeset
   492
      if Exn.is_interrupt_exn res then
44247
270366301bd7 more systematic handling of parallel exceptions;
wenzelm
parents: 44198
diff changeset
   493
        (case Task_Queue.group_status (Task_Queue.group_of_task (task_of x)) of
53190
5d92649a310e simplified Goal.forked_proofs: status is determined via group instead of dummy future (see also Pure/PIDE/execution.ML);
wenzelm
parents: 53189
diff changeset
   494
          [] => res
5d92649a310e simplified Goal.forked_proofs: status is determined via group instead of dummy future (see also Pure/PIDE/execution.ML);
wenzelm
parents: 53189
diff changeset
   495
        | exns => Exn.Exn (Par_Exn.make exns))
39232
69c6d3e87660 more abstract treatment of interrupts in structure Exn -- hardly ever need to mention Interrupt literally;
wenzelm
parents: 38236
diff changeset
   496
      else res);
28186
6a8417f36837 cancel: check_scheduler;
wenzelm
parents: 28177
diff changeset
   497
49935
d1ecb3554b25 clarified Future.map (again): finished value is mapped in-place, which saves task structures and changes error behaviour slightly (tolerance against canceled group of old value etc.);
wenzelm
parents: 49910
diff changeset
   498
local
d1ecb3554b25 clarified Future.map (again): finished value is mapped in-place, which saves task structures and changes error behaviour slightly (tolerance against canceled group of old value etc.);
wenzelm
parents: 49910
diff changeset
   499
59465
c21b65a6834b clarified worker_wait;
wenzelm
parents: 59340
diff changeset
   500
fun join_next atts deps = (*requires SYNCHRONIZED*)
41695
afdbec23b92b eliminated slightly odd abstract type Task_Queue.deps;
wenzelm
parents: 41683
diff changeset
   501
  if null deps then NONE
32224
a46f5e9b1718 dequeue_towards: always return active tasks;
wenzelm
parents: 32222
diff changeset
   502
  else
41681
b5d7b15166bf Future.join_results: discontinued post-hoc recording of dynamic dependencies;
wenzelm
parents: 41680
diff changeset
   503
    (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
   504
      (NONE, []) => NONE
afdbec23b92b eliminated slightly odd abstract type Task_Queue.deps;
wenzelm
parents: 41683
diff changeset
   505
    | (NONE, deps') =>
59465
c21b65a6834b clarified worker_wait;
wenzelm
parents: 59340
diff changeset
   506
       (worker_waiting deps' (fn () =>
c21b65a6834b clarified worker_wait;
wenzelm
parents: 59340
diff changeset
   507
          Multithreading.with_attributes atts (fn _ =>
c21b65a6834b clarified worker_wait;
wenzelm
parents: 59340
diff changeset
   508
            Exn.release (worker_wait Waiting work_finished)));
c21b65a6834b clarified worker_wait;
wenzelm
parents: 59340
diff changeset
   509
        join_next atts deps')
32224
a46f5e9b1718 dequeue_towards: always return active tasks;
wenzelm
parents: 32222
diff changeset
   510
    | (SOME work, deps') => SOME (work, deps'));
32095
ad4be204fdfe tuned tracing;
wenzelm
parents: 32058
diff changeset
   511
59465
c21b65a6834b clarified worker_wait;
wenzelm
parents: 59340
diff changeset
   512
fun join_loop atts deps =
c21b65a6834b clarified worker_wait;
wenzelm
parents: 59340
diff changeset
   513
  (case SYNCHRONIZED "join" (fn () => join_next atts deps) of
c21b65a6834b clarified worker_wait;
wenzelm
parents: 59340
diff changeset
   514
    NONE => ()
c21b65a6834b clarified worker_wait;
wenzelm
parents: 59340
diff changeset
   515
  | SOME (work, deps') => (worker_joining (fn () => worker_exec work); join_loop atts deps'));
32814
81897d30b97f added Task_Queue.depend (again) -- light-weight version for transitive graph;
wenzelm
parents: 32738
diff changeset
   516
29551
95e469919c3e join_results: when dependencies are resulved (but not finished yet),
wenzelm
parents: 29431
diff changeset
   517
in
95e469919c3e join_results: when dependencies are resulved (but not finished yet),
wenzelm
parents: 29431
diff changeset
   518
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   519
fun join_results xs =
41679
79716cb61bfd refined task timing: joining vs. waiting;
wenzelm
parents: 41674
diff changeset
   520
  let
79716cb61bfd refined task timing: joining vs. waiting;
wenzelm
parents: 41674
diff changeset
   521
    val _ =
79716cb61bfd refined task timing: joining vs. waiting;
wenzelm
parents: 41674
diff changeset
   522
      if forall is_finished xs then ()
59465
c21b65a6834b clarified worker_wait;
wenzelm
parents: 59340
diff changeset
   523
      else if is_some (worker_task ()) then
c21b65a6834b clarified worker_wait;
wenzelm
parents: 59340
diff changeset
   524
        Multithreading.with_attributes Multithreading.no_interrupts
c21b65a6834b clarified worker_wait;
wenzelm
parents: 59340
diff changeset
   525
          (fn orig_atts => join_loop orig_atts (map task_of xs))
41681
b5d7b15166bf Future.join_results: discontinued post-hoc recording of dynamic dependencies;
wenzelm
parents: 41680
diff changeset
   526
      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
   527
  in map get_result xs end;
28186
6a8417f36837 cancel: check_scheduler;
wenzelm
parents: 28177
diff changeset
   528
29551
95e469919c3e join_results: when dependencies are resulved (but not finished yet),
wenzelm
parents: 29431
diff changeset
   529
end;
95e469919c3e join_results: when dependencies are resulved (but not finished yet),
wenzelm
parents: 29431
diff changeset
   530
28647
8068cdc84e7e join_results: allow CRITICAL join of finished futures;
wenzelm
parents: 28645
diff changeset
   531
fun join_result x = singleton join_results x;
44330
b28e091f683a added Future.joins convenience;
wenzelm
parents: 44301
diff changeset
   532
fun joins xs = Par_Exn.release_all (join_results xs);
28647
8068cdc84e7e join_results: allow CRITICAL join of finished futures;
wenzelm
parents: 28645
diff changeset
   533
fun join x = Exn.release (join_result x);
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   534
54369
7bf7b2903fb9 tuned signature;
wenzelm
parents: 53190
diff changeset
   535
fun join_tasks tasks =
7bf7b2903fb9 tuned signature;
wenzelm
parents: 53190
diff changeset
   536
  if null tasks then ()
7bf7b2903fb9 tuned signature;
wenzelm
parents: 53190
diff changeset
   537
  else
7bf7b2903fb9 tuned signature;
wenzelm
parents: 53190
diff changeset
   538
    (singleton o forks)
7bf7b2903fb9 tuned signature;
wenzelm
parents: 53190
diff changeset
   539
      {name = "join_tasks", group = SOME (new_group NONE),
7bf7b2903fb9 tuned signature;
wenzelm
parents: 53190
diff changeset
   540
        deps = tasks, pri = 0, interrupts = false} I
7bf7b2903fb9 tuned signature;
wenzelm
parents: 53190
diff changeset
   541
    |> join;
7bf7b2903fb9 tuned signature;
wenzelm
parents: 53190
diff changeset
   542
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   543
54649
99b9249b3e05 more official task context via Task_Queue.enroll, which is required to participate in group cancellation (e.g. to terminate command exec);
wenzelm
parents: 54637
diff changeset
   544
(* task context for running thread *)
99b9249b3e05 more official task context via Task_Queue.enroll, which is required to participate in group cancellation (e.g. to terminate command exec);
wenzelm
parents: 54637
diff changeset
   545
99b9249b3e05 more official task context via Task_Queue.enroll, which is required to participate in group cancellation (e.g. to terminate command exec);
wenzelm
parents: 54637
diff changeset
   546
fun task_context name group f x =
99b9249b3e05 more official task context via Task_Queue.enroll, which is required to participate in group cancellation (e.g. to terminate command exec);
wenzelm
parents: 54637
diff changeset
   547
  Multithreading.with_attributes Multithreading.no_interrupts (fn orig_atts =>
99b9249b3e05 more official task context via Task_Queue.enroll, which is required to participate in group cancellation (e.g. to terminate command exec);
wenzelm
parents: 54637
diff changeset
   548
    let
99b9249b3e05 more official task context via Task_Queue.enroll, which is required to participate in group cancellation (e.g. to terminate command exec);
wenzelm
parents: 54637
diff changeset
   549
      val (result, job) = future_job group orig_atts (fn () => f x);
99b9249b3e05 more official task context via Task_Queue.enroll, which is required to participate in group cancellation (e.g. to terminate command exec);
wenzelm
parents: 54637
diff changeset
   550
      val task =
99b9249b3e05 more official task context via Task_Queue.enroll, which is required to participate in group cancellation (e.g. to terminate command exec);
wenzelm
parents: 54637
diff changeset
   551
        SYNCHRONIZED "enroll" (fn () =>
99b9249b3e05 more official task context via Task_Queue.enroll, which is required to participate in group cancellation (e.g. to terminate command exec);
wenzelm
parents: 54637
diff changeset
   552
          Unsynchronized.change_result queue (Task_Queue.enroll (Thread.self ()) name group));
99b9249b3e05 more official task context via Task_Queue.enroll, which is required to participate in group cancellation (e.g. to terminate command exec);
wenzelm
parents: 54637
diff changeset
   553
      val _ = worker_exec (task, [job]);
99b9249b3e05 more official task context via Task_Queue.enroll, which is required to participate in group cancellation (e.g. to terminate command exec);
wenzelm
parents: 54637
diff changeset
   554
    in
99b9249b3e05 more official task context via Task_Queue.enroll, which is required to participate in group cancellation (e.g. to terminate command exec);
wenzelm
parents: 54637
diff changeset
   555
      (case Single_Assignment.peek result of
99b9249b3e05 more official task context via Task_Queue.enroll, which is required to participate in group cancellation (e.g. to terminate command exec);
wenzelm
parents: 54637
diff changeset
   556
        NONE => raise Fail "Missing task context result"
99b9249b3e05 more official task context via Task_Queue.enroll, which is required to participate in group cancellation (e.g. to terminate command exec);
wenzelm
parents: 54637
diff changeset
   557
      | SOME res => Exn.release res)
99b9249b3e05 more official task context via Task_Queue.enroll, which is required to participate in group cancellation (e.g. to terminate command exec);
wenzelm
parents: 54637
diff changeset
   558
    end);
99b9249b3e05 more official task context via Task_Queue.enroll, which is required to participate in group cancellation (e.g. to terminate command exec);
wenzelm
parents: 54637
diff changeset
   559
99b9249b3e05 more official task context via Task_Queue.enroll, which is required to participate in group cancellation (e.g. to terminate command exec);
wenzelm
parents: 54637
diff changeset
   560
51325
bcd6b1aa4db5 more uniform Future.map: always internalize failure;
wenzelm
parents: 51283
diff changeset
   561
(* fast-path operations -- bypass task queue if possible *)
34277
7325a5e3587f added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents: 33416
diff changeset
   562
44294
a0ddd5760444 clarified Future.cond_forks: more uniform handling of exceptional situations;
wenzelm
parents: 44268
diff changeset
   563
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
   564
  let
45136
2afb928c71ca static dummy_task (again) to avoid a few extra allocations;
wenzelm
parents: 44427
diff changeset
   565
    val task = Task_Queue.dummy_task;
41683
73dde8006820 maintain Task_Queue.group within Task_Queue.task;
wenzelm
parents: 41681
diff changeset
   566
    val group = Task_Queue.group_of_task task;
35016
c0f2e49bccfc result: Single_Assignment.var;
wenzelm
parents: 34281
diff changeset
   567
    val result = Single_Assignment.var "value" : 'a result;
50914
fe4714886d92 identify future results more carefully, to avoid odd duplication of error messages, notably from forked goals;
wenzelm
parents: 50911
diff changeset
   568
    val _ = assign_result group result (identify_result (Position.thread_data ()) res);
41683
73dde8006820 maintain Task_Queue.group within Task_Queue.task;
wenzelm
parents: 41681
diff changeset
   569
  in Future {promised = false, task = task, result = result} end;
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   570
44294
a0ddd5760444 clarified Future.cond_forks: more uniform handling of exceptional situations;
wenzelm
parents: 44268
diff changeset
   571
fun value x = value_result (Exn.Res x);
a0ddd5760444 clarified Future.cond_forks: more uniform handling of exceptional situations;
wenzelm
parents: 44268
diff changeset
   572
44330
b28e091f683a added Future.joins convenience;
wenzelm
parents: 44301
diff changeset
   573
fun cond_forks args es =
b28e091f683a added Future.joins convenience;
wenzelm
parents: 44301
diff changeset
   574
  if Multithreading.enabled () then forks args es
b28e091f683a added Future.joins convenience;
wenzelm
parents: 44301
diff changeset
   575
  else map (fn e => value_result (Exn.interruptible_capture e ())) es;
b28e091f683a added Future.joins convenience;
wenzelm
parents: 44301
diff changeset
   576
29384
a3c7e9ae9b71 more robust propagation of errors through bulk jobs;
wenzelm
parents: 29366
diff changeset
   577
fun map_future f x =
51325
bcd6b1aa4db5 more uniform Future.map: always internalize failure;
wenzelm
parents: 51283
diff changeset
   578
  if is_finished x then value_result (Exn.interruptible_capture (f o join) x)
49935
d1ecb3554b25 clarified Future.map (again): finished value is mapped in-place, which saves task structures and changes error behaviour slightly (tolerance against canceled group of old value etc.);
wenzelm
parents: 49910
diff changeset
   579
  else
d1ecb3554b25 clarified Future.map (again): finished value is mapped in-place, which saves task structures and changes error behaviour slightly (tolerance against canceled group of old value etc.);
wenzelm
parents: 49910
diff changeset
   580
    let
d1ecb3554b25 clarified Future.map (again): finished value is mapped in-place, which saves task structures and changes error behaviour slightly (tolerance against canceled group of old value etc.);
wenzelm
parents: 49910
diff changeset
   581
      val task = task_of x;
d1ecb3554b25 clarified Future.map (again): finished value is mapped in-place, which saves task structures and changes error behaviour slightly (tolerance against canceled group of old value etc.);
wenzelm
parents: 49910
diff changeset
   582
      val group = Task_Queue.group_of_task task;
54649
99b9249b3e05 more official task context via Task_Queue.enroll, which is required to participate in group cancellation (e.g. to terminate command exec);
wenzelm
parents: 54637
diff changeset
   583
      val (result, job) =
99b9249b3e05 more official task context via Task_Queue.enroll, which is required to participate in group cancellation (e.g. to terminate command exec);
wenzelm
parents: 54637
diff changeset
   584
        future_job group Multithreading.private_interrupts (fn () => f (join x));
29384
a3c7e9ae9b71 more robust propagation of errors through bulk jobs;
wenzelm
parents: 29366
diff changeset
   585
49935
d1ecb3554b25 clarified Future.map (again): finished value is mapped in-place, which saves task structures and changes error behaviour slightly (tolerance against canceled group of old value etc.);
wenzelm
parents: 49910
diff changeset
   586
      val extended = SYNCHRONIZED "extend" (fn () =>
d1ecb3554b25 clarified Future.map (again): finished value is mapped in-place, which saves task structures and changes error behaviour slightly (tolerance against canceled group of old value etc.);
wenzelm
parents: 49910
diff changeset
   587
        (case Task_Queue.extend task job (! queue) of
d1ecb3554b25 clarified Future.map (again): finished value is mapped in-place, which saves task structures and changes error behaviour slightly (tolerance against canceled group of old value etc.);
wenzelm
parents: 49910
diff changeset
   588
          SOME queue' => (queue := queue'; true)
d1ecb3554b25 clarified Future.map (again): finished value is mapped in-place, which saves task structures and changes error behaviour slightly (tolerance against canceled group of old value etc.);
wenzelm
parents: 49910
diff changeset
   589
        | NONE => false));
d1ecb3554b25 clarified Future.map (again): finished value is mapped in-place, which saves task structures and changes error behaviour slightly (tolerance against canceled group of old value etc.);
wenzelm
parents: 49910
diff changeset
   590
    in
d1ecb3554b25 clarified Future.map (again): finished value is mapped in-place, which saves task structures and changes error behaviour slightly (tolerance against canceled group of old value etc.);
wenzelm
parents: 49910
diff changeset
   591
      if extended then Future {promised = false, task = task, result = result}
d1ecb3554b25 clarified Future.map (again): finished value is mapped in-place, which saves task structures and changes error behaviour slightly (tolerance against canceled group of old value etc.);
wenzelm
parents: 49910
diff changeset
   592
      else
d1ecb3554b25 clarified Future.map (again): finished value is mapped in-place, which saves task structures and changes error behaviour slightly (tolerance against canceled group of old value etc.);
wenzelm
parents: 49910
diff changeset
   593
        (singleton o cond_forks)
d1ecb3554b25 clarified Future.map (again): finished value is mapped in-place, which saves task structures and changes error behaviour slightly (tolerance against canceled group of old value etc.);
wenzelm
parents: 49910
diff changeset
   594
          {name = "map_future", group = SOME group, deps = [task],
d1ecb3554b25 clarified Future.map (again): finished value is mapped in-place, which saves task structures and changes error behaviour slightly (tolerance against canceled group of old value etc.);
wenzelm
parents: 49910
diff changeset
   595
            pri = Task_Queue.pri_of_task task, interrupts = true}
d1ecb3554b25 clarified Future.map (again): finished value is mapped in-place, which saves task structures and changes error behaviour slightly (tolerance against canceled group of old value etc.);
wenzelm
parents: 49910
diff changeset
   596
          (fn () => f (join x))
d1ecb3554b25 clarified Future.map (again): finished value is mapped in-place, which saves task structures and changes error behaviour slightly (tolerance against canceled group of old value etc.);
wenzelm
parents: 49910
diff changeset
   597
    end;
28979
3ce619d8d432 fork/map: no inheritance of group (structure is nested, not parallel);
wenzelm
parents: 28972
diff changeset
   598
28191
9e5f556409c6 future: allow explicit group;
wenzelm
parents: 28186
diff changeset
   599
34277
7325a5e3587f added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents: 33416
diff changeset
   600
(* promised futures -- fulfilled by external means *)
7325a5e3587f added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents: 33416
diff changeset
   601
44298
b8f8488704e2 Future.promise: explicit abort operation (like uninterruptible future job);
wenzelm
parents: 44295
diff changeset
   602
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
   603
  let
35016
c0f2e49bccfc result: Single_Assignment.var;
wenzelm
parents: 34281
diff changeset
   604
    val result = Single_Assignment.var "promise" : 'a result;
44298
b8f8488704e2 Future.promise: explicit abort operation (like uninterruptible future job);
wenzelm
parents: 44295
diff changeset
   605
    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
   606
      handle Fail _ => true
307e3d07d19f Future.promise: more robust treatment of concurrent abort vs. fulfill (amending 047c96f41455);
wenzelm
parents: 39232
diff changeset
   607
        | exn =>
44298
b8f8488704e2 Future.promise: explicit abort operation (like uninterruptible future job);
wenzelm
parents: 44295
diff changeset
   608
            if Exn.is_interrupt exn
b8f8488704e2 Future.promise: explicit abort operation (like uninterruptible future job);
wenzelm
parents: 44295
diff changeset
   609
            then raise Fail "Concurrent attempt to fulfill promise"
62505
9e2a65912111 clarified modules;
wenzelm
parents: 62359
diff changeset
   610
            else Exn.reraise exn;
44298
b8f8488704e2 Future.promise: explicit abort operation (like uninterruptible future job);
wenzelm
parents: 44295
diff changeset
   611
    fun job () =
b8f8488704e2 Future.promise: explicit abort operation (like uninterruptible future job);
wenzelm
parents: 44295
diff changeset
   612
      Multithreading.with_attributes Multithreading.no_interrupts
47423
8a179a0493e3 more robust Future.fulfill wrt. duplicate assignment and interrupt;
wenzelm
parents: 47421
diff changeset
   613
        (fn _ => Exn.release (Exn.capture assign () before abort ()));
37854
047c96f41455 back to more strict dependencies, even for canceled groups (reverting parts of 02936e77a07c);
wenzelm
parents: 37852
diff changeset
   614
    val task = SYNCHRONIZED "enqueue_passive" (fn () =>
44298
b8f8488704e2 Future.promise: explicit abort operation (like uninterruptible future job);
wenzelm
parents: 44295
diff changeset
   615
      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
   616
  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
   617
44298
b8f8488704e2 Future.promise: explicit abort operation (like uninterruptible future job);
wenzelm
parents: 44295
diff changeset
   618
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
   619
41683
73dde8006820 maintain Task_Queue.group within Task_Queue.task;
wenzelm
parents: 41681
diff changeset
   620
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
   621
  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
   622
  else
307e3d07d19f Future.promise: more robust treatment of concurrent abort vs. fulfill (amending 047c96f41455);
wenzelm
parents: 39232
diff changeset
   623
    let
41683
73dde8006820 maintain Task_Queue.group within Task_Queue.task;
wenzelm
parents: 41681
diff changeset
   624
      val group = Task_Queue.group_of_task task;
50914
fe4714886d92 identify future results more carefully, to avoid odd duplication of error messages, notably from forked goals;
wenzelm
parents: 50911
diff changeset
   625
      val pos = Position.thread_data ();
fe4714886d92 identify future results more carefully, to avoid odd duplication of error messages, notably from forked goals;
wenzelm
parents: 50911
diff changeset
   626
      fun job ok =
fe4714886d92 identify future results more carefully, to avoid odd duplication of error messages, notably from forked goals;
wenzelm
parents: 50911
diff changeset
   627
        assign_result group result (if ok then identify_result pos res else Exn.interrupt_exn);
39243
307e3d07d19f Future.promise: more robust treatment of concurrent abort vs. fulfill (amending 047c96f41455);
wenzelm
parents: 39232
diff changeset
   628
      val _ =
307e3d07d19f Future.promise: more robust treatment of concurrent abort vs. fulfill (amending 047c96f41455);
wenzelm
parents: 39232
diff changeset
   629
        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
   630
          let
47423
8a179a0493e3 more robust Future.fulfill wrt. duplicate assignment and interrupt;
wenzelm
parents: 47421
diff changeset
   631
            val passive_job =
39243
307e3d07d19f Future.promise: more robust treatment of concurrent abort vs. fulfill (amending 047c96f41455);
wenzelm
parents: 39232
diff changeset
   632
              SYNCHRONIZED "fulfill_result" (fn () =>
307e3d07d19f Future.promise: more robust treatment of concurrent abort vs. fulfill (amending 047c96f41455);
wenzelm
parents: 39232
diff changeset
   633
                Unsynchronized.change_result queue
307e3d07d19f Future.promise: more robust treatment of concurrent abort vs. fulfill (amending 047c96f41455);
wenzelm
parents: 39232
diff changeset
   634
                  (Task_Queue.dequeue_passive (Thread.self ()) task));
47423
8a179a0493e3 more robust Future.fulfill wrt. duplicate assignment and interrupt;
wenzelm
parents: 47421
diff changeset
   635
          in
8a179a0493e3 more robust Future.fulfill wrt. duplicate assignment and interrupt;
wenzelm
parents: 47421
diff changeset
   636
            (case passive_job of
8a179a0493e3 more robust Future.fulfill wrt. duplicate assignment and interrupt;
wenzelm
parents: 47421
diff changeset
   637
              SOME true => worker_exec (task, [job])
8a179a0493e3 more robust Future.fulfill wrt. duplicate assignment and interrupt;
wenzelm
parents: 47421
diff changeset
   638
            | SOME false => ()
8a179a0493e3 more robust Future.fulfill wrt. duplicate assignment and interrupt;
wenzelm
parents: 47421
diff changeset
   639
            | NONE => ignore (job (not (Task_Queue.is_canceled group))))
8a179a0493e3 more robust Future.fulfill wrt. duplicate assignment and interrupt;
wenzelm
parents: 47421
diff changeset
   640
          end);
41681
b5d7b15166bf Future.join_results: discontinued post-hoc recording of dynamic dependencies;
wenzelm
parents: 41680
diff changeset
   641
      val _ =
41695
afdbec23b92b eliminated slightly odd abstract type Task_Queue.deps;
wenzelm
parents: 41683
diff changeset
   642
        if is_some (Single_Assignment.peek result) then ()
afdbec23b92b eliminated slightly odd abstract type Task_Queue.deps;
wenzelm
parents: 41683
diff changeset
   643
        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
   644
    in () end;
34277
7325a5e3587f added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents: 33416
diff changeset
   645
43761
e72ba84ae58f tuned signature -- corresponding to Scala version;
wenzelm
parents: 43665
diff changeset
   646
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
   647
7325a5e3587f added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents: 33416
diff changeset
   648
54369
7bf7b2903fb9 tuned signature;
wenzelm
parents: 53190
diff changeset
   649
(* group snapshot *)
7bf7b2903fb9 tuned signature;
wenzelm
parents: 53190
diff changeset
   650
7bf7b2903fb9 tuned signature;
wenzelm
parents: 53190
diff changeset
   651
fun group_snapshot group =
7bf7b2903fb9 tuned signature;
wenzelm
parents: 53190
diff changeset
   652
  SYNCHRONIZED "group_snapshot" (fn () =>
7bf7b2903fb9 tuned signature;
wenzelm
parents: 53190
diff changeset
   653
    Task_Queue.group_tasks (! queue) group);
7bf7b2903fb9 tuned signature;
wenzelm
parents: 53190
diff changeset
   654
7bf7b2903fb9 tuned signature;
wenzelm
parents: 53190
diff changeset
   655
32228
7622c03141b0 scheduler: shutdown spontaneously (after some delay) if queue is empty;
wenzelm
parents: 32227
diff changeset
   656
(* shutdown *)
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   657
28203
88f18387f1c9 shutdown: global join-and-shutdown operation;
wenzelm
parents: 28202
diff changeset
   658
fun shutdown () =
62359
6709e51d5c11 unconditional Multithreading;
wenzelm
parents: 61556
diff changeset
   659
  if is_some (worker_task ()) then
51283
fefd07697979 disallow shutdown from worker, which would lead to deadlock since the scheduler cannot terminate;
wenzelm
parents: 51281
diff changeset
   660
    raise Fail "Cannot shutdown while running as worker thread"
fefd07697979 disallow shutdown from worker, which would lead to deadlock since the scheduler cannot terminate;
wenzelm
parents: 51281
diff changeset
   661
  else
28276
fbc707811203 shutdown only if Multithreading.available;
wenzelm
parents: 28242
diff changeset
   662
    SYNCHRONIZED "shutdown" (fn () =>
51279
f4a2fa9286e9 tuned messages;
wenzelm
parents: 51046
diff changeset
   663
      while scheduler_active () do
59340
734bb148503e explicit shutdown of scheduler and worker thread farm, assuming Session.shutdown() before saving heap image;
wenzelm
parents: 59338
diff changeset
   664
       (do_shutdown := true;
734bb148503e explicit shutdown of scheduler and worker thread farm, assuming Session.shutdown() before saving heap image;
wenzelm
parents: 59338
diff changeset
   665
        Multithreading.tracing 1 (fn () => "SHUTDOWN: wait");
51283
fefd07697979 disallow shutdown from worker, which would lead to deadlock since the scheduler cannot terminate;
wenzelm
parents: 51281
diff changeset
   666
        wait scheduler_event));
28203
88f18387f1c9 shutdown: global join-and-shutdown operation;
wenzelm
parents: 28202
diff changeset
   667
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   668
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   669
(*final declarations of this structure!*)
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   670
val map = map_future;
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   671
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   672
end;
28972
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
   673
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
   674
type 'a future = 'a Future.future;