| 5 |
|
*/ |
| 6 |
|
|
| 7 |
|
package jsr166y; |
| 8 |
– |
|
| 8 |
|
import java.io.Serializable; |
| 9 |
|
import java.util.Collection; |
| 11 |
– |
import java.util.Collections; |
| 10 |
|
import java.util.List; |
| 11 |
|
import java.util.RandomAccess; |
| 14 |
– |
import java.util.Map; |
| 12 |
|
import java.lang.ref.WeakReference; |
| 13 |
|
import java.lang.ref.ReferenceQueue; |
| 14 |
|
import java.util.concurrent.Callable; |
| 15 |
|
import java.util.concurrent.CancellationException; |
| 16 |
|
import java.util.concurrent.ExecutionException; |
| 20 |
– |
import java.util.concurrent.Executor; |
| 21 |
– |
import java.util.concurrent.ExecutorService; |
| 17 |
|
import java.util.concurrent.Future; |
| 18 |
|
import java.util.concurrent.RejectedExecutionException; |
| 19 |
|
import java.util.concurrent.RunnableFuture; |
| 42 |
|
* <p>A {@code ForkJoinTask} is a lightweight form of {@link Future}. |
| 43 |
|
* The efficiency of {@code ForkJoinTask}s stems from a set of |
| 44 |
|
* restrictions (that are only partially statically enforceable) |
| 45 |
< |
* reflecting their intended use as computational tasks calculating |
| 46 |
< |
* pure functions or operating on purely isolated objects. The |
| 47 |
< |
* primary coordination mechanisms are {@link #fork}, that arranges |
| 45 |
> |
* reflecting their main use as computational tasks calculating pure |
| 46 |
> |
* functions or operating on purely isolated objects. The primary |
| 47 |
> |
* coordination mechanisms are {@link #fork}, that arranges |
| 48 |
|
* asynchronous execution, and {@link #join}, that doesn't proceed |
| 49 |
|
* until the task's result has been computed. Computations should |
| 50 |
< |
* avoid {@code synchronized} methods or blocks, and should minimize |
| 51 |
< |
* other blocking synchronization apart from joining other tasks or |
| 52 |
< |
* using synchronizers such as Phasers that are advertised to |
| 53 |
< |
* cooperate with fork/join scheduling. Tasks should also not perform |
| 54 |
< |
* blocking IO, and should ideally access variables that are |
| 55 |
< |
* completely independent of those accessed by other running |
| 56 |
< |
* tasks. Minor breaches of these restrictions, for example using |
| 57 |
< |
* shared output streams, may be tolerable in practice, but frequent |
| 58 |
< |
* use may result in poor performance, and the potential to |
| 59 |
< |
* indefinitely stall if the number of threads not waiting for IO or |
| 60 |
< |
* other external synchronization becomes exhausted. This usage |
| 61 |
< |
* restriction is in part enforced by not permitting checked |
| 62 |
< |
* exceptions such as {@code IOExceptions} to be thrown. However, |
| 63 |
< |
* computations may still encounter unchecked exceptions, that are |
| 64 |
< |
* rethrown to callers attempting to join them. These exceptions may |
| 65 |
< |
* additionally include {@link RejectedExecutionException} stemming |
| 66 |
< |
* from internal resource exhaustion, such as failure to allocate |
| 67 |
< |
* internal task queues. Rethrown exceptions behave in the same way as |
| 68 |
< |
* regular exceptions, but, when possible, contain stack traces (as |
| 69 |
< |
* displayed for example using {@code ex.printStackTrace()}) of both |
| 70 |
< |
* the thread that initiated the computation as well as the thread |
| 71 |
< |
* actually encountering the exception; minimally only the latter. |
| 50 |
> |
* ideally avoid {@code synchronized} methods or blocks, and should |
| 51 |
> |
* minimize other blocking synchronization apart from joining other |
| 52 |
> |
* tasks or using synchronizers such as Phasers that are advertised to |
| 53 |
> |
* cooperate with fork/join scheduling. Subdividable tasks should also |
| 54 |
> |
* not perform blocking IO, and should ideally access variables that |
| 55 |
> |
* are completely independent of those accessed by other running |
| 56 |
> |
* tasks. These guidelines are loosely enforced by not permitting |
| 57 |
> |
* checked exceptions such as {@code IOExceptions} to be |
| 58 |
> |
* thrown. However, computations may still encounter unchecked |
| 59 |
> |
* exceptions, that are rethrown to callers attempting to join |
| 60 |
> |
* them. These exceptions may additionally include {@link |
| 61 |
> |
* RejectedExecutionException} stemming from internal resource |
| 62 |
> |
* exhaustion, such as failure to allocate internal task |
| 63 |
> |
* queues. Rethrown exceptions behave in the same way as regular |
| 64 |
> |
* exceptions, but, when possible, contain stack traces (as displayed |
| 65 |
> |
* for example using {@code ex.printStackTrace()}) of both the thread |
| 66 |
> |
* that initiated the computation as well as the thread actually |
| 67 |
> |
* encountering the exception; minimally only the latter. |
| 68 |
> |
* |
| 69 |
> |
* <p>It is possible to define and use ForkJoinTasks that may block, |
| 70 |
> |
* but doing do requires three further considerations: (1) Completion |
| 71 |
> |
* of few if any <em>other</em> tasks should be dependent on a task |
| 72 |
> |
* that blocks on external synchronization or IO. Event-style async |
| 73 |
> |
* tasks that are never joined often fall into this category. (2) To |
| 74 |
> |
* minimize resource impact, tasks should be small; ideally performing |
| 75 |
> |
* only the (possibly) blocking action. (3) Unless the {@link |
| 76 |
> |
* ForkJoinPool.ManagedBlocker} API is used, or the number of possibly |
| 77 |
> |
* blocked tasks is known to be less than the pool's {@link |
| 78 |
> |
* ForkJoinPool#getParallelism} level, the pool cannot guarantee that |
| 79 |
> |
* enough threads will be available to ensure progress or good |
| 80 |
> |
* performance. |
| 81 |
|
* |
| 82 |
|
* <p>The primary method for awaiting completion and extracting |
| 83 |
|
* results of a task is {@link #join}, but there are several variants: |
| 93 |
|
* performs the most common form of parallel invocation: forking a set |
| 94 |
|
* of tasks and joining them all. |
| 95 |
|
* |
| 96 |
+ |
* <p>In the most typical usages, a fork-join pair act like a a call |
| 97 |
+ |
* (fork) and return (join) from a parallel recursive function. As is |
| 98 |
+ |
* the case with other forms of recursive calls, returns (joins) |
| 99 |
+ |
* should be performed innermost-first. For example, {@code a.fork(); |
| 100 |
+ |
* b.fork(); b.join(); a.join();} is likely to be substantially more |
| 101 |
+ |
* efficient than joining {@code a} before {@code b}. |
| 102 |
+ |
* |
| 103 |
|
* <p>The execution status of tasks may be queried at several levels |
| 104 |
|
* of detail: {@link #isDone} is true if a task completed in any way |
| 105 |
|
* (including the case where a task was cancelled without executing); |
| 136 |
|
* supports other methods and techniques (for example the use of |
| 137 |
|
* {@link Phaser}, {@link #helpQuiesce}, and {@link #complete}) that |
| 138 |
|
* may be of use in constructing custom subclasses for problems that |
| 139 |
< |
* are not statically structured as DAGs. |
| 139 |
> |
* are not statically structured as DAGs. To support such usages a |
| 140 |
> |
* ForkJoinTask may be atomically <em>marked</em> using {@link |
| 141 |
> |
* #markForkJoinTask} and checked for marking using {@link |
| 142 |
> |
* #isMarkedForkJoinTask}. The ForkJoinTask implementation does not |
| 143 |
> |
* use these {@code protected} methods or marks for any purpose, but |
| 144 |
> |
* they may be of use in the construction of specialized subclasses. |
| 145 |
> |
* For example, parallel graph traversals can use the supplied methods |
| 146 |
> |
* to avoid revisiting nodes/tasks that have already been |
| 147 |
> |
* processed. Also, completion based designs can use them to record |
| 148 |
> |
* that one subtask has completed. (Method names for marking are bulky |
| 149 |
> |
* in part to encourage definition of methods that reflect their usage |
| 150 |
> |
* patterns.) |
| 151 |
|
* |
| 152 |
|
* <p>Most base support methods are {@code final}, to prevent |
| 153 |
|
* overriding of implementations that are intrinsically tied to the |
| 187 |
|
* See the internal documentation of class ForkJoinPool for a |
| 188 |
|
* general implementation overview. ForkJoinTasks are mainly |
| 189 |
|
* responsible for maintaining their "status" field amidst relays |
| 190 |
< |
* to methods in ForkJoinWorkerThread and ForkJoinPool. The |
| 191 |
< |
* methods of this class are more-or-less layered into (1) basic |
| 192 |
< |
* status maintenance (2) execution and awaiting completion (3) |
| 193 |
< |
* user-level methods that additionally report results. This is |
| 194 |
< |
* sometimes hard to see because this file orders exported methods |
| 195 |
< |
* in a way that flows well in javadocs. |
| 190 |
> |
* to methods in ForkJoinWorkerThread and ForkJoinPool. |
| 191 |
> |
* |
| 192 |
> |
* The methods of this class are more-or-less layered into |
| 193 |
> |
* (1) basic status maintenance |
| 194 |
> |
* (2) execution and awaiting completion |
| 195 |
> |
* (3) user-level methods that additionally report results. |
| 196 |
> |
* This is sometimes hard to see because this file orders exported |
| 197 |
> |
* methods in a way that flows well in javadocs. |
| 198 |
> |
*/ |
| 199 |
> |
|
| 200 |
> |
/** |
| 201 |
> |
* The number of times to try to help join a task without any |
| 202 |
> |
* apparent progress before giving up and blocking. The value is |
| 203 |
> |
* arbitrary but should be large enough to cope with transient |
| 204 |
> |
* stalls (due to GC etc) that can cause helping methods not to be |
| 205 |
> |
* able to proceed because other workers have not progressed to |
| 206 |
> |
* the point where subtasks can be found or taken. |
| 207 |
|
*/ |
| 208 |
+ |
private static final int HELP_RETRIES = 32; |
| 209 |
|
|
| 210 |
|
/* |
| 211 |
|
* The status field holds run control status bits packed into a |
| 226 |
|
|
| 227 |
|
/** The run status of this task */ |
| 228 |
|
volatile int status; // accessed directly by pool and workers |
| 229 |
< |
private static final int NORMAL = -1; |
| 230 |
< |
private static final int CANCELLED = -2; |
| 231 |
< |
private static final int EXCEPTIONAL = -3; |
| 232 |
< |
private static final int SIGNAL = 1; |
| 229 |
> |
static final int NORMAL = 0xfffffffc; // negative with low 2 bits 0 |
| 230 |
> |
static final int CANCELLED = 0xfffffff8; // must be < NORMAL |
| 231 |
> |
static final int EXCEPTIONAL = 0xfffffff4; // must be < CANCELLED |
| 232 |
> |
static final int SIGNAL = 0x00000001; |
| 233 |
> |
static final int MARKED = 0x00000002; |
| 234 |
|
|
| 235 |
|
/** |
| 236 |
< |
* Marks completion and wakes up threads waiting to join this task, |
| 237 |
< |
* also clearing signal request bits. |
| 236 |
> |
* Marks completion and wakes up threads waiting to join this |
| 237 |
> |
* task, also clearing signal request bits. A specialization for |
| 238 |
> |
* NORMAL completion is in method doExec |
| 239 |
|
* |
| 240 |
|
* @param completion one of NORMAL, CANCELLED, EXCEPTIONAL |
| 241 |
|
* @return completion status on exit |
| 244 |
|
for (int s;;) { |
| 245 |
|
if ((s = status) < 0) |
| 246 |
|
return s; |
| 247 |
< |
if (UNSAFE.compareAndSwapInt(this, statusOffset, s, completion)) { |
| 248 |
< |
if (s != 0) |
| 247 |
> |
if (U.compareAndSwapInt(this, STATUS, s, (s & ~SIGNAL)|completion)) { |
| 248 |
> |
if ((s & SIGNAL) != 0) |
| 249 |
|
synchronized (this) { notifyAll(); } |
| 250 |
|
return completion; |
| 251 |
|
} |
| 253 |
|
} |
| 254 |
|
|
| 255 |
|
/** |
| 256 |
< |
* Tries to block a worker thread until completed or timed out. |
| 257 |
< |
* Uses Object.wait time argument conventions. |
| 258 |
< |
* May fail on contention or interrupt. |
| 256 |
> |
* Primary execution method for stolen tasks. Unless done, calls |
| 257 |
> |
* exec and records status if completed, but doesn't wait for |
| 258 |
> |
* completion otherwise. |
| 259 |
|
* |
| 260 |
< |
* @param millis if > 0, wait time. |
| 260 |
> |
* @return status on exit from this method |
| 261 |
|
*/ |
| 262 |
< |
final void tryAwaitDone(long millis) { |
| 263 |
< |
int s; |
| 264 |
< |
try { |
| 265 |
< |
if (((s = status) > 0 || |
| 266 |
< |
(s == 0 && |
| 267 |
< |
UNSAFE.compareAndSwapInt(this, statusOffset, 0, SIGNAL))) && |
| 268 |
< |
status > 0) { |
| 269 |
< |
synchronized (this) { |
| 270 |
< |
if (status > 0) |
| 271 |
< |
wait(millis); |
| 262 |
> |
final int doExec() { |
| 263 |
> |
int s; boolean completed; |
| 264 |
> |
if ((s = status) >= 0) { |
| 265 |
> |
try { |
| 266 |
> |
completed = exec(); |
| 267 |
> |
} catch (Throwable rex) { |
| 268 |
> |
return setExceptionalCompletion(rex); |
| 269 |
> |
} |
| 270 |
> |
while ((s = status) >= 0 && completed) { |
| 271 |
> |
if (U.compareAndSwapInt(this, STATUS, s, (s & ~SIGNAL)|NORMAL)) { |
| 272 |
> |
if ((s & SIGNAL) != 0) |
| 273 |
> |
synchronized (this) { notifyAll(); } |
| 274 |
> |
return NORMAL; |
| 275 |
|
} |
| 276 |
|
} |
| 238 |
– |
} catch (InterruptedException ie) { |
| 239 |
– |
// caller must check termination |
| 277 |
|
} |
| 278 |
+ |
return s; |
| 279 |
|
} |
| 280 |
|
|
| 281 |
|
/** |
| 288 |
|
boolean interrupted = false; |
| 289 |
|
synchronized (this) { |
| 290 |
|
while ((s = status) >= 0) { |
| 291 |
< |
if (s == 0) |
| 254 |
< |
UNSAFE.compareAndSwapInt(this, statusOffset, |
| 255 |
< |
0, SIGNAL); |
| 256 |
< |
else { |
| 291 |
> |
if (U.compareAndSwapInt(this, STATUS, s, s | SIGNAL)) { |
| 292 |
|
try { |
| 293 |
|
wait(); |
| 294 |
|
} catch (InterruptedException ie) { |
| 314 |
|
if ((s = status) >= 0) { |
| 315 |
|
synchronized (this) { |
| 316 |
|
while ((s = status) >= 0) { |
| 317 |
< |
if (s == 0) |
| 283 |
< |
UNSAFE.compareAndSwapInt(this, statusOffset, |
| 284 |
< |
0, SIGNAL); |
| 285 |
< |
else { |
| 317 |
> |
if (U.compareAndSwapInt(this, STATUS, s, s | SIGNAL)) { |
| 318 |
|
wait(millis); |
| 319 |
|
if (millis > 0L) |
| 320 |
|
break; |
| 325 |
|
return s; |
| 326 |
|
} |
| 327 |
|
|
| 328 |
+ |
|
| 329 |
|
/** |
| 330 |
< |
* Primary execution method for stolen tasks. Unless done, calls |
| 331 |
< |
* exec and records status if completed, but doesn't wait for |
| 332 |
< |
* completion otherwise. |
| 330 |
> |
* Implementation for join, get, quietlyJoin. Directly handles |
| 331 |
> |
* only cases of already-completed, external wait, and |
| 332 |
> |
* unfork+exec. Others are relayed to awaitJoin. |
| 333 |
> |
* |
| 334 |
> |
* @return status upon completion |
| 335 |
|
*/ |
| 336 |
< |
final void doExec() { |
| 337 |
< |
if (status >= 0) { |
| 338 |
< |
boolean completed; |
| 339 |
< |
try { |
| 340 |
< |
completed = exec(); |
| 341 |
< |
} catch (Throwable rex) { |
| 342 |
< |
setExceptionalCompletion(rex); |
| 343 |
< |
return; |
| 309 |
< |
} |
| 310 |
< |
if (completed) |
| 311 |
< |
setCompletion(NORMAL); // must be outside try block |
| 336 |
> |
private int doJoin() { |
| 337 |
> |
int s; Thread t; ForkJoinWorkerThread wt; ForkJoinPool.WorkQueue w; |
| 338 |
> |
if ((s = status) >= 0) { |
| 339 |
> |
if (!((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)) |
| 340 |
> |
s = externalAwaitDone(); |
| 341 |
> |
else if (!(w = (wt = (ForkJoinWorkerThread)t).workQueue). |
| 342 |
> |
tryUnpush(this) || (s = doExec()) >= 0) |
| 343 |
> |
s = awaitJoin(w, wt.pool); |
| 344 |
|
} |
| 345 |
+ |
return s; |
| 346 |
|
} |
| 347 |
|
|
| 348 |
|
/** |
| 349 |
< |
* Primary mechanics for join, get, quietlyJoin. |
| 349 |
> |
* Helps and/or blocks until joined. |
| 350 |
> |
* |
| 351 |
> |
* @param w the joiner |
| 352 |
> |
* @param p the pool |
| 353 |
|
* @return status upon completion |
| 354 |
|
*/ |
| 355 |
< |
private int doJoin() { |
| 356 |
< |
Thread t; ForkJoinWorkerThread w; int s; boolean completed; |
| 357 |
< |
if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) { |
| 358 |
< |
if ((s = status) < 0) |
| 359 |
< |
return s; |
| 360 |
< |
if ((w = (ForkJoinWorkerThread)t).unpushTask(this)) { |
| 355 |
> |
private int awaitJoin(ForkJoinPool.WorkQueue w, ForkJoinPool p) { |
| 356 |
> |
int s; |
| 357 |
> |
ForkJoinTask<?> prevJoin = w.currentJoin; |
| 358 |
> |
w.currentJoin = this; |
| 359 |
> |
for (int k = HELP_RETRIES; (s = status) >= 0;) { |
| 360 |
> |
if ((w.queueSize() > 0) ? |
| 361 |
> |
w.tryRemoveAndExec(this) : // self-help |
| 362 |
> |
p.tryHelpStealer(w, this)) // help process tasks |
| 363 |
> |
k = HELP_RETRIES; // reset if made progress |
| 364 |
> |
else if ((s = status) < 0) // recheck |
| 365 |
> |
break; |
| 366 |
> |
else if (--k > 0) { |
| 367 |
> |
if ((k & 3) == 1) |
| 368 |
> |
Thread.yield(); // occasionally yield |
| 369 |
> |
} |
| 370 |
> |
else if (k == 0) |
| 371 |
> |
p.tryPollForAndExec(w, this); // uncommon self-help case |
| 372 |
> |
else if (p.tryCompensate()) { // true if can block |
| 373 |
|
try { |
| 374 |
< |
completed = exec(); |
| 375 |
< |
} catch (Throwable rex) { |
| 376 |
< |
return setExceptionalCompletion(rex); |
| 374 |
> |
int ss = status; |
| 375 |
> |
if (ss >= 0 && // assert need signal |
| 376 |
> |
U.compareAndSwapInt(this, STATUS, ss, ss | SIGNAL)) { |
| 377 |
> |
synchronized (this) { |
| 378 |
> |
if (status >= 0) // block |
| 379 |
> |
wait(); |
| 380 |
> |
} |
| 381 |
> |
} |
| 382 |
> |
} catch (InterruptedException ignore) { |
| 383 |
> |
} finally { |
| 384 |
> |
p.incrementActiveCount(); // re-activate |
| 385 |
|
} |
| 330 |
– |
if (completed) |
| 331 |
– |
return setCompletion(NORMAL); |
| 386 |
|
} |
| 333 |
– |
return w.joinTask(this); |
| 387 |
|
} |
| 388 |
< |
else |
| 389 |
< |
return externalAwaitDone(); |
| 388 |
> |
w.currentJoin = prevJoin; |
| 389 |
> |
return s; |
| 390 |
|
} |
| 391 |
|
|
| 392 |
|
/** |
| 393 |
< |
* Primary mechanics for invoke, quietlyInvoke. |
| 393 |
> |
* Implementation for invoke, quietlyInvoke. |
| 394 |
> |
* |
| 395 |
|
* @return status upon completion |
| 396 |
|
*/ |
| 397 |
|
private int doInvoke() { |
| 398 |
< |
int s; boolean completed; |
| 399 |
< |
if ((s = status) < 0) |
| 400 |
< |
return s; |
| 401 |
< |
try { |
| 402 |
< |
completed = exec(); |
| 403 |
< |
} catch (Throwable rex) { |
| 404 |
< |
return setExceptionalCompletion(rex); |
| 398 |
> |
int s; Thread t; |
| 399 |
> |
if ((s = doExec()) >= 0) { |
| 400 |
> |
if (!((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)) |
| 401 |
> |
s = externalAwaitDone(); |
| 402 |
> |
else { |
| 403 |
> |
ForkJoinWorkerThread wt = (ForkJoinWorkerThread)t; |
| 404 |
> |
s = awaitJoin(wt.workQueue, wt.pool); |
| 405 |
> |
} |
| 406 |
|
} |
| 407 |
< |
if (completed) |
| 353 |
< |
return setCompletion(NORMAL); |
| 354 |
< |
else |
| 355 |
< |
return doJoin(); |
| 407 |
> |
return s; |
| 408 |
|
} |
| 409 |
|
|
| 410 |
|
// Exception table support |
| 479 |
|
} |
| 480 |
|
|
| 481 |
|
/** |
| 482 |
+ |
* Cancels, ignoring any exceptions thrown by cancel. Used during |
| 483 |
+ |
* worker and pool shutdown. Cancel is spec'ed not to throw any |
| 484 |
+ |
* exceptions, but if it does anyway, we have no recourse during |
| 485 |
+ |
* shutdown, so guard against this case. |
| 486 |
+ |
*/ |
| 487 |
+ |
static final void cancelIgnoringExceptions(ForkJoinTask<?> t) { |
| 488 |
+ |
if (t != null && t.status >= 0) { |
| 489 |
+ |
try { |
| 490 |
+ |
t.cancel(false); |
| 491 |
+ |
} catch (Throwable ignore) { |
| 492 |
+ |
} |
| 493 |
+ |
} |
| 494 |
+ |
} |
| 495 |
+ |
|
| 496 |
+ |
/** |
| 497 |
|
* Removes exception node and clears status |
| 498 |
|
*/ |
| 499 |
|
private void clearExceptionalCompletion() { |
| 632 |
|
if ((s = status) == CANCELLED) |
| 633 |
|
throw new CancellationException(); |
| 634 |
|
if (s == EXCEPTIONAL && (ex = getThrowableException()) != null) |
| 635 |
< |
UNSAFE.throwException(ex); |
| 635 |
> |
U.throwException(ex); |
| 636 |
|
return getRawResult(); |
| 637 |
|
} |
| 638 |
|
|
| 657 |
|
* @return {@code this}, to simplify usage |
| 658 |
|
*/ |
| 659 |
|
public final ForkJoinTask<V> fork() { |
| 660 |
< |
((ForkJoinWorkerThread) Thread.currentThread()) |
| 661 |
< |
.pushTask(this); |
| 660 |
> |
ForkJoinWorkerThread wt; |
| 661 |
> |
(wt = (ForkJoinWorkerThread)Thread.currentThread()). |
| 662 |
> |
workQueue.push(this, wt.pool); |
| 663 |
|
return this; |
| 664 |
|
} |
| 665 |
|
|
| 765 |
|
if (t != null) { |
| 766 |
|
if (ex != null) |
| 767 |
|
t.cancel(false); |
| 768 |
< |
else if (t.doJoin() < NORMAL && ex == null) |
| 768 |
> |
else if (t.doJoin() < NORMAL) |
| 769 |
|
ex = t.getException(); |
| 770 |
|
} |
| 771 |
|
} |
| 772 |
|
if (ex != null) |
| 773 |
< |
UNSAFE.throwException(ex); |
| 773 |
> |
U.throwException(ex); |
| 774 |
|
} |
| 775 |
|
|
| 776 |
|
/** |
| 822 |
|
if (t != null) { |
| 823 |
|
if (ex != null) |
| 824 |
|
t.cancel(false); |
| 825 |
< |
else if (t.doJoin() < NORMAL && ex == null) |
| 825 |
> |
else if (t.doJoin() < NORMAL) |
| 826 |
|
ex = t.getException(); |
| 827 |
|
} |
| 828 |
|
} |
| 829 |
|
if (ex != null) |
| 830 |
< |
UNSAFE.throwException(ex); |
| 830 |
> |
U.throwException(ex); |
| 831 |
|
return tasks; |
| 832 |
|
} |
| 833 |
|
|
| 862 |
|
return setCompletion(CANCELLED) == CANCELLED; |
| 863 |
|
} |
| 864 |
|
|
| 797 |
– |
/** |
| 798 |
– |
* Cancels, ignoring any exceptions thrown by cancel. Used during |
| 799 |
– |
* worker and pool shutdown. Cancel is spec'ed not to throw any |
| 800 |
– |
* exceptions, but if it does anyway, we have no recourse during |
| 801 |
– |
* shutdown, so guard against this case. |
| 802 |
– |
*/ |
| 803 |
– |
final void cancelIgnoringExceptions() { |
| 804 |
– |
try { |
| 805 |
– |
cancel(false); |
| 806 |
– |
} catch (Throwable ignore) { |
| 807 |
– |
} |
| 808 |
– |
} |
| 809 |
– |
|
| 865 |
|
public final boolean isDone() { |
| 866 |
|
return status < 0; |
| 867 |
|
} |
| 985 |
|
*/ |
| 986 |
|
public final V get(long timeout, TimeUnit unit) |
| 987 |
|
throws InterruptedException, ExecutionException, TimeoutException { |
| 988 |
+ |
// Messy in part because we measure in nanos, but wait in millis |
| 989 |
+ |
int s; long millis, nanos; |
| 990 |
|
Thread t = Thread.currentThread(); |
| 991 |
< |
if (t instanceof ForkJoinWorkerThread) { |
| 992 |
< |
ForkJoinWorkerThread w = (ForkJoinWorkerThread) t; |
| 993 |
< |
long nanos = unit.toNanos(timeout); |
| 994 |
< |
if (status >= 0) { |
| 995 |
< |
boolean completed = false; |
| 996 |
< |
if (w.unpushTask(this)) { |
| 997 |
< |
try { |
| 998 |
< |
completed = exec(); |
| 999 |
< |
} catch (Throwable rex) { |
| 1000 |
< |
setExceptionalCompletion(rex); |
| 991 |
> |
if (!(t instanceof ForkJoinWorkerThread)) { |
| 992 |
> |
if ((millis = unit.toMillis(timeout)) > 0L) |
| 993 |
> |
s = externalInterruptibleAwaitDone(millis); |
| 994 |
> |
else |
| 995 |
> |
s = status; |
| 996 |
> |
} |
| 997 |
> |
else if ((s = status) >= 0 && (nanos = unit.toNanos(timeout)) > 0L) { |
| 998 |
> |
long deadline = System.nanoTime() + nanos; |
| 999 |
> |
ForkJoinWorkerThread wt = (ForkJoinWorkerThread)t; |
| 1000 |
> |
ForkJoinPool.WorkQueue w = wt.workQueue; |
| 1001 |
> |
ForkJoinPool p = wt.pool; |
| 1002 |
> |
if (w.tryUnpush(this)) |
| 1003 |
> |
doExec(); |
| 1004 |
> |
boolean blocking = false; |
| 1005 |
> |
try { |
| 1006 |
> |
while ((s = status) >= 0) { |
| 1007 |
> |
if (w.runState < 0) |
| 1008 |
> |
cancelIgnoringExceptions(this); |
| 1009 |
> |
else if (!blocking) |
| 1010 |
> |
blocking = p.tryCompensate(); |
| 1011 |
> |
else { |
| 1012 |
> |
millis = TimeUnit.NANOSECONDS.toMillis(nanos); |
| 1013 |
> |
if (millis > 0L && |
| 1014 |
> |
U.compareAndSwapInt(this, STATUS, s, s | SIGNAL)) { |
| 1015 |
> |
try { |
| 1016 |
> |
synchronized (this) { |
| 1017 |
> |
if (status >= 0) |
| 1018 |
> |
wait(millis); |
| 1019 |
> |
} |
| 1020 |
> |
} catch (InterruptedException ie) { |
| 1021 |
> |
} |
| 1022 |
> |
} |
| 1023 |
> |
if ((s = status) < 0 || |
| 1024 |
> |
(nanos = deadline - System.nanoTime()) <= 0L) |
| 1025 |
> |
break; |
| 1026 |
|
} |
| 1027 |
|
} |
| 1028 |
< |
if (completed) |
| 1029 |
< |
setCompletion(NORMAL); |
| 1030 |
< |
else if (status >= 0 && nanos > 0) |
| 949 |
< |
w.pool.timedAwaitJoin(this, nanos); |
| 1028 |
> |
} finally { |
| 1029 |
> |
if (blocking) |
| 1030 |
> |
p.incrementActiveCount(); |
| 1031 |
|
} |
| 1032 |
|
} |
| 952 |
– |
else { |
| 953 |
– |
long millis = unit.toMillis(timeout); |
| 954 |
– |
if (millis > 0) |
| 955 |
– |
externalInterruptibleAwaitDone(millis); |
| 956 |
– |
} |
| 957 |
– |
int s = status; |
| 1033 |
|
if (s != NORMAL) { |
| 1034 |
|
Throwable ex; |
| 1035 |
|
if (s == CANCELLED) |
| 1075 |
|
* ClassCastException}. |
| 1076 |
|
*/ |
| 1077 |
|
public static void helpQuiesce() { |
| 1078 |
< |
((ForkJoinWorkerThread) Thread.currentThread()) |
| 1079 |
< |
.helpQuiescePool(); |
| 1078 |
> |
ForkJoinWorkerThread wt = |
| 1079 |
> |
(ForkJoinWorkerThread)Thread.currentThread(); |
| 1080 |
> |
wt.pool.helpQuiescePool(wt.workQueue); |
| 1081 |
|
} |
| 1082 |
|
|
| 1083 |
|
/** |
| 1145 |
|
* @return {@code true} if unforked |
| 1146 |
|
*/ |
| 1147 |
|
public boolean tryUnfork() { |
| 1148 |
< |
return ((ForkJoinWorkerThread) Thread.currentThread()) |
| 1149 |
< |
.unpushTask(this); |
| 1148 |
> |
return ((ForkJoinWorkerThread)Thread.currentThread()) |
| 1149 |
> |
.workQueue.tryUnpush(this); |
| 1150 |
|
} |
| 1151 |
|
|
| 1152 |
|
/** |
| 1165 |
|
*/ |
| 1166 |
|
public static int getQueuedTaskCount() { |
| 1167 |
|
return ((ForkJoinWorkerThread) Thread.currentThread()) |
| 1168 |
< |
.getQueueSize(); |
| 1168 |
> |
.workQueue.queueSize(); |
| 1169 |
|
} |
| 1170 |
|
|
| 1171 |
|
/** |
| 1187 |
|
* @return the surplus number of tasks, which may be negative |
| 1188 |
|
*/ |
| 1189 |
|
public static int getSurplusQueuedTaskCount() { |
| 1190 |
< |
return ((ForkJoinWorkerThread) Thread.currentThread()) |
| 1191 |
< |
.getEstimatedSurplusTaskCount(); |
| 1190 |
> |
/* |
| 1191 |
> |
* The aim of this method is to return a cheap heuristic guide |
| 1192 |
> |
* for task partitioning when programmers, frameworks, tools, |
| 1193 |
> |
* or languages have little or no idea about task granularity. |
| 1194 |
> |
* In essence by offering this method, we ask users only about |
| 1195 |
> |
* tradeoffs in overhead vs expected throughput and its |
| 1196 |
> |
* variance, rather than how finely to partition tasks. |
| 1197 |
> |
* |
| 1198 |
> |
* In a steady state strict (tree-structured) computation, |
| 1199 |
> |
* each thread makes available for stealing enough tasks for |
| 1200 |
> |
* other threads to remain active. Inductively, if all threads |
| 1201 |
> |
* play by the same rules, each thread should make available |
| 1202 |
> |
* only a constant number of tasks. |
| 1203 |
> |
* |
| 1204 |
> |
* The minimum useful constant is just 1. But using a value of |
| 1205 |
> |
* 1 would require immediate replenishment upon each steal to |
| 1206 |
> |
* maintain enough tasks, which is infeasible. Further, |
| 1207 |
> |
* partitionings/granularities of offered tasks should |
| 1208 |
> |
* minimize steal rates, which in general means that threads |
| 1209 |
> |
* nearer the top of computation tree should generate more |
| 1210 |
> |
* than those nearer the bottom. In perfect steady state, each |
| 1211 |
> |
* thread is at approximately the same level of computation |
| 1212 |
> |
* tree. However, producing extra tasks amortizes the |
| 1213 |
> |
* uncertainty of progress and diffusion assumptions. |
| 1214 |
> |
* |
| 1215 |
> |
* So, users will want to use values larger, but not much |
| 1216 |
> |
* larger than 1 to both smooth over transient shortages and |
| 1217 |
> |
* hedge against uneven progress; as traded off against the |
| 1218 |
> |
* cost of extra task overhead. We leave the user to pick a |
| 1219 |
> |
* threshold value to compare with the results of this call to |
| 1220 |
> |
* guide decisions, but recommend values such as 3. |
| 1221 |
> |
* |
| 1222 |
> |
* When all threads are active, it is on average OK to |
| 1223 |
> |
* estimate surplus strictly locally. In steady-state, if one |
| 1224 |
> |
* thread is maintaining say 2 surplus tasks, then so are |
| 1225 |
> |
* others. So we can just use estimated queue length. |
| 1226 |
> |
* However, this strategy alone leads to serious mis-estimates |
| 1227 |
> |
* in some non-steady-state conditions (ramp-up, ramp-down, |
| 1228 |
> |
* other stalls). We can detect many of these by further |
| 1229 |
> |
* considering the number of "idle" threads, that are known to |
| 1230 |
> |
* have zero queued tasks, so compensate by a factor of |
| 1231 |
> |
* (#idle/#active) threads. |
| 1232 |
> |
*/ |
| 1233 |
> |
ForkJoinWorkerThread wt = |
| 1234 |
> |
(ForkJoinWorkerThread)Thread.currentThread(); |
| 1235 |
> |
return wt.workQueue.queueSize() - wt.pool.idlePerActive(); |
| 1236 |
|
} |
| 1237 |
|
|
| 1238 |
|
// Extension methods |
| 1289 |
|
* @return the next task, or {@code null} if none are available |
| 1290 |
|
*/ |
| 1291 |
|
protected static ForkJoinTask<?> peekNextLocalTask() { |
| 1292 |
< |
return ((ForkJoinWorkerThread) Thread.currentThread()) |
| 1173 |
< |
.peekTask(); |
| 1292 |
> |
return ((ForkJoinWorkerThread) Thread.currentThread()).workQueue.peek(); |
| 1293 |
|
} |
| 1294 |
|
|
| 1295 |
|
/** |
| 1308 |
|
*/ |
| 1309 |
|
protected static ForkJoinTask<?> pollNextLocalTask() { |
| 1310 |
|
return ((ForkJoinWorkerThread) Thread.currentThread()) |
| 1311 |
< |
.pollLocalTask(); |
| 1311 |
> |
.workQueue.nextLocalTask(); |
| 1312 |
|
} |
| 1313 |
|
|
| 1314 |
|
/** |
| 1330 |
|
* @return a task, or {@code null} if none are available |
| 1331 |
|
*/ |
| 1332 |
|
protected static ForkJoinTask<?> pollTask() { |
| 1333 |
< |
return ((ForkJoinWorkerThread) Thread.currentThread()) |
| 1334 |
< |
.pollTask(); |
| 1333 |
> |
ForkJoinWorkerThread wt = |
| 1334 |
> |
(ForkJoinWorkerThread)Thread.currentThread(); |
| 1335 |
> |
return wt.pool.nextTaskFor(wt.workQueue); |
| 1336 |
> |
} |
| 1337 |
> |
|
| 1338 |
> |
// Mark-bit operations |
| 1339 |
> |
|
| 1340 |
> |
/** |
| 1341 |
> |
* Returns true if this task is marked. |
| 1342 |
> |
* |
| 1343 |
> |
* @return true if this task is marked |
| 1344 |
> |
* @since 1.8 |
| 1345 |
> |
*/ |
| 1346 |
> |
public final boolean isMarkedForkJoinTask() { |
| 1347 |
> |
return (status & MARKED) != 0; |
| 1348 |
> |
} |
| 1349 |
> |
|
| 1350 |
> |
/** |
| 1351 |
> |
* Atomically sets the mark on this task. |
| 1352 |
> |
* |
| 1353 |
> |
* @return true if this task was previously unmarked |
| 1354 |
> |
* @since 1.8 |
| 1355 |
> |
*/ |
| 1356 |
> |
public final boolean markForkJoinTask() { |
| 1357 |
> |
for (int s;;) { |
| 1358 |
> |
if (((s = status) & MARKED) != 0) |
| 1359 |
> |
return false; |
| 1360 |
> |
if (U.compareAndSwapInt(this, STATUS, s, s | MARKED)) |
| 1361 |
> |
return true; |
| 1362 |
> |
} |
| 1363 |
> |
} |
| 1364 |
> |
|
| 1365 |
> |
/** |
| 1366 |
> |
* Atomically clears the mark on this task. |
| 1367 |
> |
* |
| 1368 |
> |
* @return true if this task was previously marked |
| 1369 |
> |
* @since 1.8 |
| 1370 |
> |
*/ |
| 1371 |
> |
public final boolean unmarkForkJoinTask() { |
| 1372 |
> |
for (int s;;) { |
| 1373 |
> |
if (((s = status) & MARKED) == 0) |
| 1374 |
> |
return false; |
| 1375 |
> |
if (U.compareAndSwapInt(this, STATUS, s, s & ~MARKED)) |
| 1376 |
> |
return true; |
| 1377 |
> |
} |
| 1378 |
|
} |
| 1379 |
|
|
| 1380 |
|
/** |
| 1475 |
|
private static final long serialVersionUID = -7721805057305804111L; |
| 1476 |
|
|
| 1477 |
|
/** |
| 1478 |
< |
* Saves the state to a stream (that is, serializes it). |
| 1478 |
> |
* Saves this task to a stream (that is, serializes it). |
| 1479 |
|
* |
| 1480 |
|
* @serialData the current run status and the exception thrown |
| 1481 |
|
* during execution, or {@code null} if none |
| 1320 |
– |
* @param s the stream |
| 1482 |
|
*/ |
| 1483 |
|
private void writeObject(java.io.ObjectOutputStream s) |
| 1484 |
|
throws java.io.IOException { |
| 1487 |
|
} |
| 1488 |
|
|
| 1489 |
|
/** |
| 1490 |
< |
* Reconstitutes the instance from a stream (that is, deserializes it). |
| 1330 |
< |
* |
| 1331 |
< |
* @param s the stream |
| 1490 |
> |
* Reconstitutes this task from a stream (that is, deserializes it). |
| 1491 |
|
*/ |
| 1492 |
|
private void readObject(java.io.ObjectInputStream s) |
| 1493 |
|
throws java.io.IOException, ClassNotFoundException { |
| 1498 |
|
} |
| 1499 |
|
|
| 1500 |
|
// Unsafe mechanics |
| 1501 |
< |
private static final sun.misc.Unsafe UNSAFE; |
| 1502 |
< |
private static final long statusOffset; |
| 1501 |
> |
private static final sun.misc.Unsafe U; |
| 1502 |
> |
private static final long STATUS; |
| 1503 |
|
static { |
| 1504 |
|
exceptionTableLock = new ReentrantLock(); |
| 1505 |
|
exceptionTableRefQueue = new ReferenceQueue<Object>(); |
| 1506 |
|
exceptionTable = new ExceptionNode[EXCEPTION_MAP_CAPACITY]; |
| 1507 |
|
try { |
| 1508 |
< |
UNSAFE = getUnsafe(); |
| 1509 |
< |
statusOffset = UNSAFE.objectFieldOffset |
| 1508 |
> |
U = getUnsafe(); |
| 1509 |
> |
STATUS = U.objectFieldOffset |
| 1510 |
|
(ForkJoinTask.class.getDeclaredField("status")); |
| 1511 |
|
} catch (Exception e) { |
| 1512 |
|
throw new Error(e); |
