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

# 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 dl 1.26 * ForkJoinPool.ForkJoinWorkerThreadFactory} to use it in a {@code
21     * ForkJoinPool}.
22 jsr166 1.6 *
23 jsr166 1.13 * @since 1.7
24     * @author Doug Lea
25 dl 1.1 */
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 jsr166 1.10 * both a successful pop and deq mainly entail CAS'ing a non-null
49 dl 1.1 * 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 dl 1.28 * 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 dl 1.1 *
69 dl 1.23 * 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 dl 1.1 * 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 dl 1.28 * (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 dl 1.1 * AtomicIntegers and AtomicReferenceArrays for the indices and
91     * array is significantly slower because of memory locality and
92 dl 1.28 * indirection effects.)
93 jsr166 1.29 *
94 dl 1.28 * 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 dl 1.1 * 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 dl 1.28 * 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 dl 1.1 *
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 dl 1.5 * 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 dl 1.1 */
149 dl 1.5 private static final int MAXIMUM_QUEUE_CAPACITY = 1 << 28;
150 dl 1.1
151     /**
152 jsr166 1.16 * The pool this thread works in. Accessed directly by ForkJoinTask.
153 dl 1.1 */
154 dl 1.5 final ForkJoinPool pool;
155 dl 1.1
156     /**
157     * The work-stealing queue array. Size must be a power of two.
158 dl 1.5 * Initialized when thread starts, to improve memory locality.
159 dl 1.1 */
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 dl 1.5 * 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 jsr166 1.16 * calling pool.sync.
181 dl 1.1 */
182 dl 1.5 private boolean active;
183 dl 1.1
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 dl 1.5 * Seed for random number generator for choosing steal victims.
192     * Uses Marsaglia xorshift. Must be nonzero upon initialization.
193 dl 1.1 */
194 dl 1.5 private int seed;
195 dl 1.1
196     /**
197     * Number of steals, transferred to pool when idle
198     */
199     private int stealCount;
200    
201     /**
202 dl 1.5 * Index of this worker in pool array. Set once by pool before
203 jsr166 1.16 * running, and accessed directly by pool during cleanup etc.
204 dl 1.1 */
205 dl 1.5 int poolIndex;
206 dl 1.1
207     /**
208 dl 1.5 * The last barrier event waited for. Accessed in pool callback
209     * methods, but only by current thread.
210 dl 1.1 */
211 dl 1.5 long lastEventCount;
212 dl 1.1
213     /**
214 dl 1.7 * True if use local fifo, not default lifo, for local polling
215     */
216     private boolean locallyFifo;
217    
218     /**
219 dl 1.1 * Creates a ForkJoinWorkerThread operating in the given pool.
220 jsr166 1.11 *
221 dl 1.1 * @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 dl 1.5 // Note: poolIndex is set by pool during construction
228     // Remaining initialization is deferred to onStart
229 dl 1.1 }
230    
231 jsr166 1.6 // Public access methods
232 dl 1.2
233     /**
234 jsr166 1.11 * Returns the pool hosting this thread.
235     *
236 dl 1.2 * @return the pool
237     */
238 dl 1.4 public ForkJoinPool getPool() {
239     return pool;
240 dl 1.2 }
241    
242     /**
243 dl 1.4 * 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 dl 1.5 * collect results per-worker rather than per-task.
248 jsr166 1.11 *
249     * @return the index number
250 dl 1.2 */
251 dl 1.4 public int getPoolIndex() {
252     return poolIndex;
253 dl 1.2 }
254    
255 dl 1.7 /**
256     * Establishes local first-in-first-out scheduling mode for forked
257 jsr166 1.8 * tasks that are never joined.
258 jsr166 1.11 *
259 dl 1.7 * @param async if true, use locally FIFO scheduling
260     */
261     void setAsyncMode(boolean async) {
262     locallyFifo = async;
263     }
264 dl 1.5
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 jsr166 1.21 * Transitions to at least the given state.
281     *
282     * @return {@code true} if not already at least at given state
283 dl 1.5 */
284     private boolean transitionRunStateTo(int state) {
285     for (;;) {
286     int s = runState;
287     if (s >= state)
288     return false;
289 jsr166 1.12 if (UNSAFE.compareAndSwapInt(this, runStateOffset, s, state))
290 dl 1.5 return true;
291     }
292     }
293    
294     /**
295 jsr166 1.11 * Tries to set status to active; fails on contention.
296 dl 1.5 */
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 jsr166 1.16 * Tries to set status to inactive; fails on contention.
308 dl 1.5 */
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 jsr166 1.11 * Computes next value for random victim probe. Scans don't
320 dl 1.5 * require a very high quality generator, but also not a crummy
321 jsr166 1.11 * one. Marsaglia xor-shift is cheap and works well.
322 dl 1.5 */
323     private static int xorShift(int r) {
324 dl 1.23 r ^= (r << 13);
325     r ^= (r >>> 17);
326 jsr166 1.24 return r ^ (r << 5);
327 dl 1.5 }
328    
329     // Lifecycle methods
330 dl 1.1
331     /**
332 dl 1.5 * 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 dl 1.1 */
336 dl 1.5 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 dl 1.1 }
348    
349     /**
350 jsr166 1.11 * Executes tasks until shut down.
351 dl 1.1 */
352 dl 1.5 private void mainLoop() {
353     while (!isShutdown()) {
354     ForkJoinTask<?> t = pollTask();
355 jsr166 1.6 if (t != null || (t = pollSubmission()) != null)
356 dl 1.5 t.quietlyExec();
357     else if (tryInactivate())
358     pool.sync(this);
359     }
360 dl 1.1 }
361    
362 dl 1.5 /**
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 dl 1.1
381     /**
382 jsr166 1.11 * Performs cleanup associated with termination of this worker
383 dl 1.5 * thread. If you override this method, you must invoke
384 jsr166 1.16 * {@code super.onTermination} at the end of the overridden method.
385 dl 1.5 *
386     * @param exception the exception causing this thread to abort due
387 jsr166 1.21 * to an unrecoverable error, or {@code null} if completed normally
388 dl 1.1 */
389 dl 1.5 protected void onTermination(Throwable exception) {
390     // Execute remaining local tasks unless aborting or terminating
391 dl 1.27 while (exception == null && pool.isProcessingTasks() && base != sp) {
392 dl 1.5 try {
393     ForkJoinTask<?> t = popTask();
394     if (t != null)
395     t.quietlyExec();
396 jsr166 1.16 } catch (Throwable ex) {
397 dl 1.5 exception = ex;
398     }
399     }
400     // Cancel other tasks, transition status, notify pool, and
401     // propagate exception to uncaught exception handler
402     try {
403 jsr166 1.16 do {} while (!tryInactivate()); // ensure inactive
404 jsr166 1.6 cancelTasks();
405 dl 1.5 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 dl 1.1 }
415    
416 jsr166 1.6 // Intrinsics-based support for queue operations.
417 dl 1.5
418 jsr166 1.30 private static long slotOffset(int i) {
419     return ((long) i << qShift) + qBase;
420     }
421    
422 dl 1.1 /**
423 jsr166 1.10 * 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 dl 1.1 */
426     private static void setSlot(ForkJoinTask<?>[] q, int i,
427 jsr166 1.14 ForkJoinTask<?> t) {
428 jsr166 1.30 UNSAFE.putOrderedObject(q, slotOffset(i), t);
429 dl 1.1 }
430    
431     /**
432 jsr166 1.10 * CAS given slot of q to null. Caller must ensure q is non-null
433 dl 1.1 * and index is in range.
434     */
435     private static boolean casSlotNull(ForkJoinTask<?>[] q, int i,
436     ForkJoinTask<?> t) {
437 jsr166 1.30 return UNSAFE.compareAndSwapObject(q, slotOffset(i), t, null);
438 dl 1.1 }
439    
440 dl 1.5 /**
441     * Sets sp in store-order.
442     */
443     private void storeSp(int s) {
444 jsr166 1.12 UNSAFE.putOrderedInt(this, spOffset, s);
445 dl 1.5 }
446    
447 dl 1.1 // Main queue methods
448    
449     /**
450     * Pushes a task. Called only by current thread.
451 jsr166 1.11 *
452 jsr166 1.10 * @param t the task. Caller must ensure non-null.
453 dl 1.1 */
454     final void pushTask(ForkJoinTask<?> t) {
455     ForkJoinTask<?>[] q = queue;
456     int mask = q.length - 1;
457     int s = sp;
458 dl 1.5 setSlot(q, s & mask, t);
459     storeSp(++s);
460 dl 1.1 if ((s -= base) == 1)
461 dl 1.5 pool.signalWork();
462 dl 1.1 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 jsr166 1.11 *
470     * @return a task, or null if none or contended
471 dl 1.1 */
472 dl 1.7 final ForkJoinTask<?> deqTask() {
473 dl 1.5 ForkJoinTask<?> t;
474 dl 1.1 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 dl 1.5 casSlotNull(q, i, t)) {
481 dl 1.1 base = b + 1;
482     return t;
483     }
484     return null;
485     }
486    
487     /**
488 dl 1.23 * 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 dl 1.5 * Returns a popped task, or null if empty. Ensures active status
511 jsr166 1.10 * if non-null. Called only by current thread.
512 dl 1.1 */
513     final ForkJoinTask<?> popTask() {
514     int s = sp;
515 dl 1.5 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 dl 1.1 }
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 dl 1.5 * by current thread while active.
534 jsr166 1.11 *
535     * @param t the task. Caller must ensure non-null.
536 dl 1.1 */
537     final boolean unpushTask(ForkJoinTask<?> t) {
538     ForkJoinTask<?>[] q = queue;
539     int mask = q.length - 1;
540     int s = sp - 1;
541 dl 1.5 if (casSlotNull(q, s & mask, t)) {
542     storeSp(s);
543 dl 1.1 return true;
544     }
545     return false;
546     }
547    
548     /**
549 dl 1.23 * Returns next task or null if empty or contended
550 dl 1.1 */
551 dl 1.2 final ForkJoinTask<?> peekTask() {
552 dl 1.1 ForkJoinTask<?>[] q = queue;
553 dl 1.7 if (q == null)
554     return null;
555     int mask = q.length - 1;
556 jsr166 1.15 int i = locallyFifo ? base : (sp - 1);
557 dl 1.7 return q[i & mask];
558 dl 1.1 }
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 dl 1.5 pool.signalWork();
585 dl 1.1 }
586    
587     /**
588 dl 1.5 * 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 dl 1.1 *
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 dl 1.5 * storing it until exit.
602 dl 1.1 *
603     * @return a task, or null if none found
604     */
605 dl 1.5 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 dl 1.1 int idx = r;
613 dl 1.5 int probes = ~mask; // use random index while negative
614 dl 1.1 for (;;) {
615 dl 1.5 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 jsr166 1.15 idx = (probes++ < 0) ? r : (idx + 1);
620 dl 1.5 else
621     break;
622 dl 1.1 }
623 dl 1.5 else if (!tryActivate() || (t = v.deqTask()) == null)
624     continue outer; // restart on contention
625 dl 1.1 else
626 dl 1.5 break outer;
627 dl 1.1 }
628     }
629 dl 1.5 } while (pool.hasNewSyncEvent(this)); // retry on pool events
630     seed = r;
631     return t;
632 dl 1.1 }
633    
634     /**
635 jsr166 1.11 * Gets and removes a local or stolen task.
636     *
637 dl 1.5 * @return a task, if available
638     */
639     final ForkJoinTask<?> pollTask() {
640 dl 1.23 ForkJoinTask<?> t = locallyFifo ? locallyDeqTask() : popTask();
641 dl 1.5 if (t == null && (t = scan()) != null)
642     ++stealCount;
643     return t;
644 dl 1.1 }
645    
646     /**
647 jsr166 1.11 * Gets a local task.
648     *
649 dl 1.7 * @return a task, if available
650     */
651     final ForkJoinTask<?> pollLocalTask() {
652 dl 1.23 return locallyFifo ? locallyDeqTask() : popTask();
653 dl 1.7 }
654    
655     /**
656 dl 1.5 * Returns a pool submission, if one exists, activating first.
657 jsr166 1.11 *
658 dl 1.5 * @return a submission, if available
659 dl 1.1 */
660 dl 1.5 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 dl 1.1 }
667 dl 1.5 return null;
668 dl 1.1 }
669 dl 1.5
670     // Methods accessed only by Pool
671    
672 dl 1.1 /**
673 dl 1.5 * Removes and cancels all tasks in queue. Can be called from any
674     * thread.
675 dl 1.1 */
676 dl 1.5 final void cancelTasks() {
677 dl 1.3 ForkJoinTask<?> t;
678 dl 1.5 while (base != sp && (t = deqTask()) != null)
679     t.cancelIgnoringExceptions();
680 dl 1.1 }
681    
682     /**
683 jsr166 1.11 * Drains tasks to given collection c.
684     *
685 dl 1.7 * @return the number of tasks drained
686     */
687 dl 1.22 final int drainTasksTo(Collection<? super ForkJoinTask<?>> c) {
688 dl 1.7 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 jsr166 1.11 * Gets and clears steal count for accumulation by pool. Called
699 dl 1.5 * only when known to be idle (in pool.sync and termination).
700 dl 1.1 */
701 dl 1.5 final int getAndClearStealCount() {
702     int sc = stealCount;
703     stealCount = 0;
704     return sc;
705     }
706    
707     /**
708 jsr166 1.21 * Returns {@code true} if at least one worker in the given array
709     * appears to have at least one queued task.
710 jsr166 1.16 *
711 dl 1.5 * @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 dl 1.1 }
722     }
723     }
724 dl 1.5 return false;
725 dl 1.1 }
726    
727 dl 1.5 // Support methods for ForkJoinTask
728    
729 dl 1.1 /**
730 dl 1.2 * Returns an estimate of the number of tasks in the queue.
731     */
732     final int getQueueSize() {
733 jsr166 1.15 // suppress momentarily negative values
734     return Math.max(0, sp - base);
735 dl 1.2 }
736    
737     /**
738 dl 1.1 * Returns an estimate of the number of tasks, offset by a
739     * function of number of idle workers.
740     */
741     final int getEstimatedSurplusTaskCount() {
742 dl 1.3 // The halving approximates weighting idle vs non-idle workers
743 dl 1.1 return (sp - base) - (pool.getIdleThreadCount() >>> 1);
744     }
745    
746     /**
747 jsr166 1.16 * Scans, returning early if joinMe done.
748 dl 1.1 */
749 dl 1.5 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 dl 1.1 }
757 jsr166 1.6
758 dl 1.1 /**
759 jsr166 1.16 * Runs tasks until {@code pool.isQuiescent()}.
760 dl 1.1 */
761 dl 1.5 final void helpQuiescePool() {
762     for (;;) {
763     ForkJoinTask<?> t = pollTask();
764 jsr166 1.6 if (t != null)
765 dl 1.5 t.quietlyExec();
766     else if (tryInactivate() && pool.isQuiescent())
767     break;
768     }
769 jsr166 1.16 do {} while (!tryActivate()); // re-activate on exit
770 dl 1.1 }
771    
772 jsr166 1.20 // 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 jsr166 1.17 private static sun.misc.Unsafe getUnsafe() {
809 jsr166 1.6 try {
810 jsr166 1.17 return sun.misc.Unsafe.getUnsafe();
811 jsr166 1.6 } catch (SecurityException se) {
812     try {
813     return java.security.AccessController.doPrivileged
814 jsr166 1.20 (new java.security
815     .PrivilegedExceptionAction<sun.misc.Unsafe>() {
816 jsr166 1.17 public sun.misc.Unsafe run() throws Exception {
817 jsr166 1.20 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 jsr166 1.6 }});
822     } catch (java.security.PrivilegedActionException e) {
823 jsr166 1.17 throw new RuntimeException("Could not initialize intrinsics",
824     e.getCause());
825 jsr166 1.6 }
826     }
827     }
828 dl 1.1 }