author | wenzelm |
Mon, 27 Jul 2009 12:11:18 +0200 | |
changeset 32219 | 9a2566d1fdbd |
parent 32186 | 8026b73cd357 |
child 32220 | 01ff6781dd18 |
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 |
|
32058 | 32 |
val is_worker: unit -> bool |
32102 | 33 |
val worker_group: unit -> Task_Queue.group option |
28972 | 34 |
type 'a future |
35 |
val task_of: 'a future -> task |
|
36 |
val group_of: 'a future -> group |
|
37 |
val peek: 'a future -> 'a Exn.result option |
|
38 |
val is_finished: 'a future -> bool |
|
28997 | 39 |
val value: 'a -> 'a future |
28972 | 40 |
val fork: (unit -> 'a) -> 'a future |
28979
3ce619d8d432
fork/map: no inheritance of group (structure is nested, not parallel);
wenzelm
parents:
28972
diff
changeset
|
41 |
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
|
42 |
val fork_deps: 'b future list -> (unit -> 'a) -> 'a future |
29119 | 43 |
val fork_pri: int -> (unit -> 'a) -> 'a future |
28972 | 44 |
val join_results: 'a future list -> 'a Exn.result list |
45 |
val join_result: 'a future -> 'a Exn.result |
|
46 |
val join: 'a future -> 'a |
|
47 |
val map: ('a -> 'b) -> 'a future -> 'b future |
|
30618
046f4f986fb5
restricted interrupts for tasks running as future worker thread -- attempt to prevent interrupt race conditions;
wenzelm
parents:
30612
diff
changeset
|
48 |
val interruptible_task: ('a -> 'b) -> 'a -> 'b |
28206
bcd48c6897d4
eliminated requests, use global state variables uniformly;
wenzelm
parents:
28203
diff
changeset
|
49 |
val interrupt_task: string -> unit |
29431 | 50 |
val cancel_group: group -> unit |
28972 | 51 |
val cancel: 'a future -> unit |
28203 | 52 |
val shutdown: unit -> unit |
28156 | 53 |
end; |
54 |
||
55 |
structure Future: FUTURE = |
|
56 |
struct |
|
57 |
||
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
|
58 |
(** 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
|
59 |
|
28645 | 60 |
fun enabled () = |
29118 | 61 |
Multithreading.enabled () andalso |
28645 | 62 |
not (Multithreading.self_critical ()); |
63 |
||
64 |
||
28167 | 65 |
(* identifiers *) |
66 |
||
29119 | 67 |
type task = Task_Queue.task; |
68 |
type group = Task_Queue.group; |
|
28167 | 69 |
|
32058 | 70 |
local |
71 |
val tag = Universal.tag () : (string * task * group) option Universal.tag; |
|
72 |
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
|
73 |
fun thread_data () = the_default NONE (Thread.getLocal tag); |
32058 | 74 |
fun setmp_thread_data data f x = |
75 |
Library.setmp_thread_data tag (thread_data ()) (SOME data) f x; |
|
28167 | 76 |
end; |
77 |
||
32058 | 78 |
val is_worker = is_some o thread_data; |
32102 | 79 |
val worker_group = Option.map #3 o thread_data; |
32058 | 80 |
|
28167 | 81 |
|
82 |
(* datatype future *) |
|
83 |
||
28972 | 84 |
datatype 'a future = Future of |
28167 | 85 |
{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
|
86 |
group: group, |
28167 | 87 |
result: 'a Exn.result option ref}; |
88 |
||
89 |
fun task_of (Future {task, ...}) = task; |
|
90 |
fun group_of (Future {group, ...}) = group; |
|
91 |
||
28558 | 92 |
fun peek (Future {result, ...}) = ! result; |
93 |
fun is_finished x = is_some (peek x); |
|
28320 | 94 |
|
28997 | 95 |
fun value x = Future |
29119 | 96 |
{task = Task_Queue.new_task 0, |
32102 | 97 |
group = Task_Queue.new_group NONE, |
28997 | 98 |
result = ref (SOME (Exn.Result x))}; |
99 |
||
28167 | 100 |
|
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
|
101 |
|
8c0335bc9336
inherit group from running thread, or create a new one -- make it harder to re-use canceled groups;
wenzelm
parents:
28170
diff
changeset
|
102 |
(** 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
|
103 |
|
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 |
(* 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
|
105 |
|
29119 | 106 |
val queue = ref Task_Queue.empty; |
28468 | 107 |
val next = ref 0; |
28192 | 108 |
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
|
109 |
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
|
110 |
val excessive = ref 0; |
29119 | 111 |
val canceled = ref ([]: Task_Queue.group list); |
28206
bcd48c6897d4
eliminated requests, use global state variables uniformly;
wenzelm
parents:
28203
diff
changeset
|
112 |
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
|
113 |
|
8c0335bc9336
inherit group from running thread, or create a new one -- make it harder to re-use canceled groups;
wenzelm
parents:
28170
diff
changeset
|
114 |
|
8c0335bc9336
inherit group from running thread, or create a new one -- make it harder to re-use canceled groups;
wenzelm
parents:
28170
diff
changeset
|
115 |
(* synchronization *) |
28156 | 116 |
|
32219
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
117 |
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
|
118 |
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
|
119 |
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
|
120 |
|
28156 | 121 |
local |
122 |
val lock = Mutex.mutex (); |
|
123 |
in |
|
124 |
||
28575 | 125 |
fun SYNCHRONIZED name = SimpleThread.synchronized name lock; |
28156 | 126 |
|
32219
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
127 |
fun wait cond = (*requires SYNCHRONIZED*) |
28206
bcd48c6897d4
eliminated requests, use global state variables uniformly;
wenzelm
parents:
28203
diff
changeset
|
128 |
ConditionVar.wait (cond, lock); |
bcd48c6897d4
eliminated requests, use global state variables uniformly;
wenzelm
parents:
28203
diff
changeset
|
129 |
|
32219
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
130 |
fun wait_timeout cond timeout = (*requires SYNCHRONIZED*) |
29341
6bb007a0f9f2
more reactive scheduler: reduced loop timeout, propagate broadcast interrupt via TaskQueue.cancel_all;
wenzelm
parents:
29119
diff
changeset
|
131 |
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
|
132 |
|
32219
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
133 |
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
|
134 |
ConditionVar.signal cond; |
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
135 |
|
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
136 |
fun broadcast cond = (*requires SYNCHRONIZED*) |
28166
43087721a66e
moved task, thread_data, group, queue to task_queue.ML;
wenzelm
parents:
28163
diff
changeset
|
137 |
ConditionVar.broadcast cond; |
28156 | 138 |
|
139 |
end; |
|
140 |
||
141 |
||
28382 | 142 |
(* worker activity *) |
143 |
||
32095 | 144 |
fun count_active ws = |
145 |
fold (fn (_, active) => fn i => if active then i + 1 else i) ws 0; |
|
146 |
||
32186 | 147 |
fun trace_active () = Multithreading.tracing 6 (fn () => |
28382 | 148 |
let |
149 |
val ws = ! workers; |
|
150 |
val m = string_of_int (length ws); |
|
32095 | 151 |
val n = string_of_int (count_active ws); |
152 |
in "SCHEDULE: " ^ m ^ " workers, " ^ n ^ " active" end); |
|
28382 | 153 |
|
154 |
fun change_active active = (*requires SYNCHRONIZED*) |
|
155 |
change workers (AList.update Thread.equal (Thread.self (), active)); |
|
156 |
||
32095 | 157 |
fun overloaded () = |
158 |
count_active (! workers) > Multithreading.max_threads_value (); |
|
159 |
||
28382 | 160 |
|
32099 | 161 |
(* execute future jobs *) |
162 |
||
163 |
fun future_job group (e: unit -> 'a) = |
|
164 |
let |
|
165 |
val result = ref (NONE: 'a Exn.result option); |
|
32107
47d0da617fcc
future_job: tight scope for interrupts, to prevent shooting ourselves in the foot via cancel_group;
wenzelm
parents:
32102
diff
changeset
|
166 |
fun job ok = |
47d0da617fcc
future_job: tight scope for interrupts, to prevent shooting ourselves in the foot via cancel_group;
wenzelm
parents:
32102
diff
changeset
|
167 |
let |
47d0da617fcc
future_job: tight scope for interrupts, to prevent shooting ourselves in the foot via cancel_group;
wenzelm
parents:
32102
diff
changeset
|
168 |
val res = |
47d0da617fcc
future_job: tight scope for interrupts, to prevent shooting ourselves in the foot via cancel_group;
wenzelm
parents:
32102
diff
changeset
|
169 |
if ok then |
32109
ff3677972e3f
future_job: more robust Exn.capture outside thread attribute change;
wenzelm
parents:
32107
diff
changeset
|
170 |
Exn.capture |
ff3677972e3f
future_job: more robust Exn.capture outside thread attribute change;
wenzelm
parents:
32107
diff
changeset
|
171 |
(Multithreading.with_attributes Multithreading.restricted_interrupts |
ff3677972e3f
future_job: more robust Exn.capture outside thread attribute change;
wenzelm
parents:
32107
diff
changeset
|
172 |
(fn _ => fn () => e ())) () |
32107
47d0da617fcc
future_job: tight scope for interrupts, to prevent shooting ourselves in the foot via cancel_group;
wenzelm
parents:
32102
diff
changeset
|
173 |
else Exn.Exn Exn.Interrupt; |
47d0da617fcc
future_job: tight scope for interrupts, to prevent shooting ourselves in the foot via cancel_group;
wenzelm
parents:
32102
diff
changeset
|
174 |
val _ = result := SOME res; |
47d0da617fcc
future_job: tight scope for interrupts, to prevent shooting ourselves in the foot via cancel_group;
wenzelm
parents:
32102
diff
changeset
|
175 |
in |
47d0da617fcc
future_job: tight scope for interrupts, to prevent shooting ourselves in the foot via cancel_group;
wenzelm
parents:
32102
diff
changeset
|
176 |
(case res of |
47d0da617fcc
future_job: tight scope for interrupts, to prevent shooting ourselves in the foot via cancel_group;
wenzelm
parents:
32102
diff
changeset
|
177 |
Exn.Exn exn => (Task_Queue.cancel_group group exn; false) |
47d0da617fcc
future_job: tight scope for interrupts, to prevent shooting ourselves in the foot via cancel_group;
wenzelm
parents:
32102
diff
changeset
|
178 |
| Exn.Result _ => true) |
47d0da617fcc
future_job: tight scope for interrupts, to prevent shooting ourselves in the foot via cancel_group;
wenzelm
parents:
32102
diff
changeset
|
179 |
end; |
32099 | 180 |
in (result, job) end; |
28156 | 181 |
|
29341
6bb007a0f9f2
more reactive scheduler: reduced loop timeout, propagate broadcast interrupt via TaskQueue.cancel_all;
wenzelm
parents:
29119
diff
changeset
|
182 |
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
|
183 |
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
|
184 |
|
29366 | 185 |
fun execute name (task, group, jobs) = |
28167 | 186 |
let |
28382 | 187 |
val _ = trace_active (); |
32102 | 188 |
val valid = not (Task_Queue.is_canceled group); |
32058 | 189 |
val ok = setmp_thread_data (name, task, group) (fn () => |
29384
a3c7e9ae9b71
more robust propagation of errors through bulk jobs;
wenzelm
parents:
29366
diff
changeset
|
190 |
fold (fn job => fn ok => job valid andalso ok) jobs true) (); |
28192 | 191 |
val _ = SYNCHRONIZED "execute" (fn () => |
32219
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
192 |
let |
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
193 |
val maximal = change_result queue (Task_Queue.finish task); |
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
194 |
val _ = |
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
195 |
if ok then () |
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
196 |
else if Task_Queue.cancel (! queue) group then () |
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
197 |
else do_cancel group; |
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
198 |
val _ = broadcast work_finished; |
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
199 |
val _ = if maximal then () else broadcast work_available; |
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
200 |
in () end); |
28167 | 201 |
in () end; |
202 |
||
203 |
||
204 |
(* worker threads *) |
|
205 |
||
32219
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
206 |
fun worker_wait cond = (*requires SYNCHRONIZED*) |
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
207 |
(change_active false; broadcast scheduler_event; |
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
208 |
wait cond; |
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
209 |
change_active true; broadcast scheduler_event); |
28162 | 210 |
|
29119 | 211 |
fun worker_next () = (*requires SYNCHRONIZED*) |
28167 | 212 |
if ! excessive > 0 then |
213 |
(dec excessive; |
|
28192 | 214 |
change workers (filter_out (fn (thread, _) => Thread.equal (thread, Thread.self ()))); |
32219
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
215 |
broadcast scheduler_event; |
28167 | 216 |
NONE) |
32219
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
217 |
else if overloaded () then (worker_wait scheduler_event; worker_next ()) |
28166
43087721a66e
moved task, thread_data, group, queue to task_queue.ML;
wenzelm
parents:
28163
diff
changeset
|
218 |
else |
29119 | 219 |
(case change_result queue Task_Queue.dequeue of |
32219
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
220 |
NONE => (worker_wait work_available; worker_next ()) |
28166
43087721a66e
moved task, thread_data, group, queue to task_queue.ML;
wenzelm
parents:
28163
diff
changeset
|
221 |
| some => some); |
28156 | 222 |
|
28167 | 223 |
fun worker_loop name = |
32107
47d0da617fcc
future_job: tight scope for interrupts, to prevent shooting ourselves in the foot via cancel_group;
wenzelm
parents:
32102
diff
changeset
|
224 |
(case SYNCHRONIZED name (fn () => worker_next ()) of |
29119 | 225 |
NONE => () |
28167 | 226 |
| SOME work => (execute name work; worker_loop name)); |
28156 | 227 |
|
28167 | 228 |
fun worker_start name = (*requires SYNCHRONIZED*) |
28242 | 229 |
change workers (cons (SimpleThread.fork false (fn () => worker_loop name), true)); |
28156 | 230 |
|
231 |
||
232 |
(* scheduler *) |
|
233 |
||
28206
bcd48c6897d4
eliminated requests, use global state variables uniformly;
wenzelm
parents:
28203
diff
changeset
|
234 |
fun scheduler_next () = (*requires SYNCHRONIZED*) |
28156 | 235 |
let |
32053 | 236 |
(*queue status*) |
32186 | 237 |
val _ = Multithreading.tracing 6 (fn () => |
32053 | 238 |
let val {ready, pending, running} = Task_Queue.status (! queue) in |
239 |
"SCHEDULE: " ^ |
|
240 |
string_of_int ready ^ " ready, " ^ |
|
241 |
string_of_int pending ^ " pending, " ^ |
|
242 |
string_of_int running ^ " running" |
|
243 |
end); |
|
244 |
||
28206
bcd48c6897d4
eliminated requests, use global state variables uniformly;
wenzelm
parents:
28203
diff
changeset
|
245 |
(*worker threads*) |
28191 | 246 |
val _ = |
32219
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
247 |
if forall (Thread.isActive o #1) (! workers) then () |
32095 | 248 |
else |
32219
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
249 |
(case List.partition (Thread.isActive o #1) (! workers) of |
32095 | 250 |
(_, []) => () |
251 |
| (active, inactive) => |
|
252 |
(workers := active; Multithreading.tracing 0 (fn () => |
|
253 |
"SCHEDULE: disposed " ^ string_of_int (length inactive) ^ " dead worker threads"))); |
|
28382 | 254 |
val _ = trace_active (); |
28191 | 255 |
|
28206
bcd48c6897d4
eliminated requests, use global state variables uniformly;
wenzelm
parents:
28203
diff
changeset
|
256 |
val m = if ! do_shutdown then 0 else Multithreading.max_threads_value (); |
32095 | 257 |
val mm = (m * 3) div 2; |
32219
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
258 |
val l = length (! workers); |
32095 | 259 |
val _ = excessive := l - mm; |
28203 | 260 |
val _ = |
32095 | 261 |
if mm > l then |
32219
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
262 |
(funpow (mm - l) (fn () => worker_start ("worker " ^ string_of_int (inc next))) (); |
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
263 |
broadcast scheduler_event) |
28203 | 264 |
else (); |
28206
bcd48c6897d4
eliminated requests, use global state variables uniformly;
wenzelm
parents:
28203
diff
changeset
|
265 |
|
bcd48c6897d4
eliminated requests, use global state variables uniformly;
wenzelm
parents:
28203
diff
changeset
|
266 |
(*canceled groups*) |
32095 | 267 |
val _ = change canceled (filter_out (Task_Queue.cancel (! queue))); |
28206
bcd48c6897d4
eliminated requests, use global state variables uniformly;
wenzelm
parents:
28203
diff
changeset
|
268 |
|
32219
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
269 |
val timeout = |
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
270 |
Time.fromMilliseconds (if not (! do_shutdown) andalso null (! canceled) then 500 else 50); |
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
271 |
val _ = interruptible (fn () => wait_timeout scheduler_event timeout) () |
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
272 |
handle Exn.Interrupt => List.app do_cancel (Task_Queue.cancel_all (! queue)); |
28167 | 273 |
|
32219
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
274 |
(*shutdown*) |
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
275 |
val continue = not (! do_shutdown andalso null (! workers)); |
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
276 |
val _ = if continue then () else scheduler := NONE; |
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
277 |
val _ = broadcast scheduler_event; |
28206
bcd48c6897d4
eliminated requests, use global state variables uniformly;
wenzelm
parents:
28203
diff
changeset
|
278 |
in continue end; |
bcd48c6897d4
eliminated requests, use global state variables uniformly;
wenzelm
parents:
28203
diff
changeset
|
279 |
|
bcd48c6897d4
eliminated requests, use global state variables uniformly;
wenzelm
parents:
28203
diff
changeset
|
280 |
fun scheduler_loop () = |
32107
47d0da617fcc
future_job: tight scope for interrupts, to prevent shooting ourselves in the foot via cancel_group;
wenzelm
parents:
32102
diff
changeset
|
281 |
while SYNCHRONIZED "scheduler" (fn () => scheduler_next ()) do (); |
28156 | 282 |
|
28203 | 283 |
fun scheduler_active () = (*requires SYNCHRONIZED*) |
284 |
(case ! scheduler of NONE => false | SOME thread => Thread.isActive thread); |
|
285 |
||
28464 | 286 |
fun scheduler_check name = SYNCHRONIZED name (fn () => |
28206
bcd48c6897d4
eliminated requests, use global state variables uniformly;
wenzelm
parents:
28203
diff
changeset
|
287 |
if not (scheduler_active ()) then |
32219
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
288 |
(do_shutdown := false; scheduler := SOME (SimpleThread.fork false scheduler_loop); |
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
289 |
broadcast scheduler_event) |
28206
bcd48c6897d4
eliminated requests, use global state variables uniformly;
wenzelm
parents:
28203
diff
changeset
|
290 |
else if ! do_shutdown then error "Scheduler shutdown in progress" |
bcd48c6897d4
eliminated requests, use global state variables uniformly;
wenzelm
parents:
28203
diff
changeset
|
291 |
else ()); |
28156 | 292 |
|
293 |
||
29366 | 294 |
|
295 |
(** futures **) |
|
28156 | 296 |
|
29366 | 297 |
(* fork *) |
298 |
||
299 |
fun fork_future opt_group deps pri e = |
|
300 |
let |
|
301 |
val _ = scheduler_check "future check"; |
|
302 |
||
32102 | 303 |
val group = |
304 |
(case opt_group of |
|
305 |
SOME group => group |
|
306 |
| NONE => Task_Queue.new_group (worker_group ())); |
|
29366 | 307 |
val (result, job) = future_job group e; |
28192 | 308 |
val task = SYNCHRONIZED "future" (fn () => |
32219
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
309 |
let |
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
310 |
val (task, minimal) = change_result queue (Task_Queue.enqueue group deps pri job); |
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
311 |
val _ = if minimal then signal work_available else (); |
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
312 |
in task end); |
28166
43087721a66e
moved task, thread_data, group, queue to task_queue.ML;
wenzelm
parents:
28163
diff
changeset
|
313 |
in Future {task = task, group = group, result = result} end; |
28162 | 314 |
|
29366 | 315 |
fun fork e = fork_future NONE [] 0 e; |
316 |
fun fork_group group e = fork_future (SOME group) [] 0 e; |
|
317 |
fun fork_deps deps e = fork_future NONE (map task_of deps) 0 e; |
|
318 |
fun fork_pri pri e = fork_future NONE [] pri e; |
|
28186 | 319 |
|
320 |
||
29366 | 321 |
(* join *) |
322 |
||
29551
95e469919c3e
join_results: when dependencies are resulved (but not finished yet),
wenzelm
parents:
29431
diff
changeset
|
323 |
local |
95e469919c3e
join_results: when dependencies are resulved (but not finished yet),
wenzelm
parents:
29431
diff
changeset
|
324 |
|
32099 | 325 |
fun get_result x = |
326 |
(case peek x of |
|
32102 | 327 |
NONE => Exn.Exn (SYS_ERROR "unfinished future") |
328 |
| SOME (Exn.Exn Exn.Interrupt) => |
|
329 |
Exn.Exn (Exn.EXCEPTIONS (Exn.flatten_list (Task_Queue.group_status (group_of x)))) |
|
330 |
| SOME res => res); |
|
28186 | 331 |
|
32095 | 332 |
fun join_next deps = (*requires SYNCHRONIZED*) |
32219
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
333 |
if overloaded () then (worker_wait scheduler_event; join_next deps) |
32095 | 334 |
else change_result queue (Task_Queue.dequeue_towards deps); |
335 |
||
32055
6a46898aa805
recovered a version of dequeue_towards (cf. bb7b5a5942c7);
wenzelm
parents:
32053
diff
changeset
|
336 |
fun join_deps deps = |
32095 | 337 |
(case SYNCHRONIZED "join" (fn () => join_next deps) of |
29551
95e469919c3e
join_results: when dependencies are resulved (but not finished yet),
wenzelm
parents:
29431
diff
changeset
|
338 |
NONE => () |
32055
6a46898aa805
recovered a version of dequeue_towards (cf. bb7b5a5942c7);
wenzelm
parents:
32053
diff
changeset
|
339 |
| SOME (work, deps') => (execute "join" work; join_deps deps')); |
29551
95e469919c3e
join_results: when dependencies are resulved (but not finished yet),
wenzelm
parents:
29431
diff
changeset
|
340 |
|
95e469919c3e
join_results: when dependencies are resulved (but not finished yet),
wenzelm
parents:
29431
diff
changeset
|
341 |
in |
95e469919c3e
join_results: when dependencies are resulved (but not finished yet),
wenzelm
parents:
29431
diff
changeset
|
342 |
|
29366 | 343 |
fun join_results xs = |
344 |
if forall is_finished xs then map get_result xs |
|
345 |
else uninterruptible (fn _ => fn () => |
|
346 |
let |
|
347 |
val _ = scheduler_check "join check"; |
|
348 |
val _ = Multithreading.self_critical () andalso |
|
349 |
error "Cannot join future values within critical section"; |
|
32055
6a46898aa805
recovered a version of dequeue_towards (cf. bb7b5a5942c7);
wenzelm
parents:
32053
diff
changeset
|
350 |
|
32058 | 351 |
val worker = is_worker (); |
32096 | 352 |
val _ = if worker then join_deps (map task_of xs) else (); |
353 |
||
32055
6a46898aa805
recovered a version of dequeue_towards (cf. bb7b5a5942c7);
wenzelm
parents:
32053
diff
changeset
|
354 |
fun join_wait x = |
6a46898aa805
recovered a version of dequeue_towards (cf. bb7b5a5942c7);
wenzelm
parents:
32053
diff
changeset
|
355 |
if SYNCHRONIZED "join_wait" (fn () => |
32219
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
356 |
is_finished x orelse ((if worker then worker_wait else wait) work_finished; false)) |
32055
6a46898aa805
recovered a version of dequeue_towards (cf. bb7b5a5942c7);
wenzelm
parents:
32053
diff
changeset
|
357 |
then () else join_wait x; |
6a46898aa805
recovered a version of dequeue_towards (cf. bb7b5a5942c7);
wenzelm
parents:
32053
diff
changeset
|
358 |
|
32186 | 359 |
val _ = xs |> List.app (fn x => |
360 |
let val time = Multithreading.real_time join_wait x in |
|
361 |
Multithreading.tracing_time true time |
|
362 |
(fn () => "joined after " ^ Time.toString time) |
|
363 |
end); |
|
29366 | 364 |
in map get_result xs end) (); |
28186 | 365 |
|
29551
95e469919c3e
join_results: when dependencies are resulved (but not finished yet),
wenzelm
parents:
29431
diff
changeset
|
366 |
end; |
95e469919c3e
join_results: when dependencies are resulved (but not finished yet),
wenzelm
parents:
29431
diff
changeset
|
367 |
|
28647
8068cdc84e7e
join_results: allow CRITICAL join of finished futures;
wenzelm
parents:
28645
diff
changeset
|
368 |
fun join_result x = singleton join_results x; |
8068cdc84e7e
join_results: allow CRITICAL join of finished futures;
wenzelm
parents:
28645
diff
changeset
|
369 |
fun join x = Exn.release (join_result x); |
28156 | 370 |
|
29366 | 371 |
|
372 |
(* map *) |
|
373 |
||
29384
a3c7e9ae9b71
more robust propagation of errors through bulk jobs;
wenzelm
parents:
29366
diff
changeset
|
374 |
fun map_future f x = |
29366 | 375 |
let |
376 |
val _ = scheduler_check "map_future check"; |
|
377 |
||
29384
a3c7e9ae9b71
more robust propagation of errors through bulk jobs;
wenzelm
parents:
29366
diff
changeset
|
378 |
val task = task_of x; |
32102 | 379 |
val group = Task_Queue.new_group (SOME (group_of x)); |
29384
a3c7e9ae9b71
more robust propagation of errors through bulk jobs;
wenzelm
parents:
29366
diff
changeset
|
380 |
val (result, job) = future_job group (fn () => f (join x)); |
a3c7e9ae9b71
more robust propagation of errors through bulk jobs;
wenzelm
parents:
29366
diff
changeset
|
381 |
|
29366 | 382 |
val extended = SYNCHRONIZED "map_future" (fn () => |
383 |
(case Task_Queue.extend task job (! queue) of |
|
384 |
SOME queue' => (queue := queue'; true) |
|
385 |
| NONE => false)); |
|
386 |
in |
|
29384
a3c7e9ae9b71
more robust propagation of errors through bulk jobs;
wenzelm
parents:
29366
diff
changeset
|
387 |
if extended then Future {task = task, group = group, result = result} |
32099 | 388 |
else fork_future (SOME group) [task] (Task_Queue.pri_of_task task) (fn () => f (join x)) |
29366 | 389 |
end; |
28979
3ce619d8d432
fork/map: no inheritance of group (structure is nested, not parallel);
wenzelm
parents:
28972
diff
changeset
|
390 |
|
28191 | 391 |
|
29431 | 392 |
(* cancellation *) |
28202
23cb9a974630
added focus, which indicates a particular collection of high-priority tasks;
wenzelm
parents:
28201
diff
changeset
|
393 |
|
30618
046f4f986fb5
restricted interrupts for tasks running as future worker thread -- attempt to prevent interrupt race conditions;
wenzelm
parents:
30612
diff
changeset
|
394 |
fun interruptible_task f x = |
046f4f986fb5
restricted interrupts for tasks running as future worker thread -- attempt to prevent interrupt race conditions;
wenzelm
parents:
30612
diff
changeset
|
395 |
if Multithreading.available then |
046f4f986fb5
restricted interrupts for tasks running as future worker thread -- attempt to prevent interrupt race conditions;
wenzelm
parents:
30612
diff
changeset
|
396 |
Multithreading.with_attributes |
32058 | 397 |
(if is_worker () |
30618
046f4f986fb5
restricted interrupts for tasks running as future worker thread -- attempt to prevent interrupt race conditions;
wenzelm
parents:
30612
diff
changeset
|
398 |
then Multithreading.restricted_interrupts |
046f4f986fb5
restricted interrupts for tasks running as future worker thread -- attempt to prevent interrupt race conditions;
wenzelm
parents:
30612
diff
changeset
|
399 |
else Multithreading.regular_interrupts) |
046f4f986fb5
restricted interrupts for tasks running as future worker thread -- attempt to prevent interrupt race conditions;
wenzelm
parents:
30612
diff
changeset
|
400 |
(fn _ => f) x |
046f4f986fb5
restricted interrupts for tasks running as future worker thread -- attempt to prevent interrupt race conditions;
wenzelm
parents:
30612
diff
changeset
|
401 |
else interruptible f x; |
046f4f986fb5
restricted interrupts for tasks running as future worker thread -- attempt to prevent interrupt race conditions;
wenzelm
parents:
30612
diff
changeset
|
402 |
|
28202
23cb9a974630
added focus, which indicates a particular collection of high-priority tasks;
wenzelm
parents:
28201
diff
changeset
|
403 |
(*interrupt: permissive signal, may get ignored*) |
28197 | 404 |
fun interrupt_task id = SYNCHRONIZED "interrupt" |
29119 | 405 |
(fn () => Task_Queue.interrupt_external (! queue) id); |
28191 | 406 |
|
28206
bcd48c6897d4
eliminated requests, use global state variables uniformly;
wenzelm
parents:
28203
diff
changeset
|
407 |
(*cancel: present and future group members will be interrupted eventually*) |
29431 | 408 |
fun cancel_group group = |
28464 | 409 |
(scheduler_check "cancel check"; |
32219
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
410 |
SYNCHRONIZED "cancel" (fn () => (do_cancel group; broadcast scheduler_event))); |
28206
bcd48c6897d4
eliminated requests, use global state variables uniformly;
wenzelm
parents:
28203
diff
changeset
|
411 |
|
29431 | 412 |
fun cancel x = cancel_group (group_of x); |
28206
bcd48c6897d4
eliminated requests, use global state variables uniformly;
wenzelm
parents:
28203
diff
changeset
|
413 |
|
29366 | 414 |
|
415 |
(** global join and shutdown **) |
|
416 |
||
28203 | 417 |
fun shutdown () = |
28276 | 418 |
if Multithreading.available then |
28464 | 419 |
(scheduler_check "shutdown check"; |
28276 | 420 |
SYNCHRONIZED "shutdown" (fn () => |
32219
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
421 |
(while not (scheduler_active ()) do wait scheduler_event; |
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
422 |
while not (Task_Queue.is_empty (! queue)) do wait scheduler_event; |
28276 | 423 |
do_shutdown := true; |
32219
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
424 |
while scheduler_active () do |
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
425 |
(broadcast work_available; |
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
426 |
broadcast scheduler_event; |
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
427 |
wait scheduler_event)))) |
28276 | 428 |
else (); |
28203 | 429 |
|
29366 | 430 |
|
431 |
(*final declarations of this structure!*) |
|
432 |
val map = map_future; |
|
433 |
||
28156 | 434 |
end; |
28972 | 435 |
|
436 |
type 'a future = 'a Future.future; |
|
437 |