ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/jsr166/jsr166/src/jsr166y/ForkJoinWorkerThread.java
Revision: 1.30
Committed: Tue Oct 6 19:02:48 2009 UTC (14 years, 7 months ago) by jsr166
Branch: MAIN
Changes since 1.29: +6 -2 lines
Log Message:
6888149: AtomicReferenceArray causes SIGSEGV -> SEGV_MAPERR error

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