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.19 by jsr166, Sun Jul 26 05:55:34 2009 UTC vs.
Revision 1.24 by jsr166, Fri Jul 31 19:52:39 2009 UTC

# Line 65 | Line 65 | public class ForkJoinWorkerThread extend
65       * which gives threads a chance to activate if necessary before
66       * stealing (see below).
67       *
68 +     * This approach also enables support for "async mode" where local
69 +     * task processing is in FIFO, not LIFO order; simply by using a
70 +     * version of deq rather than pop when locallyFifo is true (as set
71 +     * by the ForkJoinPool).  This allows use in message-passing
72 +     * frameworks in which tasks are never joined.
73 +     *
74       * Efficient implementation of this approach currently relies on
75       * an uncomfortable amount of "Unsafe" mechanics. To maintain
76       * correct orderings, reads and writes of variable base require
# Line 265 | Line 271 | public class ForkJoinWorkerThread extend
271      final boolean shutdownNow()   { return transitionRunStateTo(TERMINATING); }
272  
273      /**
274 <     * Transitions to at least the given state.  Returns true if not
275 <     * already at least at given state.
274 >     * Transitions to at least the given state.
275 >     *
276 >     * @return {@code true} if not already at least at given state
277       */
278      private boolean transitionRunStateTo(int state) {
279          for (;;) {
# Line 308 | Line 315 | public class ForkJoinWorkerThread extend
315       * one.  Marsaglia xor-shift is cheap and works well.
316       */
317      private static int xorShift(int r) {
318 <        r ^= r << 1;
319 <        r ^= r >>> 3;
320 <        r ^= r << 10;
314 <        return r;
318 >        r ^= (r << 13);
319 >        r ^= (r >>> 17);
320 >        return r ^ (r << 5);
321      }
322  
323      // Lifecycle methods
# Line 372 | Line 378 | public class ForkJoinWorkerThread extend
378       * {@code super.onTermination} at the end of the overridden method.
379       *
380       * @param exception the exception causing this thread to abort due
381 <     * to an unrecoverable error, or null if completed normally
381 >     * to an unrecoverable error, or {@code null} if completed normally
382       */
383      protected void onTermination(Throwable exception) {
384          // Execute remaining local tasks unless aborting or terminating
# Line 469 | Line 475 | public class ForkJoinWorkerThread extend
475      }
476  
477      /**
478 +     * Tries to take a task from the base of own queue, activating if
479 +     * necessary, failing only if empty. Called only by current thread.
480 +     *
481 +     * @return a task, or null if none
482 +     */
483 +    final ForkJoinTask<?> locallyDeqTask() {
484 +        int b;
485 +        while (sp != (b = base)) {
486 +            if (tryActivate()) {
487 +                ForkJoinTask<?>[] q = queue;
488 +                int i = (q.length - 1) & b;
489 +                ForkJoinTask<?> t = q[i];
490 +                if (t != null && casSlotNull(q, i, t)) {
491 +                    base = b + 1;
492 +                    return t;
493 +                }
494 +            }
495 +        }
496 +        return null;
497 +    }
498 +
499 +    /**
500       * Returns a popped task, or null if empty. Ensures active status
501       * if non-null. Called only by current thread.
502       */
# Line 508 | Line 536 | public class ForkJoinWorkerThread extend
536      }
537  
538      /**
539 <     * Returns next task.
539 >     * Returns next task or null if empty or contended
540       */
541      final ForkJoinTask<?> peekTask() {
542          ForkJoinTask<?>[] q = queue;
# Line 599 | Line 627 | public class ForkJoinWorkerThread extend
627       * @return a task, if available
628       */
629      final ForkJoinTask<?> pollTask() {
630 <        ForkJoinTask<?> t = locallyFifo ? deqTask() : popTask();
630 >        ForkJoinTask<?> t = locallyFifo ? locallyDeqTask() : popTask();
631          if (t == null && (t = scan()) != null)
632              ++stealCount;
633          return t;
# Line 611 | Line 639 | public class ForkJoinWorkerThread extend
639       * @return a task, if available
640       */
641      final ForkJoinTask<?> pollLocalTask() {
642 <        return locallyFifo ? deqTask() : popTask();
642 >        return locallyFifo ? locallyDeqTask() : popTask();
643      }
644  
645      /**
# Line 646 | Line 674 | public class ForkJoinWorkerThread extend
674       *
675       * @return the number of tasks drained
676       */
677 <    final int drainTasksTo(Collection<ForkJoinTask<?>> c) {
677 >    final int drainTasksTo(Collection<? super ForkJoinTask<?>> c) {
678          int n = 0;
679          ForkJoinTask<?> t;
680          while (base != sp && (t = deqTask()) != null) {
# Line 667 | Line 695 | public class ForkJoinWorkerThread extend
695      }
696  
697      /**
698 <     * Returns true if at least one worker in the given array appears
699 <     * to have at least one queued task.
698 >     * Returns {@code true} if at least one worker in the given array
699 >     * appears to have at least one queued task.
700       *
701       * @param ws array of workers
702       */
# Line 731 | Line 759 | public class ForkJoinWorkerThread extend
759          do {} while (!tryActivate()); // re-activate on exit
760      }
761  
762 <    // Unsafe mechanics for jsr166y 3rd party package.
735 <    private static sun.misc.Unsafe getUnsafe() {
736 <        try {
737 <            return sun.misc.Unsafe.getUnsafe();
738 <        } catch (SecurityException se) {
739 <            try {
740 <                return java.security.AccessController.doPrivileged
741 <                    (new java.security.PrivilegedExceptionAction<sun.misc.Unsafe>() {
742 <                        public sun.misc.Unsafe run() throws Exception {
743 <                            return getUnsafeByReflection();
744 <                        }});
745 <            } catch (java.security.PrivilegedActionException e) {
746 <                throw new RuntimeException("Could not initialize intrinsics",
747 <                                           e.getCause());
748 <            }
749 <        }
750 <    }
751 <
752 <    private static sun.misc.Unsafe getUnsafeByReflection()
753 <            throws NoSuchFieldException, IllegalAccessException {
754 <        java.lang.reflect.Field f =
755 <            sun.misc.Unsafe.class.getDeclaredField("theUnsafe");
756 <        f.setAccessible(true);
757 <        return (sun.misc.Unsafe) f.get(null);
758 <    }
759 <
760 <    private static long fieldOffset(String fieldName, Class<?> klazz) {
761 <        try {
762 <            return UNSAFE.objectFieldOffset(klazz.getDeclaredField(fieldName));
763 <        } catch (NoSuchFieldException e) {
764 <            // Convert Exception to Error
765 <            NoSuchFieldError error = new NoSuchFieldError(fieldName);
766 <            error.initCause(e);
767 <            throw error;
768 <        }
769 <    }
762 >    // Unsafe mechanics
763  
764      private static final sun.misc.Unsafe UNSAFE = getUnsafe();
765      private static final long spOffset =
766 <        fieldOffset("sp", ForkJoinWorkerThread.class);
766 >        objectFieldOffset("sp", ForkJoinWorkerThread.class);
767      private static final long runStateOffset =
768 <        fieldOffset("runState", ForkJoinWorkerThread.class);
768 >        objectFieldOffset("runState", ForkJoinWorkerThread.class);
769      private static final long qBase;
770      private static final int qShift;
771  
# Line 783 | Line 776 | public class ForkJoinWorkerThread extend
776              throw new Error("data type scale not a power of two");
777          qShift = 31 - Integer.numberOfLeadingZeros(s);
778      }
779 +
780 +    private static long objectFieldOffset(String field, Class<?> klazz) {
781 +        try {
782 +            return UNSAFE.objectFieldOffset(klazz.getDeclaredField(field));
783 +        } catch (NoSuchFieldException e) {
784 +            // Convert Exception to corresponding Error
785 +            NoSuchFieldError error = new NoSuchFieldError(field);
786 +            error.initCause(e);
787 +            throw error;
788 +        }
789 +    }
790 +
791 +    /**
792 +     * Returns a sun.misc.Unsafe.  Suitable for use in a 3rd party package.
793 +     * Replace with a simple call to Unsafe.getUnsafe when integrating
794 +     * into a jdk.
795 +     *
796 +     * @return a sun.misc.Unsafe
797 +     */
798 +    private static sun.misc.Unsafe getUnsafe() {
799 +        try {
800 +            return sun.misc.Unsafe.getUnsafe();
801 +        } catch (SecurityException se) {
802 +            try {
803 +                return java.security.AccessController.doPrivileged
804 +                    (new java.security
805 +                     .PrivilegedExceptionAction<sun.misc.Unsafe>() {
806 +                        public sun.misc.Unsafe run() throws Exception {
807 +                            java.lang.reflect.Field f = sun.misc
808 +                                .Unsafe.class.getDeclaredField("theUnsafe");
809 +                            f.setAccessible(true);
810 +                            return (sun.misc.Unsafe) f.get(null);
811 +                        }});
812 +            } catch (java.security.PrivilegedActionException e) {
813 +                throw new RuntimeException("Could not initialize intrinsics",
814 +                                           e.getCause());
815 +            }
816 +        }
817 +    }
818   }

Diff Legend

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