ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/jsr166/jsr166/src/jsr166y/ForkJoinWorkerThread.java
(Generate patch)

Comparing jsr166/src/jsr166y/ForkJoinWorkerThread.java (file contents):
Revision 1.21 by jsr166, Mon Jul 27 20:57:44 2009 UTC vs.
Revision 1.27 by dl, Mon Aug 3 13:01:15 2009 UTC

# Line 13 | Line 13 | import java.util.Collection;
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.
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
# Line 65 | Line 66 | public class ForkJoinWorkerThread extend
66       * which gives threads a chance to activate if necessary before
67       * 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
# Line 309 | Line 316 | public class ForkJoinWorkerThread extend
316       * one.  Marsaglia xor-shift is cheap and works well.
317       */
318      private static int xorShift(int r) {
319 <        r ^= r << 1;
320 <        r ^= r >>> 3;
321 <        r ^= r << 10;
315 <        return r;
319 >        r ^= (r << 13);
320 >        r ^= (r >>> 17);
321 >        return r ^ (r << 5);
322      }
323  
324      // Lifecycle methods
# Line 377 | Line 383 | public class ForkJoinWorkerThread extend
383       */
384      protected void onTermination(Throwable exception) {
385          // Execute remaining local tasks unless aborting or terminating
386 <        while (exception == null &&  !pool.isTerminating() && base != sp) {
386 >        while (exception == null && pool.isProcessingTasks() && base != sp) {
387              try {
388                  ForkJoinTask<?> t = popTask();
389                  if (t != null)
# Line 470 | Line 476 | public class ForkJoinWorkerThread extend
476      }
477  
478      /**
479 +     * Tries to take a task from the base of own queue, activating if
480 +     * necessary, failing only if empty. Called only by current thread.
481 +     *
482 +     * @return a task, or null if none
483 +     */
484 +    final ForkJoinTask<?> locallyDeqTask() {
485 +        int b;
486 +        while (sp != (b = base)) {
487 +            if (tryActivate()) {
488 +                ForkJoinTask<?>[] q = queue;
489 +                int i = (q.length - 1) & b;
490 +                ForkJoinTask<?> t = q[i];
491 +                if (t != null && casSlotNull(q, i, t)) {
492 +                    base = b + 1;
493 +                    return t;
494 +                }
495 +            }
496 +        }
497 +        return null;
498 +    }
499 +
500 +    /**
501       * Returns a popped task, or null if empty. Ensures active status
502       * if non-null. Called only by current thread.
503       */
# Line 509 | Line 537 | public class ForkJoinWorkerThread extend
537      }
538  
539      /**
540 <     * Returns next task.
540 >     * Returns next task or null if empty or contended
541       */
542      final ForkJoinTask<?> peekTask() {
543          ForkJoinTask<?>[] q = queue;
# Line 600 | Line 628 | public class ForkJoinWorkerThread extend
628       * @return a task, if available
629       */
630      final ForkJoinTask<?> pollTask() {
631 <        ForkJoinTask<?> t = locallyFifo ? deqTask() : popTask();
631 >        ForkJoinTask<?> t = locallyFifo ? locallyDeqTask() : popTask();
632          if (t == null && (t = scan()) != null)
633              ++stealCount;
634          return t;
# Line 612 | Line 640 | public class ForkJoinWorkerThread extend
640       * @return a task, if available
641       */
642      final ForkJoinTask<?> pollLocalTask() {
643 <        return locallyFifo ? deqTask() : popTask();
643 >        return locallyFifo ? locallyDeqTask() : popTask();
644      }
645  
646      /**
# Line 647 | Line 675 | public class ForkJoinWorkerThread extend
675       *
676       * @return the number of tasks drained
677       */
678 <    final int drainTasksTo(Collection<ForkJoinTask<?>> c) {
678 >    final int drainTasksTo(Collection<? super ForkJoinTask<?>> c) {
679          int n = 0;
680          ForkJoinTask<?> t;
681          while (base != sp && (t = deqTask()) != null) {

Diff Legend

– Removed lines
+ Added lines
< Changed lines
> Changed lines