ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/jsr166/jsr166/src/jsr166y/ForkJoinWorkerThread.java
Revision: 1.23
Committed: Fri Jul 31 16:27:08 2009 UTC (14 years, 9 months ago) by dl
Branch: MAIN
Changes since 1.22: +34 -7 lines
Log Message:
Refactor Adapted tasks into ForkJoinTask; mesh peek/pollNextLocalTask specs and code

File Contents

# Content
1 /*
2 * Written by Doug Lea with assistance from members of JCP JSR-166
3 * Expert Group and released to the public domain, as explained at
4 * http://creativecommons.org/licenses/publicdomain
5 */
6
7 package jsr166y;
8
9 import java.util.concurrent.*;
10
11 import java.util.Collection;
12
13 /**
14 * A thread managed by a {@link ForkJoinPool}. This class is
15 * subclassable solely for the sake of adding functionality -- there
16 * are no overridable methods dealing with scheduling or
17 * execution. However, you can override initialization and termination
18 * methods surrounding the main task processing loop. If you do
19 * create such a subclass, you will also need to supply a custom
20 * ForkJoinWorkerThreadFactory to use it in a ForkJoinPool.
21 *
22 * @since 1.7
23 * @author Doug Lea
24 */
25 public class ForkJoinWorkerThread extends Thread {
26 /*
27 * Algorithm overview:
28 *
29 * 1. Work-Stealing: Work-stealing queues are special forms of
30 * Deques that support only three of the four possible
31 * end-operations -- push, pop, and deq (aka steal), and only do
32 * so under the constraints that push and pop are called only from
33 * the owning thread, while deq may be called from other threads.
34 * (If you are unfamiliar with them, you probably want to read
35 * Herlihy and Shavit's book "The Art of Multiprocessor
36 * programming", chapter 16 describing these in more detail before
37 * proceeding.) The main work-stealing queue design is roughly
38 * similar to "Dynamic Circular Work-Stealing Deque" by David
39 * Chase and Yossi Lev, SPAA 2005
40 * (http://research.sun.com/scalable/pubs/index.html). The main
41 * difference ultimately stems from gc requirements that we null
42 * out taken slots as soon as we can, to maintain as small a
43 * footprint as possible even in programs generating huge numbers
44 * of tasks. To accomplish this, we shift the CAS arbitrating pop
45 * vs deq (steal) from being on the indices ("base" and "sp") to
46 * the slots themselves (mainly via method "casSlotNull()"). So,
47 * both a successful pop and deq mainly entail CAS'ing a non-null
48 * slot to null. Because we rely on CASes of references, we do
49 * not need tag bits on base or sp. They are simple ints as used
50 * in any circular array-based queue (see for example ArrayDeque).
51 * Updates to the indices must still be ordered in a way that
52 * guarantees that (sp - base) > 0 means the queue is empty, but
53 * otherwise may err on the side of possibly making the queue
54 * appear nonempty when a push, pop, or deq have not fully
55 * committed. Note that this means that the deq operation,
56 * considered individually, is not wait-free. One thief cannot
57 * successfully continue until another in-progress one (or, if
58 * previously empty, a push) completes. However, in the
59 * aggregate, we ensure at least probabilistic non-blockingness. If
60 * an attempted steal fails, a thief always chooses a different
61 * random victim target to try next. So, in order for one thief to
62 * progress, it suffices for any in-progress deq or new push on
63 * any empty queue to complete. One reason this works well here is
64 * that apparently-nonempty often means soon-to-be-stealable,
65 * which gives threads a chance to activate if necessary before
66 * stealing (see below).
67 *
68 * This approach also enables support for "async mode" where local
69 * task processing is in FIFO, not LIFO order; simply by using a
70 * version of deq rather than pop when locallyFifo is true (as set
71 * by the ForkJoinPool). This allows use in message-passing
72 * frameworks in which tasks are never joined.
73 *
74 * Efficient implementation of this approach currently relies on
75 * an uncomfortable amount of "Unsafe" mechanics. To maintain
76 * correct orderings, reads and writes of variable base require
77 * volatile ordering. Variable sp does not require volatile write
78 * but needs cheaper store-ordering on writes. Because they are
79 * protected by volatile base reads, reads of the queue array and
80 * its slots do not need volatile load semantics, but writes (in
81 * push) require store order and CASes (in pop and deq) require
82 * (volatile) CAS semantics. Since these combinations aren't
83 * supported using ordinary volatiles, the only way to accomplish
84 * these efficiently is to use direct Unsafe calls. (Using external
85 * AtomicIntegers and AtomicReferenceArrays for the indices and
86 * array is significantly slower because of memory locality and
87 * indirection effects.) Further, performance on most platforms is
88 * very sensitive to placement and sizing of the (resizable) queue
89 * array. Even though these queues don't usually become all that
90 * big, the initial size must be large enough to counteract cache
91 * contention effects across multiple queues (especially in the
92 * presence of GC cardmarking). Also, to improve thread-locality,
93 * queues are currently initialized immediately after the thread
94 * gets the initial signal to start processing tasks. However,
95 * all queue-related methods except pushTask are written in a way
96 * that allows them to instead be lazily allocated and/or disposed
97 * of when empty. All together, these low-level implementation
98 * choices produce as much as a factor of 4 performance
99 * improvement compared to naive implementations, and enable the
100 * processing of billions of tasks per second, sometimes at the
101 * expense of ugliness.
102 *
103 * 2. Run control: The primary run control is based on a global
104 * counter (activeCount) held by the pool. It uses an algorithm
105 * similar to that in Herlihy and Shavit section 17.6 to cause
106 * threads to eventually block when all threads declare they are
107 * inactive. (See variable "scans".) For this to work, threads
108 * must be declared active when executing tasks, and before
109 * stealing a task. They must be inactive before blocking on the
110 * Pool Barrier (awaiting a new submission or other Pool
111 * event). In between, there is some free play which we take
112 * advantage of to avoid contention and rapid flickering of the
113 * global activeCount: If inactive, we activate only if a victim
114 * queue appears to be nonempty (see above). Similarly, a thread
115 * tries to inactivate only after a full scan of other threads.
116 * The net effect is that contention on activeCount is rarely a
117 * measurable performance issue. (There are also a few other cases
118 * where we scan for work rather than retry/block upon
119 * contention.)
120 *
121 * 3. Selection control. We maintain policy of always choosing to
122 * run local tasks rather than stealing, and always trying to
123 * steal tasks before trying to run a new submission. All steals
124 * are currently performed in randomly-chosen deq-order. It may be
125 * worthwhile to bias these with locality / anti-locality
126 * information, but doing this well probably requires more
127 * lower-level information from JVMs than currently provided.
128 */
129
130 /**
131 * Capacity of work-stealing queue array upon initialization.
132 * Must be a power of two. Initial size must be at least 2, but is
133 * padded to minimize cache effects.
134 */
135 private static final int INITIAL_QUEUE_CAPACITY = 1 << 13;
136
137 /**
138 * Maximum work-stealing queue array size. Must be less than or
139 * equal to 1 << 28 to ensure lack of index wraparound. (This
140 * is less than usual bounds, because we need leftshift by 3
141 * to be in int range).
142 */
143 private static final int MAXIMUM_QUEUE_CAPACITY = 1 << 28;
144
145 /**
146 * The pool this thread works in. Accessed directly by ForkJoinTask.
147 */
148 final ForkJoinPool pool;
149
150 /**
151 * The work-stealing queue array. Size must be a power of two.
152 * Initialized when thread starts, to improve memory locality.
153 */
154 private ForkJoinTask<?>[] queue;
155
156 /**
157 * Index (mod queue.length) of next queue slot to push to or pop
158 * from. It is written only by owner thread, via ordered store.
159 * Both sp and base are allowed to wrap around on overflow, but
160 * (sp - base) still estimates size.
161 */
162 private volatile int sp;
163
164 /**
165 * Index (mod queue.length) of least valid queue slot, which is
166 * always the next position to steal from if nonempty.
167 */
168 private volatile int base;
169
170 /**
171 * Activity status. When true, this worker is considered active.
172 * Must be false upon construction. It must be true when executing
173 * tasks, and BEFORE stealing a task. It must be false before
174 * calling pool.sync.
175 */
176 private boolean active;
177
178 /**
179 * Run state of this worker. Supports simple versions of the usual
180 * shutdown/shutdownNow control.
181 */
182 private volatile int runState;
183
184 /**
185 * Seed for random number generator for choosing steal victims.
186 * Uses Marsaglia xorshift. Must be nonzero upon initialization.
187 */
188 private int seed;
189
190 /**
191 * Number of steals, transferred to pool when idle
192 */
193 private int stealCount;
194
195 /**
196 * Index of this worker in pool array. Set once by pool before
197 * running, and accessed directly by pool during cleanup etc.
198 */
199 int poolIndex;
200
201 /**
202 * The last barrier event waited for. Accessed in pool callback
203 * methods, but only by current thread.
204 */
205 long lastEventCount;
206
207 /**
208 * True if use local fifo, not default lifo, for local polling
209 */
210 private boolean locallyFifo;
211
212 /**
213 * Creates a ForkJoinWorkerThread operating in the given pool.
214 *
215 * @param pool the pool this thread works in
216 * @throws NullPointerException if pool is null
217 */
218 protected ForkJoinWorkerThread(ForkJoinPool pool) {
219 if (pool == null) throw new NullPointerException();
220 this.pool = pool;
221 // Note: poolIndex is set by pool during construction
222 // Remaining initialization is deferred to onStart
223 }
224
225 // Public access methods
226
227 /**
228 * Returns the pool hosting this thread.
229 *
230 * @return the pool
231 */
232 public ForkJoinPool getPool() {
233 return pool;
234 }
235
236 /**
237 * Returns the index number of this thread in its pool. The
238 * returned value ranges from zero to the maximum number of
239 * threads (minus one) that have ever been created in the pool.
240 * This method may be useful for applications that track status or
241 * collect results per-worker rather than per-task.
242 *
243 * @return the index number
244 */
245 public int getPoolIndex() {
246 return poolIndex;
247 }
248
249 /**
250 * Establishes local first-in-first-out scheduling mode for forked
251 * tasks that are never joined.
252 *
253 * @param async if true, use locally FIFO scheduling
254 */
255 void setAsyncMode(boolean async) {
256 locallyFifo = async;
257 }
258
259 // Runstate management
260
261 // Runstate values. Order matters
262 private static final int RUNNING = 0;
263 private static final int SHUTDOWN = 1;
264 private static final int TERMINATING = 2;
265 private static final int TERMINATED = 3;
266
267 final boolean isShutdown() { return runState >= SHUTDOWN; }
268 final boolean isTerminating() { return runState >= TERMINATING; }
269 final boolean isTerminated() { return runState == TERMINATED; }
270 final boolean shutdown() { return transitionRunStateTo(SHUTDOWN); }
271 final boolean shutdownNow() { return transitionRunStateTo(TERMINATING); }
272
273 /**
274 * Transitions to at least the given state.
275 *
276 * @return {@code true} if not already at least at given state
277 */
278 private boolean transitionRunStateTo(int state) {
279 for (;;) {
280 int s = runState;
281 if (s >= state)
282 return false;
283 if (UNSAFE.compareAndSwapInt(this, runStateOffset, s, state))
284 return true;
285 }
286 }
287
288 /**
289 * Tries to set status to active; fails on contention.
290 */
291 private boolean tryActivate() {
292 if (!active) {
293 if (!pool.tryIncrementActiveCount())
294 return false;
295 active = true;
296 }
297 return true;
298 }
299
300 /**
301 * Tries to set status to inactive; fails on contention.
302 */
303 private boolean tryInactivate() {
304 if (active) {
305 if (!pool.tryDecrementActiveCount())
306 return false;
307 active = false;
308 }
309 return true;
310 }
311
312 /**
313 * Computes next value for random victim probe. Scans don't
314 * require a very high quality generator, but also not a crummy
315 * one. Marsaglia xor-shift is cheap and works well.
316 */
317 private static int xorShift(int r) {
318 r ^= (r << 13);
319 r ^= (r >>> 17);
320 return r ^ (r << 5);
321 }
322
323 // Lifecycle methods
324
325 /**
326 * This method is required to be public, but should never be
327 * called explicitly. It performs the main run loop to execute
328 * ForkJoinTasks.
329 */
330 public void run() {
331 Throwable exception = null;
332 try {
333 onStart();
334 pool.sync(this); // await first pool event
335 mainLoop();
336 } catch (Throwable ex) {
337 exception = ex;
338 } finally {
339 onTermination(exception);
340 }
341 }
342
343 /**
344 * Executes tasks until shut down.
345 */
346 private void mainLoop() {
347 while (!isShutdown()) {
348 ForkJoinTask<?> t = pollTask();
349 if (t != null || (t = pollSubmission()) != null)
350 t.quietlyExec();
351 else if (tryInactivate())
352 pool.sync(this);
353 }
354 }
355
356 /**
357 * Initializes internal state after construction but before
358 * processing any tasks. If you override this method, you must
359 * invoke super.onStart() at the beginning of the method.
360 * Initialization requires care: Most fields must have legal
361 * default values, to ensure that attempted accesses from other
362 * threads work correctly even before this thread starts
363 * processing tasks.
364 */
365 protected void onStart() {
366 // Allocate while starting to improve chances of thread-local
367 // isolation
368 queue = new ForkJoinTask<?>[INITIAL_QUEUE_CAPACITY];
369 // Initial value of seed need not be especially random but
370 // should differ across workers and must be nonzero
371 int p = poolIndex + 1;
372 seed = p + (p << 8) + (p << 16) + (p << 24); // spread bits
373 }
374
375 /**
376 * Performs cleanup associated with termination of this worker
377 * thread. If you override this method, you must invoke
378 * {@code super.onTermination} at the end of the overridden method.
379 *
380 * @param exception the exception causing this thread to abort due
381 * to an unrecoverable error, or {@code null} if completed normally
382 */
383 protected void onTermination(Throwable exception) {
384 // Execute remaining local tasks unless aborting or terminating
385 while (exception == null && !pool.isTerminating() && base != sp) {
386 try {
387 ForkJoinTask<?> t = popTask();
388 if (t != null)
389 t.quietlyExec();
390 } catch (Throwable ex) {
391 exception = ex;
392 }
393 }
394 // Cancel other tasks, transition status, notify pool, and
395 // propagate exception to uncaught exception handler
396 try {
397 do {} while (!tryInactivate()); // ensure inactive
398 cancelTasks();
399 runState = TERMINATED;
400 pool.workerTerminated(this);
401 } catch (Throwable ex) { // Shouldn't ever happen
402 if (exception == null) // but if so, at least rethrown
403 exception = ex;
404 } finally {
405 if (exception != null)
406 ForkJoinTask.rethrowException(exception);
407 }
408 }
409
410 // Intrinsics-based support for queue operations.
411
412 /**
413 * Adds in store-order the given task at given slot of q to null.
414 * Caller must ensure q is non-null and index is in range.
415 */
416 private static void setSlot(ForkJoinTask<?>[] q, int i,
417 ForkJoinTask<?> t) {
418 UNSAFE.putOrderedObject(q, (i << qShift) + qBase, t);
419 }
420
421 /**
422 * CAS given slot of q to null. Caller must ensure q is non-null
423 * and index is in range.
424 */
425 private static boolean casSlotNull(ForkJoinTask<?>[] q, int i,
426 ForkJoinTask<?> t) {
427 return UNSAFE.compareAndSwapObject(q, (i << qShift) + qBase, t, null);
428 }
429
430 /**
431 * Sets sp in store-order.
432 */
433 private void storeSp(int s) {
434 UNSAFE.putOrderedInt(this, spOffset, s);
435 }
436
437 // Main queue methods
438
439 /**
440 * Pushes a task. Called only by current thread.
441 *
442 * @param t the task. Caller must ensure non-null.
443 */
444 final void pushTask(ForkJoinTask<?> t) {
445 ForkJoinTask<?>[] q = queue;
446 int mask = q.length - 1;
447 int s = sp;
448 setSlot(q, s & mask, t);
449 storeSp(++s);
450 if ((s -= base) == 1)
451 pool.signalWork();
452 else if (s >= mask)
453 growQueue();
454 }
455
456 /**
457 * Tries to take a task from the base of the queue, failing if
458 * either empty or contended.
459 *
460 * @return a task, or null if none or contended
461 */
462 final ForkJoinTask<?> deqTask() {
463 ForkJoinTask<?> t;
464 ForkJoinTask<?>[] q;
465 int i;
466 int b;
467 if (sp != (b = base) &&
468 (q = queue) != null && // must read q after b
469 (t = q[i = (q.length - 1) & b]) != null &&
470 casSlotNull(q, i, t)) {
471 base = b + 1;
472 return t;
473 }
474 return null;
475 }
476
477 /**
478 * Tries to take a task from the base of own queue, activating if
479 * necessary, failing only if empty. Called only by current thread.
480 *
481 * @return a task, or null if none
482 */
483 final ForkJoinTask<?> locallyDeqTask() {
484 int b;
485 while (sp != (b = base)) {
486 if (tryActivate()) {
487 ForkJoinTask<?>[] q = queue;
488 int i = (q.length - 1) & b;
489 ForkJoinTask<?> t = q[i];
490 if (t != null && casSlotNull(q, i, t)) {
491 base = b + 1;
492 return t;
493 }
494 }
495 }
496 return null;
497 }
498
499 /**
500 * Returns a popped task, or null if empty. Ensures active status
501 * if non-null. Called only by current thread.
502 */
503 final ForkJoinTask<?> popTask() {
504 int s = sp;
505 while (s != base) {
506 if (tryActivate()) {
507 ForkJoinTask<?>[] q = queue;
508 int mask = q.length - 1;
509 int i = (s - 1) & mask;
510 ForkJoinTask<?> t = q[i];
511 if (t == null || !casSlotNull(q, i, t))
512 break;
513 storeSp(s - 1);
514 return t;
515 }
516 }
517 return null;
518 }
519
520 /**
521 * Specialized version of popTask to pop only if
522 * topmost element is the given task. Called only
523 * by current thread while active.
524 *
525 * @param t the task. Caller must ensure non-null.
526 */
527 final boolean unpushTask(ForkJoinTask<?> t) {
528 ForkJoinTask<?>[] q = queue;
529 int mask = q.length - 1;
530 int s = sp - 1;
531 if (casSlotNull(q, s & mask, t)) {
532 storeSp(s);
533 return true;
534 }
535 return false;
536 }
537
538 /**
539 * Returns next task or null if empty or contended
540 */
541 final ForkJoinTask<?> peekTask() {
542 ForkJoinTask<?>[] q = queue;
543 if (q == null)
544 return null;
545 int mask = q.length - 1;
546 int i = locallyFifo ? base : (sp - 1);
547 return q[i & mask];
548 }
549
550 /**
551 * Doubles queue array size. Transfers elements by emulating
552 * steals (deqs) from old array and placing, oldest first, into
553 * new array.
554 */
555 private void growQueue() {
556 ForkJoinTask<?>[] oldQ = queue;
557 int oldSize = oldQ.length;
558 int newSize = oldSize << 1;
559 if (newSize > MAXIMUM_QUEUE_CAPACITY)
560 throw new RejectedExecutionException("Queue capacity exceeded");
561 ForkJoinTask<?>[] newQ = queue = new ForkJoinTask<?>[newSize];
562
563 int b = base;
564 int bf = b + oldSize;
565 int oldMask = oldSize - 1;
566 int newMask = newSize - 1;
567 do {
568 int oldIndex = b & oldMask;
569 ForkJoinTask<?> t = oldQ[oldIndex];
570 if (t != null && !casSlotNull(oldQ, oldIndex, t))
571 t = null;
572 setSlot(newQ, b & newMask, t);
573 } while (++b != bf);
574 pool.signalWork();
575 }
576
577 /**
578 * Tries to steal a task from another worker. Starts at a random
579 * index of workers array, and probes workers until finding one
580 * with non-empty queue or finding that all are empty. It
581 * randomly selects the first n probes. If these are empty, it
582 * resorts to a full circular traversal, which is necessary to
583 * accurately set active status by caller. Also restarts if pool
584 * events occurred since last scan, which forces refresh of
585 * workers array, in case barrier was associated with resize.
586 *
587 * This method must be both fast and quiet -- usually avoiding
588 * memory accesses that could disrupt cache sharing etc other than
589 * those needed to check for and take tasks. This accounts for,
590 * among other things, updating random seed in place without
591 * storing it until exit.
592 *
593 * @return a task, or null if none found
594 */
595 private ForkJoinTask<?> scan() {
596 ForkJoinTask<?> t = null;
597 int r = seed; // extract once to keep scan quiet
598 ForkJoinWorkerThread[] ws; // refreshed on outer loop
599 int mask; // must be power 2 minus 1 and > 0
600 outer:do {
601 if ((ws = pool.workers) != null && (mask = ws.length - 1) > 0) {
602 int idx = r;
603 int probes = ~mask; // use random index while negative
604 for (;;) {
605 r = xorShift(r); // update random seed
606 ForkJoinWorkerThread v = ws[mask & idx];
607 if (v == null || v.sp == v.base) {
608 if (probes <= mask)
609 idx = (probes++ < 0) ? r : (idx + 1);
610 else
611 break;
612 }
613 else if (!tryActivate() || (t = v.deqTask()) == null)
614 continue outer; // restart on contention
615 else
616 break outer;
617 }
618 }
619 } while (pool.hasNewSyncEvent(this)); // retry on pool events
620 seed = r;
621 return t;
622 }
623
624 /**
625 * Gets and removes a local or stolen task.
626 *
627 * @return a task, if available
628 */
629 final ForkJoinTask<?> pollTask() {
630 ForkJoinTask<?> t = locallyFifo ? locallyDeqTask() : popTask();
631 if (t == null && (t = scan()) != null)
632 ++stealCount;
633 return t;
634 }
635
636 /**
637 * Gets a local task.
638 *
639 * @return a task, if available
640 */
641 final ForkJoinTask<?> pollLocalTask() {
642 return locallyFifo ? locallyDeqTask() : popTask();
643 }
644
645 /**
646 * Returns a pool submission, if one exists, activating first.
647 *
648 * @return a submission, if available
649 */
650 private ForkJoinTask<?> pollSubmission() {
651 ForkJoinPool p = pool;
652 while (p.hasQueuedSubmissions()) {
653 ForkJoinTask<?> t;
654 if (tryActivate() && (t = p.pollSubmission()) != null)
655 return t;
656 }
657 return null;
658 }
659
660 // Methods accessed only by Pool
661
662 /**
663 * Removes and cancels all tasks in queue. Can be called from any
664 * thread.
665 */
666 final void cancelTasks() {
667 ForkJoinTask<?> t;
668 while (base != sp && (t = deqTask()) != null)
669 t.cancelIgnoringExceptions();
670 }
671
672 /**
673 * Drains tasks to given collection c.
674 *
675 * @return the number of tasks drained
676 */
677 final int drainTasksTo(Collection<? super ForkJoinTask<?>> c) {
678 int n = 0;
679 ForkJoinTask<?> t;
680 while (base != sp && (t = deqTask()) != null) {
681 c.add(t);
682 ++n;
683 }
684 return n;
685 }
686
687 /**
688 * Gets and clears steal count for accumulation by pool. Called
689 * only when known to be idle (in pool.sync and termination).
690 */
691 final int getAndClearStealCount() {
692 int sc = stealCount;
693 stealCount = 0;
694 return sc;
695 }
696
697 /**
698 * Returns {@code true} if at least one worker in the given array
699 * appears to have at least one queued task.
700 *
701 * @param ws array of workers
702 */
703 static boolean hasQueuedTasks(ForkJoinWorkerThread[] ws) {
704 if (ws != null) {
705 int len = ws.length;
706 for (int j = 0; j < 2; ++j) { // need two passes for clean sweep
707 for (int i = 0; i < len; ++i) {
708 ForkJoinWorkerThread w = ws[i];
709 if (w != null && w.sp != w.base)
710 return true;
711 }
712 }
713 }
714 return false;
715 }
716
717 // Support methods for ForkJoinTask
718
719 /**
720 * Returns an estimate of the number of tasks in the queue.
721 */
722 final int getQueueSize() {
723 // suppress momentarily negative values
724 return Math.max(0, sp - base);
725 }
726
727 /**
728 * Returns an estimate of the number of tasks, offset by a
729 * function of number of idle workers.
730 */
731 final int getEstimatedSurplusTaskCount() {
732 // The halving approximates weighting idle vs non-idle workers
733 return (sp - base) - (pool.getIdleThreadCount() >>> 1);
734 }
735
736 /**
737 * Scans, returning early if joinMe done.
738 */
739 final ForkJoinTask<?> scanWhileJoining(ForkJoinTask<?> joinMe) {
740 ForkJoinTask<?> t = pollTask();
741 if (t != null && joinMe.status < 0 && sp == base) {
742 pushTask(t); // unsteal if done and this task would be stealable
743 t = null;
744 }
745 return t;
746 }
747
748 /**
749 * Runs tasks until {@code pool.isQuiescent()}.
750 */
751 final void helpQuiescePool() {
752 for (;;) {
753 ForkJoinTask<?> t = pollTask();
754 if (t != null)
755 t.quietlyExec();
756 else if (tryInactivate() && pool.isQuiescent())
757 break;
758 }
759 do {} while (!tryActivate()); // re-activate on exit
760 }
761
762 // Unsafe mechanics
763
764 private static final sun.misc.Unsafe UNSAFE = getUnsafe();
765 private static final long spOffset =
766 objectFieldOffset("sp", ForkJoinWorkerThread.class);
767 private static final long runStateOffset =
768 objectFieldOffset("runState", ForkJoinWorkerThread.class);
769 private static final long qBase;
770 private static final int qShift;
771
772 static {
773 qBase = UNSAFE.arrayBaseOffset(ForkJoinTask[].class);
774 int s = UNSAFE.arrayIndexScale(ForkJoinTask[].class);
775 if ((s & (s-1)) != 0)
776 throw new Error("data type scale not a power of two");
777 qShift = 31 - Integer.numberOfLeadingZeros(s);
778 }
779
780 private static long objectFieldOffset(String field, Class<?> klazz) {
781 try {
782 return UNSAFE.objectFieldOffset(klazz.getDeclaredField(field));
783 } catch (NoSuchFieldException e) {
784 // Convert Exception to corresponding Error
785 NoSuchFieldError error = new NoSuchFieldError(field);
786 error.initCause(e);
787 throw error;
788 }
789 }
790
791 /**
792 * Returns a sun.misc.Unsafe. Suitable for use in a 3rd party package.
793 * Replace with a simple call to Unsafe.getUnsafe when integrating
794 * into a jdk.
795 *
796 * @return a sun.misc.Unsafe
797 */
798 private static sun.misc.Unsafe getUnsafe() {
799 try {
800 return sun.misc.Unsafe.getUnsafe();
801 } catch (SecurityException se) {
802 try {
803 return java.security.AccessController.doPrivileged
804 (new java.security
805 .PrivilegedExceptionAction<sun.misc.Unsafe>() {
806 public sun.misc.Unsafe run() throws Exception {
807 java.lang.reflect.Field f = sun.misc
808 .Unsafe.class.getDeclaredField("theUnsafe");
809 f.setAccessible(true);
810 return (sun.misc.Unsafe) f.get(null);
811 }});
812 } catch (java.security.PrivilegedActionException e) {
813 throw new RuntimeException("Could not initialize intrinsics",
814 e.getCause());
815 }
816 }
817 }
818 }