jdk-sandbox: comparison jdk/src/share/classes/java/util/concurrent/ForkJoinTask.java

equal deleted inserted replaced

-:b518cd4045bc
+:d25399d849bc
 * (DAG). Otherwise, executions may encounter a form of deadlock as
 * tasks cyclically wait for each other.  However, this framework
 * supports other methods and techniques (for example the use of
 * {@link Phaser}, {@link #helpQuiesce}, and {@link #complete}) that
 * may be of use in constructing custom subclasses for problems that
-* are not statically structured as DAGs. To support such usages a
+* are not statically structured as DAGs. To support such usages, a
 * ForkJoinTask may be atomically <em>tagged</em> with a {@code short}
 * value using {@link #setForkJoinTaskTag} or {@link
 * #compareAndSetForkJoinTaskTag} and checked using {@link
 * #getForkJoinTaskTag}. The ForkJoinTask implementation does not use
 * these {@code protected} methods or tags for any purpose, but they
 * Blocks a non-worker-thread until completion.
 * @return status upon completion
 */
 private int externalAwaitDone() {
 int s;
-ForkJoinPool.externalHelpJoin(this);
+ForkJoinPool cp = ForkJoinPool.common;
-boolean interrupted = false;
+if ((s = status) >= 0) {
-while ((s = status) >= 0) {
+if (cp != null) {
-if (U.compareAndSwapInt(this, STATUS, s, s | SIGNAL)) {
+if (this instanceof CountedCompleter)
-synchronized (this) {
+s = cp.externalHelpComplete((CountedCompleter<?>)this, Integer.MAX_VALUE);
-if (status >= 0) {
+else if (cp.tryExternalUnpush(this))
-try {
+s = doExec();
-wait();
+}
-} catch (InterruptedException ie) {
+if (s >= 0 && (s = status) >= 0) {
-interrupted = true;
+boolean interrupted = false;
+do {
+if (U.compareAndSwapInt(this, STATUS, s, s | SIGNAL)) {
+synchronized (this) {
+if (status >= 0) {
+try {
+wait();
+} catch (InterruptedException ie) {
+interrupted = true;
+}
+}
+else
+notifyAll();
 }
 }
-else
+} while ((s = status) >= 0);
-notifyAll();
+if (interrupted)
-}
+Thread.currentThread().interrupt();
 }
 }
-if (interrupted)
-Thread.currentThread().interrupt();
 return s;
 }
 /**
 * Blocks a non-worker-thread until completion or interruption.
 */
 private int externalInterruptibleAwaitDone() throws InterruptedException {
 int s;
+ForkJoinPool cp = ForkJoinPool.common;
 if (Thread.interrupted())
 throw new InterruptedException();
-ForkJoinPool.externalHelpJoin(this);
+if ((s = status) >= 0 && cp != null) {
+if (this instanceof CountedCompleter)
+cp.externalHelpComplete((CountedCompleter<?>)this, Integer.MAX_VALUE);
+else if (cp.tryExternalUnpush(this))
+doExec();
+}
 while ((s = status) >= 0) {
 if (U.compareAndSwapInt(this, STATUS, s, s | SIGNAL)) {
 synchronized (this) {
 if (status >= 0)
 wait();
 }
 }
 }
 return s;
 }
 /**
 * Implementation for join, get, quietlyJoin. Directly handles
 * only cases of already-completed, external wait, and
 * unfork+exec.  Others are relayed to ForkJoinPool.awaitJoin.
 }
 /**
 * A version of "sneaky throw" to relay exceptions
 */
-static void rethrow(final Throwable ex) {
+static void rethrow(Throwable ex) {
-if (ex != null) {
+if (ex != null)
-if (ex instanceof Error)
-throw (Error)ex;
-if (ex instanceof RuntimeException)
-throw (RuntimeException)ex;
 ForkJoinTask.<RuntimeException>uncheckedThrow(ex);
-}
 }
 /**
 * The sneaky part of sneaky throw, relying on generics
 * limitations to evade compiler complaints about rethrowing
 * unchecked exceptions
 */
 @SuppressWarnings("unchecked") static <T extends Throwable>
 void uncheckedThrow(Throwable t) throws T {
-if (t != null)
+throw (T)t; // rely on vacuous cast
-throw (T)t; // rely on vacuous cast
 }
 /**
 * Throws exception, if any, associated with the given status.
 */
 if (Thread.interrupted())
 throw new InterruptedException();
 // Messy in part because we measure in nanosecs, but wait in millisecs
 int s; long ms;
 long ns = unit.toNanos(timeout);
+ForkJoinPool cp;
 if ((s = status) >= 0 && ns > 0L) {
 long deadline = System.nanoTime() + ns;
 ForkJoinPool p = null;
 ForkJoinPool.WorkQueue w = null;
 Thread t = Thread.currentThread();
 ForkJoinWorkerThread wt = (ForkJoinWorkerThread)t;
 p = wt.pool;
 w = wt.workQueue;
 p.helpJoinOnce(w, this); // no retries on failure
 }
-else
+else if ((cp = ForkJoinPool.common) != null) {
-ForkJoinPool.externalHelpJoin(this);
+if (this instanceof CountedCompleter)
+cp.externalHelpComplete((CountedCompleter<?>)this, Integer.MAX_VALUE);
+else if (cp.tryExternalUnpush(this))
+doExec();
+}
 boolean canBlock = false;
 boolean interrupted = false;
 try {
 while ((s = status) >= 0) {
 if (w != null && w.qlock < 0)
 cancelIgnoringExceptions(this);
 else if (!canBlock) {
-if (p == null || p.tryCompensate())
+if (p == null || p.tryCompensate(p.ctl))
 canBlock = true;
 }
 else {
 if ((ms = TimeUnit.NANOSECONDS.toMillis(ns)) > 0L &&
 U.compareAndSwapInt(this, STATUS, s, s | SIGNAL)) {
 */
 public boolean tryUnfork() {
 Thread t;
 return (((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
 ((ForkJoinWorkerThread)t).workQueue.tryUnpush(this) :
-ForkJoinPool.tryExternalUnpush(this));
+ForkJoinPool.common.tryExternalUnpush(this));
 }
 /**
 * Returns an estimate of the number of tasks that have been
 * forked by the current worker thread but not yet executed. This
 * before processing, otherwise exiting because the node has
 * already been visited.
 *
 * @param e the expected tag value
 * @param tag the new tag value
-* @return true if successful; i.e., the current value was
+* @return {@code true} if successful; i.e., the current value was
 * equal to e and is now tag.
 * @since 1.8
 */
 public final boolean compareAndSetForkJoinTaskTag(short e, short tag) {
 for (int s;;) {

changeset 18790	d25399d849bc
parent 16049	92a3a919d4dc
child 19048	7d0a94c79779