author | huffman |
Tue, 02 Mar 2010 09:54:50 -0800 | |
changeset 35512 | d1ef88d7de5a |
parent 35016 | c0f2e49bccfc |
child 37046 | 78d88b670a53 |
permissions | -rw-r--r-- |
28156 | 1 |
(* Title: Pure/Concurrent/future.ML |
2 |
Author: Makarius |
|
3 |
||
32246 | 4 |
Future values, see also |
5 |
http://www4.in.tum.de/~wenzelm/papers/parallel-isabelle.pdf |
|
28201 | 6 |
|
7 |
Notes: |
|
8 |
||
9 |
* Futures are similar to delayed evaluation, i.e. delay/force is |
|
10 |
generalized to fork/join (and variants). The idea is to model |
|
11 |
parallel value-oriented computations, but *not* communicating |
|
12 |
processes. |
|
13 |
||
14 |
* Futures are grouped; failure of one group member causes the whole |
|
32220 | 15 |
group to be interrupted eventually. Groups are block-structured. |
28201 | 16 |
|
17 |
* Forked futures are evaluated spontaneously by a farm of worker |
|
18 |
threads in the background; join resynchronizes the computation and |
|
19 |
delivers results (values or exceptions). |
|
20 |
||
21 |
* The pool of worker threads is limited, usually in correlation with |
|
22 |
the number of physical cores on the machine. Note that allocation |
|
23 |
of runtime resources is distorted either if workers yield CPU time |
|
24 |
(e.g. via system sleep or wait operations), or if non-worker |
|
25 |
threads contend for significant runtime resources independently. |
|
34277
7325a5e3587f
added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents:
33416
diff
changeset
|
26 |
|
7325a5e3587f
added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents:
33416
diff
changeset
|
27 |
* Promised futures are fulfilled by external means. There is no |
7325a5e3587f
added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents:
33416
diff
changeset
|
28 |
associated evaluation task, but other futures can depend on them |
7325a5e3587f
added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents:
33416
diff
changeset
|
29 |
as usual. |
28156 | 30 |
*) |
31 |
||
32 |
signature FUTURE = |
|
33 |
sig |
|
29119 | 34 |
type task = Task_Queue.task |
35 |
type group = Task_Queue.group |
|
32058 | 36 |
val is_worker: unit -> bool |
32814
81897d30b97f
added Task_Queue.depend (again) -- light-weight version for transitive graph;
wenzelm
parents:
32738
diff
changeset
|
37 |
val worker_task: unit -> Task_Queue.task option |
32102 | 38 |
val worker_group: unit -> Task_Queue.group option |
28972 | 39 |
type 'a future |
40 |
val task_of: 'a future -> task |
|
41 |
val group_of: 'a future -> group |
|
42 |
val peek: 'a future -> 'a Exn.result option |
|
43 |
val is_finished: 'a future -> bool |
|
28979
3ce619d8d432
fork/map: no inheritance of group (structure is nested, not parallel);
wenzelm
parents:
28972
diff
changeset
|
44 |
val fork_group: group -> (unit -> 'a) -> 'a future |
32724 | 45 |
val fork_deps_pri: 'b future list -> int -> (unit -> 'a) -> 'a future |
28979
3ce619d8d432
fork/map: no inheritance of group (structure is nested, not parallel);
wenzelm
parents:
28972
diff
changeset
|
46 |
val fork_deps: 'b future list -> (unit -> 'a) -> 'a future |
29119 | 47 |
val fork_pri: int -> (unit -> 'a) -> 'a future |
32724 | 48 |
val fork: (unit -> 'a) -> 'a future |
28972 | 49 |
val join_results: 'a future list -> 'a Exn.result list |
50 |
val join_result: 'a future -> 'a Exn.result |
|
51 |
val join: 'a future -> 'a |
|
34277
7325a5e3587f
added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents:
33416
diff
changeset
|
52 |
val value: 'a -> 'a future |
28972 | 53 |
val map: ('a -> 'b) -> 'a future -> 'b future |
34277
7325a5e3587f
added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents:
33416
diff
changeset
|
54 |
val promise_group: group -> 'a future |
7325a5e3587f
added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents:
33416
diff
changeset
|
55 |
val promise: unit -> 'a future |
7325a5e3587f
added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents:
33416
diff
changeset
|
56 |
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
|
57 |
val fulfill: 'a future -> 'a -> unit |
30618
046f4f986fb5
restricted interrupts for tasks running as future worker thread -- attempt to prevent interrupt race conditions;
wenzelm
parents:
30612
diff
changeset
|
58 |
val interruptible_task: ('a -> 'b) -> 'a -> 'b |
29431 | 59 |
val cancel_group: group -> unit |
28972 | 60 |
val cancel: 'a future -> unit |
28203 | 61 |
val shutdown: unit -> unit |
28156 | 62 |
end; |
63 |
||
64 |
structure Future: FUTURE = |
|
65 |
struct |
|
66 |
||
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
|
67 |
(** 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
|
68 |
|
28167 | 69 |
(* identifiers *) |
70 |
||
29119 | 71 |
type task = Task_Queue.task; |
72 |
type group = Task_Queue.group; |
|
28167 | 73 |
|
32058 | 74 |
local |
33408 | 75 |
val tag = Universal.tag () : (task * group) option Universal.tag; |
32058 | 76 |
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
|
77 |
fun thread_data () = the_default NONE (Thread.getLocal tag); |
32058 | 78 |
fun setmp_thread_data data f x = |
79 |
Library.setmp_thread_data tag (thread_data ()) (SOME data) f x; |
|
28167 | 80 |
end; |
81 |
||
32058 | 82 |
val is_worker = is_some o thread_data; |
33408 | 83 |
val worker_task = Option.map #1 o thread_data; |
84 |
val worker_group = Option.map #2 o thread_data; |
|
32058 | 85 |
|
34277
7325a5e3587f
added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents:
33416
diff
changeset
|
86 |
fun new_group () = Task_Queue.new_group (worker_group ()); |
7325a5e3587f
added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents:
33416
diff
changeset
|
87 |
|
28167 | 88 |
|
89 |
(* datatype future *) |
|
90 |
||
35016 | 91 |
type 'a result = 'a Exn.result Single_Assignment.var; |
92 |
||
28972 | 93 |
datatype 'a future = Future of |
34277
7325a5e3587f
added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents:
33416
diff
changeset
|
94 |
{promised: bool, |
7325a5e3587f
added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents:
33416
diff
changeset
|
95 |
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
|
96 |
group: group, |
35016 | 97 |
result: 'a result}; |
28167 | 98 |
|
99 |
fun task_of (Future {task, ...}) = task; |
|
100 |
fun group_of (Future {group, ...}) = group; |
|
32253 | 101 |
fun result_of (Future {result, ...}) = result; |
28167 | 102 |
|
35016 | 103 |
fun peek x = Single_Assignment.peek (result_of x); |
28558 | 104 |
fun is_finished x = is_some (peek x); |
28320 | 105 |
|
34277
7325a5e3587f
added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents:
33416
diff
changeset
|
106 |
fun assign_result group result res = |
7325a5e3587f
added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents:
33416
diff
changeset
|
107 |
let |
35016 | 108 |
val _ = Single_Assignment.assign result res; |
34277
7325a5e3587f
added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents:
33416
diff
changeset
|
109 |
val ok = |
7325a5e3587f
added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents:
33416
diff
changeset
|
110 |
(case res of |
7325a5e3587f
added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents:
33416
diff
changeset
|
111 |
Exn.Exn exn => (Task_Queue.cancel_group group exn; false) |
7325a5e3587f
added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents:
33416
diff
changeset
|
112 |
| Exn.Result _ => true); |
7325a5e3587f
added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents:
33416
diff
changeset
|
113 |
in ok end; |
28997 | 114 |
|
28167 | 115 |
|
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
|
116 |
|
8c0335bc9336
inherit group from running thread, or create a new one -- make it harder to re-use canceled groups;
wenzelm
parents:
28170
diff
changeset
|
117 |
(** 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
|
118 |
|
8c0335bc9336
inherit group from running thread, or create a new one -- make it harder to re-use canceled groups;
wenzelm
parents:
28170
diff
changeset
|
119 |
(* synchronization *) |
28156 | 120 |
|
32219
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
121 |
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
|
122 |
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
|
123 |
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
|
124 |
|
28156 | 125 |
local |
126 |
val lock = Mutex.mutex (); |
|
127 |
in |
|
128 |
||
28575 | 129 |
fun SYNCHRONIZED name = SimpleThread.synchronized name lock; |
28156 | 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 wait cond = (*requires SYNCHRONIZED*) |
32295
400cc493d466
renamed Multithreading.regular_interrupts to Multithreading.public_interrupts;
wenzelm
parents:
32293
diff
changeset
|
132 |
Multithreading.sync_wait NONE NONE cond lock; |
28206
bcd48c6897d4
eliminated requests, use global state variables uniformly;
wenzelm
parents:
28203
diff
changeset
|
133 |
|
32295
400cc493d466
renamed Multithreading.regular_interrupts to Multithreading.public_interrupts;
wenzelm
parents:
32293
diff
changeset
|
134 |
fun wait_timeout timeout cond = (*requires SYNCHRONIZED*) |
400cc493d466
renamed Multithreading.regular_interrupts to Multithreading.public_interrupts;
wenzelm
parents:
32293
diff
changeset
|
135 |
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
|
136 |
|
32219
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
137 |
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
|
138 |
ConditionVar.signal cond; |
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
139 |
|
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
140 |
fun broadcast cond = (*requires SYNCHRONIZED*) |
28166
43087721a66e
moved task, thread_data, group, queue to task_queue.ML;
wenzelm
parents:
28163
diff
changeset
|
141 |
ConditionVar.broadcast cond; |
28156 | 142 |
|
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
|
143 |
fun broadcast_work () = (*requires SYNCHRONIZED*) |
0241916a5f06
more precise treatment of scheduler_event: continous pulse (50ms) instead of flooding, which was burning many CPU cycles in spare threads;
wenzelm
parents:
32247
diff
changeset
|
144 |
(ConditionVar.broadcast work_available; |
32225
d5d6f47fb018
cancel: improved reactivity due to more careful broadcasting;
wenzelm
parents:
32224
diff
changeset
|
145 |
ConditionVar.broadcast work_finished); |
d5d6f47fb018
cancel: improved reactivity due to more careful broadcasting;
wenzelm
parents:
32224
diff
changeset
|
146 |
|
28156 | 147 |
end; |
148 |
||
149 |
||
33410
e351f4c1f18c
worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents:
33409
diff
changeset
|
150 |
(* global state *) |
e351f4c1f18c
worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents:
33409
diff
changeset
|
151 |
|
e351f4c1f18c
worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents:
33409
diff
changeset
|
152 |
val queue = Unsynchronized.ref Task_Queue.empty; |
e351f4c1f18c
worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents:
33409
diff
changeset
|
153 |
val next = Unsynchronized.ref 0; |
e351f4c1f18c
worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents:
33409
diff
changeset
|
154 |
val scheduler = Unsynchronized.ref (NONE: Thread.thread option); |
e351f4c1f18c
worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents:
33409
diff
changeset
|
155 |
val canceled = Unsynchronized.ref ([]: Task_Queue.group list); |
e351f4c1f18c
worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents:
33409
diff
changeset
|
156 |
val do_shutdown = Unsynchronized.ref false; |
e351f4c1f18c
worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents:
33409
diff
changeset
|
157 |
val max_workers = Unsynchronized.ref 0; |
e351f4c1f18c
worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents:
33409
diff
changeset
|
158 |
val max_active = Unsynchronized.ref 0; |
33411
a07558eb5029
worker_next: treat wait for work_available as Sleeping, not Waiting;
wenzelm
parents:
33410
diff
changeset
|
159 |
val worker_trend = Unsynchronized.ref 0; |
33410
e351f4c1f18c
worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents:
33409
diff
changeset
|
160 |
|
e351f4c1f18c
worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents:
33409
diff
changeset
|
161 |
datatype worker_state = Working | Waiting | Sleeping; |
e351f4c1f18c
worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents:
33409
diff
changeset
|
162 |
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
|
163 |
|
e351f4c1f18c
worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents:
33409
diff
changeset
|
164 |
fun count_workers state = (*requires SYNCHRONIZED*) |
e351f4c1f18c
worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents:
33409
diff
changeset
|
165 |
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
|
166 |
|
e351f4c1f18c
worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents:
33409
diff
changeset
|
167 |
|
32099 | 168 |
(* execute future jobs *) |
169 |
||
170 |
fun future_job group (e: unit -> 'a) = |
|
171 |
let |
|
35016 | 172 |
val result = Single_Assignment.var "future" : 'a result; |
32107
47d0da617fcc
future_job: tight scope for interrupts, to prevent shooting ourselves in the foot via cancel_group;
wenzelm
parents:
32102
diff
changeset
|
173 |
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
|
174 |
let |
47d0da617fcc
future_job: tight scope for interrupts, to prevent shooting ourselves in the foot via cancel_group;
wenzelm
parents:
32102
diff
changeset
|
175 |
val res = |
47d0da617fcc
future_job: tight scope for interrupts, to prevent shooting ourselves in the foot via cancel_group;
wenzelm
parents:
32102
diff
changeset
|
176 |
if ok then |
32230
9f6461b1c9cc
interruptible: Thread.testInterrupt before changing thread attributes;
wenzelm
parents:
32229
diff
changeset
|
177 |
Exn.capture (fn () => |
32295
400cc493d466
renamed Multithreading.regular_interrupts to Multithreading.public_interrupts;
wenzelm
parents:
32293
diff
changeset
|
178 |
Multithreading.with_attributes Multithreading.private_interrupts (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
|
179 |
else Exn.Exn Exn.Interrupt; |
34277
7325a5e3587f
added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents:
33416
diff
changeset
|
180 |
in assign_result group result res end; |
32099 | 181 |
in (result, job) end; |
28156 | 182 |
|
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
|
183 |
fun cancel_now group = (*requires SYNCHRONIZED*) |
34280
16bf3e9786a3
eliminated cache, which complicates the code without making a real difference (NB: deque_towards often disrupts caching, and daisy-chaining of workers already reduces queue overhead);
wenzelm
parents:
34279
diff
changeset
|
184 |
Task_Queue.cancel (! queue) group; |
34279
02936e77a07c
tasks of canceled groups are considered "ready" -- enables to purge the queue from tasks depending on unfinished promises (also improves general reactivity);
wenzelm
parents:
34277
diff
changeset
|
185 |
|
02936e77a07c
tasks of canceled groups are considered "ready" -- enables to purge the queue from tasks depending on unfinished promises (also improves general reactivity);
wenzelm
parents:
34277
diff
changeset
|
186 |
fun cancel_later group = (*requires SYNCHRONIZED*) |
32738 | 187 |
(Unsynchronized.change canceled (insert Task_Queue.eq_group group); |
188 |
broadcast scheduler_event); |
|
29341
6bb007a0f9f2
more reactive scheduler: reduced loop timeout, propagate broadcast interrupt via TaskQueue.cancel_all;
wenzelm
parents:
29119
diff
changeset
|
189 |
|
33408 | 190 |
fun execute (task, group, jobs) = |
28167 | 191 |
let |
32102 | 192 |
val valid = not (Task_Queue.is_canceled group); |
33408 | 193 |
val ok = setmp_thread_data (task, group) (fn () => |
29384
a3c7e9ae9b71
more robust propagation of errors through bulk jobs;
wenzelm
parents:
29366
diff
changeset
|
194 |
fold (fn job => fn ok => job valid andalso ok) jobs true) (); |
32246 | 195 |
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
|
196 |
let |
32738 | 197 |
val maximal = Unsynchronized.change_result queue (Task_Queue.finish task); |
32219
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
198 |
val _ = |
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
199 |
if ok 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
|
200 |
else if cancel_now group then () |
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
|
201 |
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
|
202 |
val _ = broadcast work_finished; |
33413
cb409680dda8
avoid broadcast work_available, use daisy-chained signal instead;
wenzelm
parents:
33411
diff
changeset
|
203 |
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
|
204 |
in () end); |
28167 | 205 |
in () end; |
206 |
||
207 |
||
208 |
(* worker threads *) |
|
209 |
||
33410
e351f4c1f18c
worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents:
33409
diff
changeset
|
210 |
fun worker_wait active cond = (*requires SYNCHRONIZED*) |
33406
1ddcb8472bd2
slightly leaner and more direct control of worker activity etc.;
wenzelm
parents:
33061
diff
changeset
|
211 |
let |
33410
e351f4c1f18c
worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents:
33409
diff
changeset
|
212 |
val state = |
e351f4c1f18c
worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents:
33409
diff
changeset
|
213 |
(case AList.lookup Thread.equal (! workers) (Thread.self ()) of |
e351f4c1f18c
worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents:
33409
diff
changeset
|
214 |
SOME state => state |
e351f4c1f18c
worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents:
33409
diff
changeset
|
215 |
| NONE => raise Fail "Unregistered worker thread"); |
e351f4c1f18c
worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents:
33409
diff
changeset
|
216 |
val _ = state := (if active then Waiting else Sleeping); |
33406
1ddcb8472bd2
slightly leaner and more direct control of worker activity etc.;
wenzelm
parents:
33061
diff
changeset
|
217 |
val _ = wait cond; |
33410
e351f4c1f18c
worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents:
33409
diff
changeset
|
218 |
val _ = state := Working; |
33406
1ddcb8472bd2
slightly leaner and more direct control of worker activity etc.;
wenzelm
parents:
33061
diff
changeset
|
219 |
in () end; |
28162 | 220 |
|
33415
352fe8e9162d
worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents:
33413
diff
changeset
|
221 |
fun worker_next () = (*requires SYNCHRONIZED*) |
33406
1ddcb8472bd2
slightly leaner and more direct control of worker activity etc.;
wenzelm
parents:
33061
diff
changeset
|
222 |
if length (! workers) > ! max_workers then |
1ddcb8472bd2
slightly leaner and more direct control of worker activity etc.;
wenzelm
parents:
33061
diff
changeset
|
223 |
(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
|
224 |
signal work_available; |
28167 | 225 |
NONE) |
33410
e351f4c1f18c
worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents:
33409
diff
changeset
|
226 |
else if count_workers Working > ! max_active then |
33415
352fe8e9162d
worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents:
33413
diff
changeset
|
227 |
(worker_wait false work_available; worker_next ()) |
28166
43087721a66e
moved task, thread_data, group, queue to task_queue.ML;
wenzelm
parents:
28163
diff
changeset
|
228 |
else |
32738 | 229 |
(case Unsynchronized.change_result queue (Task_Queue.dequeue (Thread.self ())) of |
33415
352fe8e9162d
worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents:
33413
diff
changeset
|
230 |
NONE => (worker_wait false work_available; worker_next ()) |
33413
cb409680dda8
avoid broadcast work_available, use daisy-chained signal instead;
wenzelm
parents:
33411
diff
changeset
|
231 |
| some => (signal work_available; some)); |
28156 | 232 |
|
28167 | 233 |
fun worker_loop name = |
33415
352fe8e9162d
worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents:
33413
diff
changeset
|
234 |
(case SYNCHRONIZED name (fn () => worker_next ()) of |
29119 | 235 |
NONE => () |
33408 | 236 |
| SOME work => (execute work; worker_loop name)); |
28156 | 237 |
|
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
|
238 |
fun worker_start name = (*requires SYNCHRONIZED*) |
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
|
239 |
Unsynchronized.change workers (cons (SimpleThread.fork false (fn () => worker_loop name), |
33410
e351f4c1f18c
worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents:
33409
diff
changeset
|
240 |
Unsynchronized.ref Working)); |
28156 | 241 |
|
242 |
||
243 |
(* scheduler *) |
|
244 |
||
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
|
245 |
val status_ticks = Unsynchronized.ref 0; |
1427333220bc
worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents:
33406
diff
changeset
|
246 |
|
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
|
247 |
val last_round = Unsynchronized.ref Time.zeroTime; |
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
|
248 |
val next_round = Time.fromMilliseconds 50; |
32226 | 249 |
|
28206
bcd48c6897d4
eliminated requests, use global state variables uniformly;
wenzelm
parents:
28203
diff
changeset
|
250 |
fun scheduler_next () = (*requires SYNCHRONIZED*) |
28156 | 251 |
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
|
252 |
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
|
253 |
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
|
254 |
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
|
255 |
|
33415
352fe8e9162d
worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents:
33413
diff
changeset
|
256 |
|
352fe8e9162d
worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents:
33413
diff
changeset
|
257 |
(* queue and worker status *) |
352fe8e9162d
worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents:
33413
diff
changeset
|
258 |
|
32226 | 259 |
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
|
260 |
if tick then Unsynchronized.change status_ticks (fn i => (i + 1) mod 10) else (); |
1427333220bc
worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents:
33406
diff
changeset
|
261 |
val _ = |
1427333220bc
worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents:
33406
diff
changeset
|
262 |
if tick andalso ! status_ticks = 0 then |
1427333220bc
worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents:
33406
diff
changeset
|
263 |
Multithreading.tracing 1 (fn () => |
1427333220bc
worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents:
33406
diff
changeset
|
264 |
let |
34277
7325a5e3587f
added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents:
33416
diff
changeset
|
265 |
val {ready, pending, running, passive} = Task_Queue.status (! queue); |
33407
1427333220bc
worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents:
33406
diff
changeset
|
266 |
val total = length (! workers); |
33410
e351f4c1f18c
worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents:
33409
diff
changeset
|
267 |
val active = count_workers Working; |
e351f4c1f18c
worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents:
33409
diff
changeset
|
268 |
val waiting = count_workers Waiting; |
33407
1427333220bc
worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents:
33406
diff
changeset
|
269 |
in |
1427333220bc
worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents:
33406
diff
changeset
|
270 |
"SCHEDULE " ^ Time.toString now ^ ": " ^ |
1427333220bc
worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents:
33406
diff
changeset
|
271 |
string_of_int ready ^ " ready, " ^ |
1427333220bc
worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents:
33406
diff
changeset
|
272 |
string_of_int pending ^ " pending, " ^ |
34277
7325a5e3587f
added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents:
33416
diff
changeset
|
273 |
string_of_int running ^ " running, " ^ |
7325a5e3587f
added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents:
33416
diff
changeset
|
274 |
string_of_int passive ^ " passive; " ^ |
33407
1427333220bc
worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents:
33406
diff
changeset
|
275 |
string_of_int total ^ " workers, " ^ |
33410
e351f4c1f18c
worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents:
33409
diff
changeset
|
276 |
string_of_int active ^ " active, " ^ |
e351f4c1f18c
worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents:
33409
diff
changeset
|
277 |
string_of_int waiting ^ " waiting " |
33407
1427333220bc
worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents:
33406
diff
changeset
|
278 |
end) |
1427333220bc
worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents:
33406
diff
changeset
|
279 |
else (); |
32053 | 280 |
|
28191 | 281 |
val _ = |
32219
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
282 |
if forall (Thread.isActive o #1) (! workers) then () |
32095 | 283 |
else |
33409 | 284 |
let |
285 |
val (alive, dead) = List.partition (Thread.isActive o #1) (! workers); |
|
286 |
val _ = workers := alive; |
|
287 |
in |
|
288 |
Multithreading.tracing 0 (fn () => |
|
289 |
"SCHEDULE: disposed " ^ string_of_int (length dead) ^ " dead worker threads") |
|
290 |
end; |
|
28191 | 291 |
|
33415
352fe8e9162d
worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents:
33413
diff
changeset
|
292 |
|
352fe8e9162d
worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents:
33413
diff
changeset
|
293 |
(* worker pool adjustments *) |
352fe8e9162d
worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents:
33413
diff
changeset
|
294 |
|
352fe8e9162d
worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents:
33413
diff
changeset
|
295 |
val max_active0 = ! max_active; |
352fe8e9162d
worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents:
33413
diff
changeset
|
296 |
val max_workers0 = ! max_workers; |
352fe8e9162d
worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents:
33413
diff
changeset
|
297 |
|
28206
bcd48c6897d4
eliminated requests, use global state variables uniformly;
wenzelm
parents:
28203
diff
changeset
|
298 |
val m = if ! do_shutdown then 0 else Multithreading.max_threads_value (); |
33406
1ddcb8472bd2
slightly leaner and more direct control of worker activity etc.;
wenzelm
parents:
33061
diff
changeset
|
299 |
val _ = max_active := m; |
1ddcb8472bd2
slightly leaner and more direct control of worker activity etc.;
wenzelm
parents:
33061
diff
changeset
|
300 |
|
33411
a07558eb5029
worker_next: treat wait for work_available as Sleeping, not Waiting;
wenzelm
parents:
33410
diff
changeset
|
301 |
val mm = |
a07558eb5029
worker_next: treat wait for work_available as Sleeping, not Waiting;
wenzelm
parents:
33410
diff
changeset
|
302 |
if ! do_shutdown then 0 |
a07558eb5029
worker_next: treat wait for work_available as Sleeping, not Waiting;
wenzelm
parents:
33410
diff
changeset
|
303 |
else if m = 9999 then 1 |
33413
cb409680dda8
avoid broadcast work_available, use daisy-chained signal instead;
wenzelm
parents:
33411
diff
changeset
|
304 |
else Int.min (Int.max (count_workers Working + 2 * count_workers Waiting, m), 4 * m); |
33411
a07558eb5029
worker_next: treat wait for work_available as Sleeping, not Waiting;
wenzelm
parents:
33410
diff
changeset
|
305 |
val _ = |
a07558eb5029
worker_next: treat wait for work_available as Sleeping, not Waiting;
wenzelm
parents:
33410
diff
changeset
|
306 |
if tick andalso mm > ! max_workers then |
a07558eb5029
worker_next: treat wait for work_available as Sleeping, not Waiting;
wenzelm
parents:
33410
diff
changeset
|
307 |
Unsynchronized.change worker_trend (fn w => if w < 0 then 0 else w + 1) |
a07558eb5029
worker_next: treat wait for work_available as Sleeping, not Waiting;
wenzelm
parents:
33410
diff
changeset
|
308 |
else if tick andalso mm < ! max_workers then |
a07558eb5029
worker_next: treat wait for work_available as Sleeping, not Waiting;
wenzelm
parents:
33410
diff
changeset
|
309 |
Unsynchronized.change worker_trend (fn w => if w > 0 then 0 else w - 1) |
a07558eb5029
worker_next: treat wait for work_available as Sleeping, not Waiting;
wenzelm
parents:
33410
diff
changeset
|
310 |
else (); |
a07558eb5029
worker_next: treat wait for work_available as Sleeping, not Waiting;
wenzelm
parents:
33410
diff
changeset
|
311 |
val _ = |
33415
352fe8e9162d
worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents:
33413
diff
changeset
|
312 |
if mm = 0 orelse ! worker_trend > 50 orelse ! worker_trend < ~50 then |
352fe8e9162d
worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents:
33413
diff
changeset
|
313 |
max_workers := mm |
352fe8e9162d
worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents:
33413
diff
changeset
|
314 |
else if ! worker_trend > 5 andalso ! max_workers < 2 * m then |
352fe8e9162d
worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents:
33413
diff
changeset
|
315 |
max_workers := Int.min (mm, 2 * m) |
33411
a07558eb5029
worker_next: treat wait for work_available as Sleeping, not Waiting;
wenzelm
parents:
33410
diff
changeset
|
316 |
else (); |
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 | 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 | 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 | 334 |
else |
335 |
(Multithreading.tracing 1 (fn () => |
|
336 |
string_of_int (length (! canceled)) ^ " canceled groups"); |
|
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
|
337 |
Unsynchronized.change canceled (filter_out cancel_now); |
32293 | 338 |
broadcast_work ()); |
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 | 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 |
|
34277
7325a5e3587f
added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents:
33416
diff
changeset
|
348 |
val _ = if Task_Queue.all_passive (! queue) then do_shutdown := true else (); |
32219
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
349 |
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
|
350 |
val _ = if continue then () else scheduler := NONE; |
33415
352fe8e9162d
worker_next: plain signalling via work_available only, not scheduler_event;
wenzelm
parents:
33413
diff
changeset
|
351 |
|
32219
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
352 |
val _ = broadcast scheduler_event; |
32295
400cc493d466
renamed Multithreading.regular_interrupts to Multithreading.public_interrupts;
wenzelm
parents:
32293
diff
changeset
|
353 |
in continue end |
400cc493d466
renamed Multithreading.regular_interrupts to Multithreading.public_interrupts;
wenzelm
parents:
32293
diff
changeset
|
354 |
handle Exn.Interrupt => |
400cc493d466
renamed Multithreading.regular_interrupts to Multithreading.public_interrupts;
wenzelm
parents:
32293
diff
changeset
|
355 |
(Multithreading.tracing 1 (fn () => "Interrupt"); |
34280
16bf3e9786a3
eliminated cache, which complicates the code without making a real difference (NB: deque_towards often disrupts caching, and daisy-chaining of workers already reduces queue overhead);
wenzelm
parents:
34279
diff
changeset
|
356 |
List.app cancel_later (Task_Queue.cancel_all (! queue)); |
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
|
357 |
broadcast_work (); true); |
32295
400cc493d466
renamed Multithreading.regular_interrupts to Multithreading.public_interrupts;
wenzelm
parents:
32293
diff
changeset
|
358 |
|
28206
bcd48c6897d4
eliminated requests, use global state variables uniformly;
wenzelm
parents:
28203
diff
changeset
|
359 |
fun scheduler_loop () = |
33416
13d00799fe49
scheduler: clarified interrupt attributes and handling;
wenzelm
parents:
33415
diff
changeset
|
360 |
while |
13d00799fe49
scheduler: clarified interrupt attributes and handling;
wenzelm
parents:
33415
diff
changeset
|
361 |
Multithreading.with_attributes |
13d00799fe49
scheduler: clarified interrupt attributes and handling;
wenzelm
parents:
33415
diff
changeset
|
362 |
(Multithreading.sync_interrupts Multithreading.public_interrupts) |
13d00799fe49
scheduler: clarified interrupt attributes and handling;
wenzelm
parents:
33415
diff
changeset
|
363 |
(fn _ => SYNCHRONIZED "scheduler" (fn () => scheduler_next ())) |
13d00799fe49
scheduler: clarified interrupt attributes and handling;
wenzelm
parents:
33415
diff
changeset
|
364 |
do (); |
28156 | 365 |
|
28203 | 366 |
fun scheduler_active () = (*requires SYNCHRONIZED*) |
367 |
(case ! scheduler of NONE => false | SOME thread => Thread.isActive thread); |
|
368 |
||
32228
7622c03141b0
scheduler: shutdown spontaneously (after some delay) if queue is empty;
wenzelm
parents:
32227
diff
changeset
|
369 |
fun scheduler_check () = (*requires SYNCHRONIZED*) |
7622c03141b0
scheduler: shutdown spontaneously (after some delay) if queue is empty;
wenzelm
parents:
32227
diff
changeset
|
370 |
(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
|
371 |
if scheduler_active () then () |
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 |
else scheduler := SOME (SimpleThread.fork false scheduler_loop)); |
28156 | 373 |
|
374 |
||
29366 | 375 |
|
376 |
(** futures **) |
|
28156 | 377 |
|
29366 | 378 |
(* fork *) |
379 |
||
380 |
fun fork_future opt_group deps pri e = |
|
381 |
let |
|
32102 | 382 |
val group = |
383 |
(case opt_group of |
|
34277
7325a5e3587f
added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents:
33416
diff
changeset
|
384 |
NONE => new_group () |
7325a5e3587f
added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents:
33416
diff
changeset
|
385 |
| SOME group => group); |
29366 | 386 |
val (result, job) = future_job group e; |
32246 | 387 |
val task = SYNCHRONIZED "enqueue" (fn () => |
32219
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
388 |
let |
32738 | 389 |
val (task, minimal) = |
390 |
Unsynchronized.change_result queue (Task_Queue.enqueue group deps pri job); |
|
32219
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
391 |
val _ = if minimal then signal work_available else (); |
32228
7622c03141b0
scheduler: shutdown spontaneously (after some delay) if queue is empty;
wenzelm
parents:
32227
diff
changeset
|
392 |
val _ = scheduler_check (); |
32219
9a2566d1fdbd
more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents:
32186
diff
changeset
|
393 |
in task end); |
34277
7325a5e3587f
added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents:
33416
diff
changeset
|
394 |
in Future {promised = false, task = task, group = group, result = result} end; |
28162 | 395 |
|
29366 | 396 |
fun fork_group group e = fork_future (SOME group) [] 0 e; |
32724 | 397 |
fun fork_deps_pri deps pri e = fork_future NONE (map task_of deps) pri e; |
398 |
fun fork_deps deps e = fork_deps_pri deps 0 e; |
|
399 |
fun fork_pri pri e = fork_deps_pri [] pri e; |
|
400 |
fun fork e = fork_deps [] e; |
|
28186 | 401 |
|
402 |
||
29366 | 403 |
(* join *) |
404 |
||
29551
95e469919c3e
join_results: when dependencies are resulved (but not finished yet),
wenzelm
parents:
29431
diff
changeset
|
405 |
local |
95e469919c3e
join_results: when dependencies are resulved (but not finished yet),
wenzelm
parents:
29431
diff
changeset
|
406 |
|
32099 | 407 |
fun get_result x = |
408 |
(case peek x of |
|
32102 | 409 |
NONE => Exn.Exn (SYS_ERROR "unfinished future") |
410 |
| SOME (Exn.Exn Exn.Interrupt) => |
|
411 |
Exn.Exn (Exn.EXCEPTIONS (Exn.flatten_list (Task_Queue.group_status (group_of x)))) |
|
412 |
| SOME res => res); |
|
28186 | 413 |
|
32095 | 414 |
fun join_next deps = (*requires SYNCHRONIZED*) |
32224 | 415 |
if null deps then NONE |
416 |
else |
|
32738 | 417 |
(case Unsynchronized.change_result queue (Task_Queue.dequeue_towards (Thread.self ()) deps) of |
32224 | 418 |
(NONE, []) => NONE |
33410
e351f4c1f18c
worker activity: distinguish between waiting (formerly active) and sleeping;
wenzelm
parents:
33409
diff
changeset
|
419 |
| (NONE, deps') => (worker_wait true work_finished; join_next deps') |
32224 | 420 |
| (SOME work, deps') => SOME (work, deps')); |
32095 | 421 |
|
32814
81897d30b97f
added Task_Queue.depend (again) -- light-weight version for transitive graph;
wenzelm
parents:
32738
diff
changeset
|
422 |
fun execute_work NONE = () |
33408 | 423 |
| execute_work (SOME (work, deps')) = (execute work; join_work deps') |
32814
81897d30b97f
added Task_Queue.depend (again) -- light-weight version for transitive graph;
wenzelm
parents:
32738
diff
changeset
|
424 |
and join_work deps = |
81897d30b97f
added Task_Queue.depend (again) -- light-weight version for transitive graph;
wenzelm
parents:
32738
diff
changeset
|
425 |
execute_work (SYNCHRONIZED "join" (fn () => join_next deps)); |
81897d30b97f
added Task_Queue.depend (again) -- light-weight version for transitive graph;
wenzelm
parents:
32738
diff
changeset
|
426 |
|
81897d30b97f
added Task_Queue.depend (again) -- light-weight version for transitive graph;
wenzelm
parents:
32738
diff
changeset
|
427 |
fun join_depend task deps = |
81897d30b97f
added Task_Queue.depend (again) -- light-weight version for transitive graph;
wenzelm
parents:
32738
diff
changeset
|
428 |
execute_work (SYNCHRONIZED "join" (fn () => |
81897d30b97f
added Task_Queue.depend (again) -- light-weight version for transitive graph;
wenzelm
parents:
32738
diff
changeset
|
429 |
(Unsynchronized.change queue (Task_Queue.depend task deps); join_next deps))); |
29551
95e469919c3e
join_results: when dependencies are resulved (but not finished yet),
wenzelm
parents:
29431
diff
changeset
|
430 |
|
95e469919c3e
join_results: when dependencies are resulved (but not finished yet),
wenzelm
parents:
29431
diff
changeset
|
431 |
in |
95e469919c3e
join_results: when dependencies are resulved (but not finished yet),
wenzelm
parents:
29431
diff
changeset
|
432 |
|
29366 | 433 |
fun join_results xs = |
434 |
if forall is_finished xs then map get_result xs |
|
32246 | 435 |
else if Multithreading.self_critical () then |
436 |
error "Cannot join future values within critical section" |
|
32814
81897d30b97f
added Task_Queue.depend (again) -- light-weight version for transitive graph;
wenzelm
parents:
32738
diff
changeset
|
437 |
else |
81897d30b97f
added Task_Queue.depend (again) -- light-weight version for transitive graph;
wenzelm
parents:
32738
diff
changeset
|
438 |
(case worker_task () of |
81897d30b97f
added Task_Queue.depend (again) -- light-weight version for transitive graph;
wenzelm
parents:
32738
diff
changeset
|
439 |
SOME task => join_depend task (map task_of xs) |
35016 | 440 |
| NONE => List.app (ignore o Single_Assignment.await o result_of) xs; |
32814
81897d30b97f
added Task_Queue.depend (again) -- light-weight version for transitive graph;
wenzelm
parents:
32738
diff
changeset
|
441 |
map get_result xs); |
28186 | 442 |
|
29551
95e469919c3e
join_results: when dependencies are resulved (but not finished yet),
wenzelm
parents:
29431
diff
changeset
|
443 |
end; |
95e469919c3e
join_results: when dependencies are resulved (but not finished yet),
wenzelm
parents:
29431
diff
changeset
|
444 |
|
28647
8068cdc84e7e
join_results: allow CRITICAL join of finished futures;
wenzelm
parents:
28645
diff
changeset
|
445 |
fun join_result x = singleton join_results x; |
8068cdc84e7e
join_results: allow CRITICAL join of finished futures;
wenzelm
parents:
28645
diff
changeset
|
446 |
fun join x = Exn.release (join_result x); |
28156 | 447 |
|
29366 | 448 |
|
34277
7325a5e3587f
added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents:
33416
diff
changeset
|
449 |
(* fast-path versions -- bypassing full task management *) |
7325a5e3587f
added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents:
33416
diff
changeset
|
450 |
|
7325a5e3587f
added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents:
33416
diff
changeset
|
451 |
fun value (x: 'a) = |
7325a5e3587f
added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents:
33416
diff
changeset
|
452 |
let |
7325a5e3587f
added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents:
33416
diff
changeset
|
453 |
val group = Task_Queue.new_group NONE; |
35016 | 454 |
val result = Single_Assignment.var "value" : 'a result; |
34277
7325a5e3587f
added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents:
33416
diff
changeset
|
455 |
val _ = assign_result group result (Exn.Result x); |
7325a5e3587f
added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents:
33416
diff
changeset
|
456 |
in Future {promised = false, task = Task_Queue.dummy_task, group = group, result = result} end; |
29366 | 457 |
|
29384
a3c7e9ae9b71
more robust propagation of errors through bulk jobs;
wenzelm
parents:
29366
diff
changeset
|
458 |
fun map_future f x = |
29366 | 459 |
let |
29384
a3c7e9ae9b71
more robust propagation of errors through bulk jobs;
wenzelm
parents:
29366
diff
changeset
|
460 |
val task = task_of x; |
32102 | 461 |
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
|
462 |
val (result, job) = future_job group (fn () => f (join x)); |
a3c7e9ae9b71
more robust propagation of errors through bulk jobs;
wenzelm
parents:
29366
diff
changeset
|
463 |
|
32246 | 464 |
val extended = SYNCHRONIZED "extend" (fn () => |
29366 | 465 |
(case Task_Queue.extend task job (! queue) of |
466 |
SOME queue' => (queue := queue'; true) |
|
467 |
| NONE => false)); |
|
468 |
in |
|
34277
7325a5e3587f
added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents:
33416
diff
changeset
|
469 |
if extended then Future {promised = false, task = task, group = group, result = result} |
32099 | 470 |
else fork_future (SOME group) [task] (Task_Queue.pri_of_task task) (fn () => f (join x)) |
29366 | 471 |
end; |
28979
3ce619d8d432
fork/map: no inheritance of group (structure is nested, not parallel);
wenzelm
parents:
28972
diff
changeset
|
472 |
|
28191 | 473 |
|
34277
7325a5e3587f
added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents:
33416
diff
changeset
|
474 |
(* promised futures -- fulfilled by external means *) |
7325a5e3587f
added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents:
33416
diff
changeset
|
475 |
|
7325a5e3587f
added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents:
33416
diff
changeset
|
476 |
fun promise_group group : 'a future = |
7325a5e3587f
added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents:
33416
diff
changeset
|
477 |
let |
35016 | 478 |
val result = Single_Assignment.var "promise" : 'a result; |
34277
7325a5e3587f
added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents:
33416
diff
changeset
|
479 |
val task = SYNCHRONIZED "enqueue" (fn () => |
7325a5e3587f
added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents:
33416
diff
changeset
|
480 |
Unsynchronized.change_result queue (Task_Queue.enqueue_passive group)); |
7325a5e3587f
added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents:
33416
diff
changeset
|
481 |
in Future {promised = true, task = task, group = group, result = result} end; |
7325a5e3587f
added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents:
33416
diff
changeset
|
482 |
|
7325a5e3587f
added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents:
33416
diff
changeset
|
483 |
fun promise () = promise_group (new_group ()); |
7325a5e3587f
added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents:
33416
diff
changeset
|
484 |
|
7325a5e3587f
added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents:
33416
diff
changeset
|
485 |
fun fulfill_result (Future {promised, task, group, result}) res = |
7325a5e3587f
added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents:
33416
diff
changeset
|
486 |
let |
7325a5e3587f
added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents:
33416
diff
changeset
|
487 |
val _ = promised orelse raise Fail "Not a promised future"; |
7325a5e3587f
added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents:
33416
diff
changeset
|
488 |
fun job ok = assign_result group result (if ok then res else Exn.Exn Exn.Interrupt); |
7325a5e3587f
added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents:
33416
diff
changeset
|
489 |
val _ = execute (task, group, [job]); |
7325a5e3587f
added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents:
33416
diff
changeset
|
490 |
in () end; |
7325a5e3587f
added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents:
33416
diff
changeset
|
491 |
|
7325a5e3587f
added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents:
33416
diff
changeset
|
492 |
fun fulfill x res = fulfill_result x (Exn.Result res); |
7325a5e3587f
added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents:
33416
diff
changeset
|
493 |
|
7325a5e3587f
added Future.promise/fulfill -- promised futures that are fulfilled by external means;
wenzelm
parents:
33416
diff
changeset
|
494 |
|
29431 | 495 |
(* cancellation *) |
28202
23cb9a974630
added focus, which indicates a particular collection of high-priority tasks;
wenzelm
parents:
28201
diff
changeset
|
496 |
|
30618
046f4f986fb5
restricted interrupts for tasks running as future worker thread -- attempt to prevent interrupt race conditions;
wenzelm
parents:
30612
diff
changeset
|
497 |
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
|
498 |
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
|
499 |
Multithreading.with_attributes |
32058 | 500 |
(if is_worker () |
32295
400cc493d466
renamed Multithreading.regular_interrupts to Multithreading.public_interrupts;
wenzelm
parents:
32293
diff
changeset
|
501 |
then Multithreading.private_interrupts |
400cc493d466
renamed Multithreading.regular_interrupts to Multithreading.public_interrupts;
wenzelm
parents:
32293
diff
changeset
|
502 |
else Multithreading.public_interrupts) |
400cc493d466
renamed Multithreading.regular_interrupts to Multithreading.public_interrupts;
wenzelm
parents:
32293
diff
changeset
|
503 |
(fn _ => f x) |
30618
046f4f986fb5
restricted interrupts for tasks running as future worker thread -- attempt to prevent interrupt race conditions;
wenzelm
parents:
30612
diff
changeset
|
504 |
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
|
505 |
|
32228
7622c03141b0
scheduler: shutdown spontaneously (after some delay) if queue is empty;
wenzelm
parents:
32227
diff
changeset
|
506 |
(*cancel: present and future group members will be interrupted eventually*) |
34281
eedea6f0b37e
more robust cancelation, notably of passive futures without scheduler running;
wenzelm
parents:
34280
diff
changeset
|
507 |
fun cancel_group group = |
eedea6f0b37e
more robust cancelation, notably of passive futures without scheduler running;
wenzelm
parents:
34280
diff
changeset
|
508 |
SYNCHRONIZED "cancel" (fn () => if cancel_now group then () else cancel_later group); |
29431 | 509 |
fun cancel x = cancel_group (group_of x); |
28206
bcd48c6897d4
eliminated requests, use global state variables uniformly;
wenzelm
parents:
28203
diff
changeset
|
510 |
|
29366 | 511 |
|
32228
7622c03141b0
scheduler: shutdown spontaneously (after some delay) if queue is empty;
wenzelm
parents:
32227
diff
changeset
|
512 |
(* shutdown *) |
29366 | 513 |
|
28203 | 514 |
fun shutdown () = |
28276 | 515 |
if Multithreading.available then |
516 |
SYNCHRONIZED "shutdown" (fn () => |
|
32228
7622c03141b0
scheduler: shutdown spontaneously (after some delay) if queue is empty;
wenzelm
parents:
32227
diff
changeset
|
517 |
while scheduler_active () do |
34279
02936e77a07c
tasks of canceled groups are considered "ready" -- enables to purge the queue from tasks depending on unfinished promises (also improves general reactivity);
wenzelm
parents:
34277
diff
changeset
|
518 |
(wait scheduler_event; broadcast_work ())) |
28276 | 519 |
else (); |
28203 | 520 |
|
29366 | 521 |
|
522 |
(*final declarations of this structure!*) |
|
523 |
val map = map_future; |
|
524 |
||
28156 | 525 |
end; |
28972 | 526 |
|
527 |
type 'a future = 'a Future.future; |
|
528 |