ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/jsr166/jsr166/src/jsr166y/ForkJoinWorkerThread.java
Revision: 1.25
Committed: Sat Aug 1 21:17:11 2009 UTC (14 years, 9 months ago) by jsr166
Branch: MAIN
Changes since 1.24: +5 -5 lines
Log Message:
@link-ify class summaries

File Contents

# User Rev Content
1 dl 1.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 jsr166 1.18
9 dl 1.1 import java.util.concurrent.*;
10 jsr166 1.18
11     import java.util.Collection;
12 dl 1.1
13     /**
14 dl 1.2 * A thread managed by a {@link ForkJoinPool}. This class is
15     * subclassable solely for the sake of adding functionality -- there
16 jsr166 1.25 * 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     * ForkJoinWorkerThreadFactory} to use it in a {@code ForkJoinPool}.
21 jsr166 1.6 *
22 jsr166 1.13 * @since 1.7
23     * @author Doug Lea
24 dl 1.1 */
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 jsr166 1.10 * both a successful pop and deq mainly entail CAS'ing a non-null
48 dl 1.1 * 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 jsr166 1.9 * aggregate, we ensure at least probabilistic non-blockingness. If
60 dl 1.1 * 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 dl 1.23 * 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 dl 1.1 * 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 jsr166 1.9 * these efficiently is to use direct Unsafe calls. (Using external
85 dl 1.1 * 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 dl 1.5 * 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 dl 1.1 */
143 dl 1.5 private static final int MAXIMUM_QUEUE_CAPACITY = 1 << 28;
144 dl 1.1
145     /**
146 jsr166 1.16 * The pool this thread works in. Accessed directly by ForkJoinTask.
147 dl 1.1 */
148 dl 1.5 final ForkJoinPool pool;
149 dl 1.1
150     /**
151     * The work-stealing queue array. Size must be a power of two.
152 dl 1.5 * Initialized when thread starts, to improve memory locality.
153 dl 1.1 */
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 dl 1.5 * 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 jsr166 1.16 * calling pool.sync.
175 dl 1.1 */
176 dl 1.5 private boolean active;
177 dl 1.1
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 dl 1.5 * Seed for random number generator for choosing steal victims.
186     * Uses Marsaglia xorshift. Must be nonzero upon initialization.
187 dl 1.1 */
188 dl 1.5 private int seed;
189 dl 1.1
190     /**
191     * Number of steals, transferred to pool when idle
192     */
193     private int stealCount;
194    
195     /**
196 dl 1.5 * Index of this worker in pool array. Set once by pool before
197 jsr166 1.16 * running, and accessed directly by pool during cleanup etc.
198 dl 1.1 */
199 dl 1.5 int poolIndex;
200 dl 1.1
201     /**
202 dl 1.5 * The last barrier event waited for. Accessed in pool callback
203     * methods, but only by current thread.
204 dl 1.1 */
205 dl 1.5 long lastEventCount;
206 dl 1.1
207     /**
208 dl 1.7 * True if use local fifo, not default lifo, for local polling
209     */
210     private boolean locallyFifo;
211    
212     /**
213 dl 1.1 * Creates a ForkJoinWorkerThread operating in the given pool.
214 jsr166 1.11 *
215 dl 1.1 * @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 dl 1.5 // Note: poolIndex is set by pool during construction
222     // Remaining initialization is deferred to onStart
223 dl 1.1 }
224    
225 jsr166 1.6 // Public access methods
226 dl 1.2
227     /**
228 jsr166 1.11 * Returns the pool hosting this thread.
229     *
230 dl 1.2 * @return the pool
231     */
232 dl 1.4 public ForkJoinPool getPool() {
233     return pool;
234 dl 1.2 }
235    
236     /**
237 dl 1.4 * 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 dl 1.5 * collect results per-worker rather than per-task.
242 jsr166 1.11 *
243     * @return the index number
244 dl 1.2 */
245 dl 1.4 public int getPoolIndex() {
246     return poolIndex;
247 dl 1.2 }
248    
249 dl 1.7 /**
250     * Establishes local first-in-first-out scheduling mode for forked
251 jsr166 1.8 * tasks that are never joined.
252 jsr166 1.11 *
253 dl 1.7 * @param async if true, use locally FIFO scheduling
254     */
255     void setAsyncMode(boolean async) {
256     locallyFifo = async;
257     }
258 dl 1.5
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 jsr166 1.21 * Transitions to at least the given state.
275     *
276     * @return {@code true} if not already at least at given state
277 dl 1.5 */
278     private boolean transitionRunStateTo(int state) {
279     for (;;) {
280     int s = runState;
281     if (s >= state)
282     return false;
283 jsr166 1.12 if (UNSAFE.compareAndSwapInt(this, runStateOffset, s, state))
284 dl 1.5 return true;
285     }
286     }
287    
288     /**
289 jsr166 1.11 * Tries to set status to active; fails on contention.
290 dl 1.5 */
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 jsr166 1.16 * Tries to set status to inactive; fails on contention.
302 dl 1.5 */
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 jsr166 1.11 * Computes next value for random victim probe. Scans don't
314 dl 1.5 * require a very high quality generator, but also not a crummy
315 jsr166 1.11 * one. Marsaglia xor-shift is cheap and works well.
316 dl 1.5 */
317     private static int xorShift(int r) {
318 dl 1.23 r ^= (r << 13);
319     r ^= (r >>> 17);
320 jsr166 1.24 return r ^ (r << 5);
321 dl 1.5 }
322    
323     // Lifecycle methods
324 dl 1.1
325     /**
326 dl 1.5 * 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 dl 1.1 */
330 dl 1.5 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 dl 1.1 }
342    
343     /**
344 jsr166 1.11 * Executes tasks until shut down.
345 dl 1.1 */
346 dl 1.5 private void mainLoop() {
347     while (!isShutdown()) {
348     ForkJoinTask<?> t = pollTask();
349 jsr166 1.6 if (t != null || (t = pollSubmission()) != null)
350 dl 1.5 t.quietlyExec();
351     else if (tryInactivate())
352     pool.sync(this);
353     }
354 dl 1.1 }
355    
356 dl 1.5 /**
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 dl 1.1
375     /**
376 jsr166 1.11 * Performs cleanup associated with termination of this worker
377 dl 1.5 * thread. If you override this method, you must invoke
378 jsr166 1.16 * {@code super.onTermination} at the end of the overridden method.
379 dl 1.5 *
380     * @param exception the exception causing this thread to abort due
381 jsr166 1.21 * to an unrecoverable error, or {@code null} if completed normally
382 dl 1.1 */
383 dl 1.5 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 jsr166 1.16 } catch (Throwable ex) {
391 dl 1.5 exception = ex;
392     }
393     }
394     // Cancel other tasks, transition status, notify pool, and
395     // propagate exception to uncaught exception handler
396     try {
397 jsr166 1.16 do {} while (!tryInactivate()); // ensure inactive
398 jsr166 1.6 cancelTasks();
399 dl 1.5 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 dl 1.1 }
409    
410 jsr166 1.6 // Intrinsics-based support for queue operations.
411 dl 1.5
412 dl 1.1 /**
413 jsr166 1.10 * 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 dl 1.1 */
416     private static void setSlot(ForkJoinTask<?>[] q, int i,
417 jsr166 1.14 ForkJoinTask<?> t) {
418 jsr166 1.12 UNSAFE.putOrderedObject(q, (i << qShift) + qBase, t);
419 dl 1.1 }
420    
421     /**
422 jsr166 1.10 * CAS given slot of q to null. Caller must ensure q is non-null
423 dl 1.1 * and index is in range.
424     */
425     private static boolean casSlotNull(ForkJoinTask<?>[] q, int i,
426     ForkJoinTask<?> t) {
427 jsr166 1.12 return UNSAFE.compareAndSwapObject(q, (i << qShift) + qBase, t, null);
428 dl 1.1 }
429    
430 dl 1.5 /**
431     * Sets sp in store-order.
432     */
433     private void storeSp(int s) {
434 jsr166 1.12 UNSAFE.putOrderedInt(this, spOffset, s);
435 dl 1.5 }
436    
437 dl 1.1 // Main queue methods
438    
439     /**
440     * Pushes a task. Called only by current thread.
441 jsr166 1.11 *
442 jsr166 1.10 * @param t the task. Caller must ensure non-null.
443 dl 1.1 */
444     final void pushTask(ForkJoinTask<?> t) {
445     ForkJoinTask<?>[] q = queue;
446     int mask = q.length - 1;
447     int s = sp;
448 dl 1.5 setSlot(q, s & mask, t);
449     storeSp(++s);
450 dl 1.1 if ((s -= base) == 1)
451 dl 1.5 pool.signalWork();
452 dl 1.1 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 jsr166 1.11 *
460     * @return a task, or null if none or contended
461 dl 1.1 */
462 dl 1.7 final ForkJoinTask<?> deqTask() {
463 dl 1.5 ForkJoinTask<?> t;
464 dl 1.1 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 dl 1.5 casSlotNull(q, i, t)) {
471 dl 1.1 base = b + 1;
472     return t;
473     }
474     return null;
475     }
476    
477     /**
478 dl 1.23 * 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 dl 1.5 * Returns a popped task, or null if empty. Ensures active status
501 jsr166 1.10 * if non-null. Called only by current thread.
502 dl 1.1 */
503     final ForkJoinTask<?> popTask() {
504     int s = sp;
505 dl 1.5 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 dl 1.1 }
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 dl 1.5 * by current thread while active.
524 jsr166 1.11 *
525     * @param t the task. Caller must ensure non-null.
526 dl 1.1 */
527     final boolean unpushTask(ForkJoinTask<?> t) {
528     ForkJoinTask<?>[] q = queue;
529     int mask = q.length - 1;
530     int s = sp - 1;
531 dl 1.5 if (casSlotNull(q, s & mask, t)) {
532     storeSp(s);
533 dl 1.1 return true;
534     }
535     return false;
536     }
537    
538     /**
539 dl 1.23 * Returns next task or null if empty or contended
540 dl 1.1 */
541 dl 1.2 final ForkJoinTask<?> peekTask() {
542 dl 1.1 ForkJoinTask<?>[] q = queue;
543 dl 1.7 if (q == null)
544     return null;
545     int mask = q.length - 1;
546 jsr166 1.15 int i = locallyFifo ? base : (sp - 1);
547 dl 1.7 return q[i & mask];
548 dl 1.1 }
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 dl 1.5 pool.signalWork();
575 dl 1.1 }
576    
577     /**
578 dl 1.5 * 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 dl 1.1 *
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 dl 1.5 * storing it until exit.
592 dl 1.1 *
593     * @return a task, or null if none found
594     */
595 dl 1.5 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 dl 1.1 int idx = r;
603 dl 1.5 int probes = ~mask; // use random index while negative
604 dl 1.1 for (;;) {
605 dl 1.5 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 jsr166 1.15 idx = (probes++ < 0) ? r : (idx + 1);
610 dl 1.5 else
611     break;
612 dl 1.1 }
613 dl 1.5 else if (!tryActivate() || (t = v.deqTask()) == null)
614     continue outer; // restart on contention
615 dl 1.1 else
616 dl 1.5 break outer;
617 dl 1.1 }
618     }
619 dl 1.5 } while (pool.hasNewSyncEvent(this)); // retry on pool events
620     seed = r;
621     return t;
622 dl 1.1 }
623    
624     /**
625 jsr166 1.11 * Gets and removes a local or stolen task.
626     *
627 dl 1.5 * @return a task, if available
628     */
629     final ForkJoinTask<?> pollTask() {
630 dl 1.23 ForkJoinTask<?> t = locallyFifo ? locallyDeqTask() : popTask();
631 dl 1.5 if (t == null && (t = scan()) != null)
632     ++stealCount;
633     return t;
634 dl 1.1 }
635    
636     /**
637 jsr166 1.11 * Gets a local task.
638     *
639 dl 1.7 * @return a task, if available
640     */
641     final ForkJoinTask<?> pollLocalTask() {
642 dl 1.23 return locallyFifo ? locallyDeqTask() : popTask();
643 dl 1.7 }
644    
645     /**
646 dl 1.5 * Returns a pool submission, if one exists, activating first.
647 jsr166 1.11 *
648 dl 1.5 * @return a submission, if available
649 dl 1.1 */
650 dl 1.5 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 dl 1.1 }
657 dl 1.5 return null;
658 dl 1.1 }
659 dl 1.5
660     // Methods accessed only by Pool
661    
662 dl 1.1 /**
663 dl 1.5 * Removes and cancels all tasks in queue. Can be called from any
664     * thread.
665 dl 1.1 */
666 dl 1.5 final void cancelTasks() {
667 dl 1.3 ForkJoinTask<?> t;
668 dl 1.5 while (base != sp && (t = deqTask()) != null)
669     t.cancelIgnoringExceptions();
670 dl 1.1 }
671    
672     /**
673 jsr166 1.11 * Drains tasks to given collection c.
674     *
675 dl 1.7 * @return the number of tasks drained
676     */
677 dl 1.22 final int drainTasksTo(Collection<? super ForkJoinTask<?>> c) {
678 dl 1.7 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 jsr166 1.11 * Gets and clears steal count for accumulation by pool. Called
689 dl 1.5 * only when known to be idle (in pool.sync and termination).
690 dl 1.1 */
691 dl 1.5 final int getAndClearStealCount() {
692     int sc = stealCount;
693     stealCount = 0;
694     return sc;
695     }
696    
697     /**
698 jsr166 1.21 * Returns {@code true} if at least one worker in the given array
699     * appears to have at least one queued task.
700 jsr166 1.16 *
701 dl 1.5 * @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 dl 1.1 }
712     }
713     }
714 dl 1.5 return false;
715 dl 1.1 }
716    
717 dl 1.5 // Support methods for ForkJoinTask
718    
719 dl 1.1 /**
720 dl 1.2 * Returns an estimate of the number of tasks in the queue.
721     */
722     final int getQueueSize() {
723 jsr166 1.15 // suppress momentarily negative values
724     return Math.max(0, sp - base);
725 dl 1.2 }
726    
727     /**
728 dl 1.1 * Returns an estimate of the number of tasks, offset by a
729     * function of number of idle workers.
730     */
731     final int getEstimatedSurplusTaskCount() {
732 dl 1.3 // The halving approximates weighting idle vs non-idle workers
733 dl 1.1 return (sp - base) - (pool.getIdleThreadCount() >>> 1);
734     }
735    
736     /**
737 jsr166 1.16 * Scans, returning early if joinMe done.
738 dl 1.1 */
739 dl 1.5 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 dl 1.1 }
747 jsr166 1.6
748 dl 1.1 /**
749 jsr166 1.16 * Runs tasks until {@code pool.isQuiescent()}.
750 dl 1.1 */
751 dl 1.5 final void helpQuiescePool() {
752     for (;;) {
753     ForkJoinTask<?> t = pollTask();
754 jsr166 1.6 if (t != null)
755 dl 1.5 t.quietlyExec();
756     else if (tryInactivate() && pool.isQuiescent())
757     break;
758     }
759 jsr166 1.16 do {} while (!tryActivate()); // re-activate on exit
760 dl 1.1 }
761    
762 jsr166 1.20 // 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 jsr166 1.17 private static sun.misc.Unsafe getUnsafe() {
799 jsr166 1.6 try {
800 jsr166 1.17 return sun.misc.Unsafe.getUnsafe();
801 jsr166 1.6 } catch (SecurityException se) {
802     try {
803     return java.security.AccessController.doPrivileged
804 jsr166 1.20 (new java.security
805     .PrivilegedExceptionAction<sun.misc.Unsafe>() {
806 jsr166 1.17 public sun.misc.Unsafe run() throws Exception {
807 jsr166 1.20 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 jsr166 1.6 }});
812     } catch (java.security.PrivilegedActionException e) {
813 jsr166 1.17 throw new RuntimeException("Could not initialize intrinsics",
814     e.getCause());
815 jsr166 1.6 }
816     }
817     }
818 dl 1.1 }