author | wenzelm |
Sat, 03 Jan 2009 21:45:53 +0100 | |
changeset 29341 | 6bb007a0f9f2 |
parent 29119 | 99941fd0cb0e |
child 29366 | 1ffc8cbf39ec |
permissions | -rw-r--r-- |
28156 | 1 |
(* Title: Pure/Concurrent/future.ML |
2 |
Author: Makarius |
|
3 |
||
28201 | 4 |
Future values. |
5 |
||
6 |
Notes: |
|
7 |
||
8 |
* Futures are similar to delayed evaluation, i.e. delay/force is |
|
9 |
generalized to fork/join (and variants). The idea is to model |
|
10 |
parallel value-oriented computations, but *not* communicating |
|
11 |
processes. |
|
12 |
||
13 |
* Futures are grouped; failure of one group member causes the whole |
|
14 |
group to be interrupted eventually. |
|
15 |
||
16 |
* Forked futures are evaluated spontaneously by a farm of worker |
|
17 |
threads in the background; join resynchronizes the computation and |
|
18 |
delivers results (values or exceptions). |
|
19 |
||
20 |
* The pool of worker threads is limited, usually in correlation with |
|
21 |
the number of physical cores on the machine. Note that allocation |
|
22 |
of runtime resources is distorted either if workers yield CPU time |
|
23 |
(e.g. via system sleep or wait operations), or if non-worker |
|
24 |
threads contend for significant runtime resources independently. |
|
28156 | 25 |
*) |
26 |
||
27 |
signature FUTURE = |
|
28 |
sig |
|
28645 | 29 |
val enabled: unit -> bool |
29119 | 30 |
type task = Task_Queue.task |
31 |
type group = Task_Queue.group |
|
28979
3ce619d8d432
fork/map: no inheritance of group (structure is nested, not parallel);
wenzelm
parents:
28972
diff
changeset
|
32 |
val thread_data: unit -> (string * task) option |
28972 | 33 |
type 'a future |
34 |
val task_of: 'a future -> task |
|
35 |
val group_of: 'a future -> group |
|
36 |
val peek: 'a future -> 'a Exn.result option |
|
37 |
val is_finished: 'a future -> bool |
|
28997 | 38 |
val value: 'a -> 'a future |
28972 | 39 |
val fork: (unit -> 'a) -> 'a future |
28979
3ce619d8d432
fork/map: no inheritance of group (structure is nested, not parallel);
wenzelm
parents:
28972
diff
changeset
|
40 |
val fork_group: group -> (unit -> 'a) -> 'a future |
3ce619d8d432
fork/map: no inheritance of group (structure is nested, not parallel);
wenzelm
parents:
28972
diff
changeset
|
41 |
val fork_deps: 'b future list -> (unit -> 'a) -> 'a future |
29119 | 42 |
val fork_pri: int -> (unit -> 'a) -> 'a future |
28972 | 43 |
val join_results: 'a future list -> 'a Exn.result list |
44 |
val join_result: 'a future -> 'a Exn.result |
|
45 |
val join: 'a future -> 'a |
|
46 |
val map: ('a -> 'b) -> 'a future -> 'b future |
|
28206
bcd48c6897d4
eliminated requests, use global state variables uniformly;
wenzelm
parents:
28203
diff
changeset
|
47 |
val interrupt_task: string -> unit |
28972 | 48 |
val cancel: 'a future -> unit |
28203 | 49 |
val shutdown: unit -> unit |
28156 | 50 |
end; |
51 |
||
52 |
structure Future: FUTURE = |
|
53 |
struct |
|
54 |
||
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
|
55 |
(** 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
|
56 |
|
28645 | 57 |
fun enabled () = |
29118 | 58 |
Multithreading.enabled () andalso |
28645 | 59 |
not (Multithreading.self_critical ()); |
60 |
||
61 |
||
28167 | 62 |
(* identifiers *) |
63 |
||
29119 | 64 |
type task = Task_Queue.task; |
65 |
type group = Task_Queue.group; |
|
28167 | 66 |
|
28979
3ce619d8d432
fork/map: no inheritance of group (structure is nested, not parallel);
wenzelm
parents:
28972
diff
changeset
|
67 |
local val tag = Universal.tag () : (string * task) option Universal.tag in |
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
|
68 |
fun thread_data () = the_default NONE (Thread.getLocal tag); |
28390
0b9fb63b8e1d
proper setmp_thread_data for nested execute (cf. join_loop);
wenzelm
parents:
28386
diff
changeset
|
69 |
fun setmp_thread_data data f x = Library.setmp_thread_data tag (thread_data ()) (SOME data) f x; |
28167 | 70 |
end; |
71 |
||
72 |
||
73 |
(* datatype future *) |
|
74 |
||
28972 | 75 |
datatype 'a future = Future of |
28167 | 76 |
{task: task, |
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
|
77 |
group: group, |
28167 | 78 |
result: 'a Exn.result option ref}; |
79 |
||
80 |
fun task_of (Future {task, ...}) = task; |
|
81 |
fun group_of (Future {group, ...}) = group; |
|
82 |
||
28558 | 83 |
fun peek (Future {result, ...}) = ! result; |
84 |
fun is_finished x = is_some (peek x); |
|
28320 | 85 |
|
28997 | 86 |
fun value x = Future |
29119 | 87 |
{task = Task_Queue.new_task 0, |
88 |
group = Task_Queue.new_group (), |
|
28997 | 89 |
result = ref (SOME (Exn.Result x))}; |
90 |
||
28167 | 91 |
|
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
|
92 |
|
8c0335bc9336
inherit group from running thread, or create a new one -- make it harder to re-use canceled groups;
wenzelm
parents:
28170
diff
changeset
|
93 |
(** 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
|
94 |
|
8c0335bc9336
inherit group from running thread, or create a new one -- make it harder to re-use canceled groups;
wenzelm
parents:
28170
diff
changeset
|
95 |
(* global state *) |
8c0335bc9336
inherit group from running thread, or create a new one -- make it harder to re-use canceled groups;
wenzelm
parents:
28170
diff
changeset
|
96 |
|
29119 | 97 |
val queue = ref Task_Queue.empty; |
28468 | 98 |
val next = ref 0; |
28192 | 99 |
val workers = ref ([]: (Thread.thread * bool) list); |
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
|
100 |
val scheduler = ref (NONE: Thread.thread option); |
8c0335bc9336
inherit group from running thread, or create a new one -- make it harder to re-use canceled groups;
wenzelm
parents:
28170
diff
changeset
|
101 |
val excessive = ref 0; |
29119 | 102 |
val canceled = ref ([]: Task_Queue.group list); |
28206
bcd48c6897d4
eliminated requests, use global state variables uniformly;
wenzelm
parents:
28203
diff
changeset
|
103 |
val do_shutdown = ref false; |
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
|
104 |
|
8c0335bc9336
inherit group from running thread, or create a new one -- make it harder to re-use canceled groups;
wenzelm
parents:
28170
diff
changeset
|
105 |
|
8c0335bc9336
inherit group from running thread, or create a new one -- make it harder to re-use canceled groups;
wenzelm
parents:
28170
diff
changeset
|
106 |
(* synchronization *) |
28156 | 107 |
|
108 |
local |
|
109 |
val lock = Mutex.mutex (); |
|
110 |
val cond = ConditionVar.conditionVar (); |
|
111 |
in |
|
112 |
||
28575 | 113 |
fun SYNCHRONIZED name = SimpleThread.synchronized name lock; |
28156 | 114 |
|
29119 | 115 |
fun wait () = (*requires SYNCHRONIZED*) |
28206
bcd48c6897d4
eliminated requests, use global state variables uniformly;
wenzelm
parents:
28203
diff
changeset
|
116 |
ConditionVar.wait (cond, lock); |
bcd48c6897d4
eliminated requests, use global state variables uniformly;
wenzelm
parents:
28203
diff
changeset
|
117 |
|
29119 | 118 |
fun wait_timeout timeout = (*requires SYNCHRONIZED*) |
29341
6bb007a0f9f2
more reactive scheduler: reduced loop timeout, propagate broadcast interrupt via TaskQueue.cancel_all;
wenzelm
parents:
29119
diff
changeset
|
119 |
ignore (ConditionVar.waitUntil (cond, lock, Time.+ (Time.now (), timeout))); |
28166
43087721a66e
moved task, thread_data, group, queue to task_queue.ML;
wenzelm
parents:
28163
diff
changeset
|
120 |
|
43087721a66e
moved task, thread_data, group, queue to task_queue.ML;
wenzelm
parents:
28163
diff
changeset
|
121 |
fun notify_all () = (*requires SYNCHRONIZED*) |
43087721a66e
moved task, thread_data, group, queue to task_queue.ML;
wenzelm
parents:
28163
diff
changeset
|
122 |
ConditionVar.broadcast cond; |
28156 | 123 |
|
124 |
end; |
|
125 |
||
126 |
||
28382 | 127 |
(* worker activity *) |
128 |
||
129 |
fun trace_active () = |
|
130 |
let |
|
131 |
val ws = ! workers; |
|
132 |
val m = string_of_int (length ws); |
|
133 |
val n = string_of_int (length (filter #2 ws)); |
|
134 |
in Multithreading.tracing 1 (fn () => "SCHEDULE: " ^ m ^ " workers, " ^ n ^ " active") end; |
|
135 |
||
136 |
fun change_active active = (*requires SYNCHRONIZED*) |
|
137 |
change workers (AList.update Thread.equal (Thread.self (), active)); |
|
138 |
||
139 |
||
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
|
140 |
(* execute *) |
28156 | 141 |
|
29341
6bb007a0f9f2
more reactive scheduler: reduced loop timeout, propagate broadcast interrupt via TaskQueue.cancel_all;
wenzelm
parents:
29119
diff
changeset
|
142 |
fun do_cancel group = (*requires SYNCHRONIZED*) |
6bb007a0f9f2
more reactive scheduler: reduced loop timeout, propagate broadcast interrupt via TaskQueue.cancel_all;
wenzelm
parents:
29119
diff
changeset
|
143 |
change canceled (insert Task_Queue.eq_group group); |
6bb007a0f9f2
more reactive scheduler: reduced loop timeout, propagate broadcast interrupt via TaskQueue.cancel_all;
wenzelm
parents:
29119
diff
changeset
|
144 |
|
28167 | 145 |
fun execute name (task, group, run) = |
146 |
let |
|
28382 | 147 |
val _ = trace_active (); |
28979
3ce619d8d432
fork/map: no inheritance of group (structure is nested, not parallel);
wenzelm
parents:
28972
diff
changeset
|
148 |
val ok = setmp_thread_data (name, task) run (); |
28192 | 149 |
val _ = SYNCHRONIZED "execute" (fn () => |
29119 | 150 |
(change queue (Task_Queue.finish task); |
28186 | 151 |
if ok then () |
29119 | 152 |
else if Task_Queue.cancel (! queue) group then () |
29341
6bb007a0f9f2
more reactive scheduler: reduced loop timeout, propagate broadcast interrupt via TaskQueue.cancel_all;
wenzelm
parents:
29119
diff
changeset
|
153 |
else do_cancel group; |
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
|
154 |
notify_all ())); |
28167 | 155 |
in () end; |
156 |
||
157 |
||
158 |
(* worker threads *) |
|
159 |
||
29119 | 160 |
fun worker_wait () = (*requires SYNCHRONIZED*) |
161 |
(change_active false; wait (); change_active true); |
|
28162 | 162 |
|
29119 | 163 |
fun worker_next () = (*requires SYNCHRONIZED*) |
28167 | 164 |
if ! excessive > 0 then |
165 |
(dec excessive; |
|
28192 | 166 |
change workers (filter_out (fn (thread, _) => Thread.equal (thread, Thread.self ()))); |
28203 | 167 |
notify_all (); |
28167 | 168 |
NONE) |
28166
43087721a66e
moved task, thread_data, group, queue to task_queue.ML;
wenzelm
parents:
28163
diff
changeset
|
169 |
else |
29119 | 170 |
(case change_result queue Task_Queue.dequeue of |
171 |
NONE => (worker_wait (); worker_next ()) |
|
28166
43087721a66e
moved task, thread_data, group, queue to task_queue.ML;
wenzelm
parents:
28163
diff
changeset
|
172 |
| some => some); |
28156 | 173 |
|
28167 | 174 |
fun worker_loop name = |
29119 | 175 |
(case SYNCHRONIZED name worker_next of |
176 |
NONE => () |
|
28167 | 177 |
| SOME work => (execute name work; worker_loop name)); |
28156 | 178 |
|
28167 | 179 |
fun worker_start name = (*requires SYNCHRONIZED*) |
28242 | 180 |
change workers (cons (SimpleThread.fork false (fn () => worker_loop name), true)); |
28156 | 181 |
|
182 |
||
183 |
(* scheduler *) |
|
184 |
||
28206
bcd48c6897d4
eliminated requests, use global state variables uniformly;
wenzelm
parents:
28203
diff
changeset
|
185 |
fun scheduler_next () = (*requires SYNCHRONIZED*) |
28156 | 186 |
let |
28206
bcd48c6897d4
eliminated requests, use global state variables uniformly;
wenzelm
parents:
28203
diff
changeset
|
187 |
(*worker threads*) |
28191 | 188 |
val _ = |
28192 | 189 |
(case List.partition (Thread.isActive o #1) (! workers) of |
28191 | 190 |
(_, []) => () |
191 |
| (active, inactive) => |
|
192 |
(workers := active; Multithreading.tracing 0 (fn () => |
|
28192 | 193 |
"SCHEDULE: disposed " ^ string_of_int (length inactive) ^ " dead worker threads"))); |
28382 | 194 |
val _ = trace_active (); |
28191 | 195 |
|
28206
bcd48c6897d4
eliminated requests, use global state variables uniformly;
wenzelm
parents:
28203
diff
changeset
|
196 |
val m = if ! do_shutdown then 0 else Multithreading.max_threads_value (); |
28167 | 197 |
val l = length (! workers); |
198 |
val _ = excessive := l - m; |
|
28203 | 199 |
val _ = |
28468 | 200 |
if m > l then funpow (m - l) (fn () => worker_start ("worker " ^ string_of_int (inc next))) () |
28203 | 201 |
else (); |
28206
bcd48c6897d4
eliminated requests, use global state variables uniformly;
wenzelm
parents:
28203
diff
changeset
|
202 |
|
bcd48c6897d4
eliminated requests, use global state variables uniformly;
wenzelm
parents:
28203
diff
changeset
|
203 |
(*canceled groups*) |
29119 | 204 |
val _ = change canceled (filter_out (Task_Queue.cancel (! queue))); |
28206
bcd48c6897d4
eliminated requests, use global state variables uniformly;
wenzelm
parents:
28203
diff
changeset
|
205 |
|
bcd48c6897d4
eliminated requests, use global state variables uniformly;
wenzelm
parents:
28203
diff
changeset
|
206 |
(*shutdown*) |
bcd48c6897d4
eliminated requests, use global state variables uniformly;
wenzelm
parents:
28203
diff
changeset
|
207 |
val continue = not (! do_shutdown andalso null (! workers)); |
bcd48c6897d4
eliminated requests, use global state variables uniformly;
wenzelm
parents:
28203
diff
changeset
|
208 |
val _ = if continue then () else scheduler := NONE; |
28167 | 209 |
|
28206
bcd48c6897d4
eliminated requests, use global state variables uniformly;
wenzelm
parents:
28203
diff
changeset
|
210 |
val _ = notify_all (); |
29341
6bb007a0f9f2
more reactive scheduler: reduced loop timeout, propagate broadcast interrupt via TaskQueue.cancel_all;
wenzelm
parents:
29119
diff
changeset
|
211 |
val _ = interruptible (fn () => wait_timeout (Time.fromSeconds 1)) () |
6bb007a0f9f2
more reactive scheduler: reduced loop timeout, propagate broadcast interrupt via TaskQueue.cancel_all;
wenzelm
parents:
29119
diff
changeset
|
212 |
handle Exn.Interrupt => List.app do_cancel (Task_Queue.cancel_all (! queue)); |
28206
bcd48c6897d4
eliminated requests, use global state variables uniformly;
wenzelm
parents:
28203
diff
changeset
|
213 |
in continue end; |
bcd48c6897d4
eliminated requests, use global state variables uniformly;
wenzelm
parents:
28203
diff
changeset
|
214 |
|
bcd48c6897d4
eliminated requests, use global state variables uniformly;
wenzelm
parents:
28203
diff
changeset
|
215 |
fun scheduler_loop () = |
29119 | 216 |
while SYNCHRONIZED "scheduler" scheduler_next do (); |
28156 | 217 |
|
28203 | 218 |
fun scheduler_active () = (*requires SYNCHRONIZED*) |
219 |
(case ! scheduler of NONE => false | SOME thread => Thread.isActive thread); |
|
220 |
||
28464 | 221 |
fun scheduler_check name = SYNCHRONIZED name (fn () => |
28206
bcd48c6897d4
eliminated requests, use global state variables uniformly;
wenzelm
parents:
28203
diff
changeset
|
222 |
if not (scheduler_active ()) then |
29119 | 223 |
(do_shutdown := false; scheduler := SOME (SimpleThread.fork false scheduler_loop)) |
28206
bcd48c6897d4
eliminated requests, use global state variables uniformly;
wenzelm
parents:
28203
diff
changeset
|
224 |
else if ! do_shutdown then error "Scheduler shutdown in progress" |
bcd48c6897d4
eliminated requests, use global state variables uniformly;
wenzelm
parents:
28203
diff
changeset
|
225 |
else ()); |
28156 | 226 |
|
227 |
||
28191 | 228 |
(* future values: fork independent computation *) |
28156 | 229 |
|
28304
4b0477452943
future tasks: support boolean priorities (true = high, false = low/irrelevant);
wenzelm
parents:
28276
diff
changeset
|
230 |
fun future opt_group deps pri (e: unit -> 'a) = |
28156 | 231 |
let |
28464 | 232 |
val _ = scheduler_check "future check"; |
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
|
233 |
|
29119 | 234 |
val group = (case opt_group of SOME group => group | NONE => Task_Queue.new_group ()); |
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
|
235 |
|
28166
43087721a66e
moved task, thread_data, group, queue to task_queue.ML;
wenzelm
parents:
28163
diff
changeset
|
236 |
val result = ref (NONE: 'a Exn.result option); |
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
|
237 |
val run = Multithreading.with_attributes (Thread.getAttributes ()) |
8c0335bc9336
inherit group from running thread, or create a new one -- make it harder to re-use canceled groups;
wenzelm
parents:
28170
diff
changeset
|
238 |
(fn _ => fn ok => |
28532 | 239 |
let |
240 |
val res = if ok then Exn.capture e () else Exn.Exn Exn.Interrupt; |
|
28548
003f52c2bb8f
future result: Interrupt invalidates group, but pretends success otherwise;
wenzelm
parents:
28534
diff
changeset
|
241 |
val _ = result := SOME res; |
28532 | 242 |
val res_ok = |
243 |
(case res of |
|
244 |
Exn.Result _ => true |
|
29119 | 245 |
| Exn.Exn Exn.Interrupt => (Task_Queue.invalidate_group group; true) |
28532 | 246 |
| _ => false); |
28548
003f52c2bb8f
future result: Interrupt invalidates group, but pretends success otherwise;
wenzelm
parents:
28534
diff
changeset
|
247 |
in res_ok end); |
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
|
248 |
|
28192 | 249 |
val task = SYNCHRONIZED "future" (fn () => |
29119 | 250 |
change_result queue (Task_Queue.enqueue group deps pri run) before notify_all ()); |
28166
43087721a66e
moved task, thread_data, group, queue to task_queue.ML;
wenzelm
parents:
28163
diff
changeset
|
251 |
in Future {task = task, group = group, result = result} end; |
28162 | 252 |
|
29119 | 253 |
fun fork e = future NONE [] 0 e; |
254 |
fun fork_group group e = future (SOME group) [] 0 e; |
|
255 |
fun fork_deps deps e = future NONE (map task_of deps) 0 e; |
|
256 |
fun fork_pri pri e = future NONE [] pri e; |
|
28186 | 257 |
|
258 |
||
28191 | 259 |
(* join: retrieve results *) |
28186 | 260 |
|
28331
33d58fdc177d
join_results: special case for empty list, works without multithreading;
wenzelm
parents:
28320
diff
changeset
|
261 |
fun join_results [] = [] |
28532 | 262 |
| join_results xs = uninterruptible (fn _ => fn () => |
28331
33d58fdc177d
join_results: special case for empty list, works without multithreading;
wenzelm
parents:
28320
diff
changeset
|
263 |
let |
28464 | 264 |
val _ = scheduler_check "join check"; |
28331
33d58fdc177d
join_results: special case for empty list, works without multithreading;
wenzelm
parents:
28320
diff
changeset
|
265 |
val _ = Multithreading.self_critical () andalso |
28647
8068cdc84e7e
join_results: allow CRITICAL join of finished futures;
wenzelm
parents:
28645
diff
changeset
|
266 |
exists (not o is_finished) xs andalso |
28331
33d58fdc177d
join_results: special case for empty list, works without multithreading;
wenzelm
parents:
28320
diff
changeset
|
267 |
error "Cannot join future values within critical section"; |
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
|
268 |
|
28386 | 269 |
fun join_loop _ [] = () |
270 |
| join_loop name tasks = |
|
271 |
(case SYNCHRONIZED name (fn () => |
|
29119 | 272 |
change_result queue (Task_Queue.dequeue_towards tasks)) of |
28382 | 273 |
NONE => () |
28386 | 274 |
| SOME (work, tasks') => (execute name work; join_loop name tasks')); |
28331
33d58fdc177d
join_results: special case for empty list, works without multithreading;
wenzelm
parents:
28320
diff
changeset
|
275 |
val _ = |
33d58fdc177d
join_results: special case for empty list, works without multithreading;
wenzelm
parents:
28320
diff
changeset
|
276 |
(case thread_data () of |
28382 | 277 |
NONE => |
278 |
(*alien thread -- refrain from contending for resources*) |
|
279 |
while exists (not o is_finished) xs |
|
29119 | 280 |
do SYNCHRONIZED "join_thread" (fn () => wait ()) |
28979
3ce619d8d432
fork/map: no inheritance of group (structure is nested, not parallel);
wenzelm
parents:
28972
diff
changeset
|
281 |
| SOME (name, task) => |
28382 | 282 |
(*proper task -- actively work towards results*) |
283 |
let |
|
284 |
val unfinished = xs |> map_filter |
|
285 |
(fn Future {task, result = ref NONE, ...} => SOME task | _ => NONE); |
|
286 |
val _ = SYNCHRONIZED "join" (fn () => |
|
29119 | 287 |
(change queue (Task_Queue.depend unfinished task); notify_all ())); |
28386 | 288 |
val _ = join_loop ("join_loop: " ^ name) unfinished; |
28382 | 289 |
val _ = |
290 |
while exists (not o is_finished) xs |
|
29119 | 291 |
do SYNCHRONIZED "join_task" (fn () => worker_wait ()); |
28382 | 292 |
in () end); |
28186 | 293 |
|
28532 | 294 |
in xs |> map (fn Future {result = ref (SOME res), ...} => res) end) (); |
28186 | 295 |
|
28647
8068cdc84e7e
join_results: allow CRITICAL join of finished futures;
wenzelm
parents:
28645
diff
changeset
|
296 |
fun join_result x = singleton join_results x; |
8068cdc84e7e
join_results: allow CRITICAL join of finished futures;
wenzelm
parents:
28645
diff
changeset
|
297 |
fun join x = Exn.release (join_result x); |
28156 | 298 |
|
29119 | 299 |
fun map f x = |
300 |
let val task = task_of x |
|
301 |
in future NONE [task] (Task_Queue.pri_of_task task) (fn () => f (join x)) end; |
|
28979
3ce619d8d432
fork/map: no inheritance of group (structure is nested, not parallel);
wenzelm
parents:
28972
diff
changeset
|
302 |
|
28191 | 303 |
|
28202
23cb9a974630
added focus, which indicates a particular collection of high-priority tasks;
wenzelm
parents:
28201
diff
changeset
|
304 |
(* misc operations *) |
23cb9a974630
added focus, which indicates a particular collection of high-priority tasks;
wenzelm
parents:
28201
diff
changeset
|
305 |
|
23cb9a974630
added focus, which indicates a particular collection of high-priority tasks;
wenzelm
parents:
28201
diff
changeset
|
306 |
(*interrupt: permissive signal, may get ignored*) |
28197 | 307 |
fun interrupt_task id = SYNCHRONIZED "interrupt" |
29119 | 308 |
(fn () => Task_Queue.interrupt_external (! queue) id); |
28191 | 309 |
|
28206
bcd48c6897d4
eliminated requests, use global state variables uniformly;
wenzelm
parents:
28203
diff
changeset
|
310 |
(*cancel: present and future group members will be interrupted eventually*) |
bcd48c6897d4
eliminated requests, use global state variables uniformly;
wenzelm
parents:
28203
diff
changeset
|
311 |
fun cancel x = |
28464 | 312 |
(scheduler_check "cancel check"; |
29341
6bb007a0f9f2
more reactive scheduler: reduced loop timeout, propagate broadcast interrupt via TaskQueue.cancel_all;
wenzelm
parents:
29119
diff
changeset
|
313 |
SYNCHRONIZED "cancel" (fn () => (do_cancel (group_of x); notify_all ()))); |
28206
bcd48c6897d4
eliminated requests, use global state variables uniformly;
wenzelm
parents:
28203
diff
changeset
|
314 |
|
bcd48c6897d4
eliminated requests, use global state variables uniformly;
wenzelm
parents:
28203
diff
changeset
|
315 |
|
28203 | 316 |
(*global join and shutdown*) |
317 |
fun shutdown () = |
|
28276 | 318 |
if Multithreading.available then |
28464 | 319 |
(scheduler_check "shutdown check"; |
28276 | 320 |
SYNCHRONIZED "shutdown" (fn () => |
29119 | 321 |
(while not (scheduler_active ()) do wait (); |
322 |
while not (Task_Queue.is_empty (! queue)) do wait (); |
|
28276 | 323 |
do_shutdown := true; |
324 |
notify_all (); |
|
29119 | 325 |
while not (null (! workers)) do wait (); |
326 |
while scheduler_active () do wait (); |
|
28470 | 327 |
OS.Process.sleep (Time.fromMilliseconds 300)))) |
28276 | 328 |
else (); |
28203 | 329 |
|
28156 | 330 |
end; |
28972 | 331 |
|
332 |
type 'a future = 'a Future.future; |
|
333 |