java线程池(七)：ForkJoinPool源码分析之三(ForkJoinTask源码）

a.fork();
b.fork();
b.join();
a.join();

public abstract class ForkJoinTask<V> implements Future<V>, Serializable {

    
}

/** The run status of this task */
volatile int status; // accessed directly by pool

 and workers
static final int DONE_MASK   = 0xf0000000;  // mask out non-completion bits
static final int NORMAL      = 0xf0000000;  // must be negative
static final int CANCELLED   = 0xc0000000;  // must be < NORMAL
static final int EXCEPTIONAL = 0x80000000;  // must be < CANCELLED
static final int SIGNAL      = 0x00010000;  // must be >= 1 << 16
static final int SMASK       = 0x0000ffff;  // short bits for tags

/**
 * Returns the result that would be returned by {@link #join}, even
 * if this task completed abnormally, or {@code null} if this task
 * is not known to have been completed.  This method is designed
 * to aid debugging, as well as to support extensions. Its use in
 * any other context is discouraged.
 *
 * @return the result, or {@code null} if not completed
 */
public abstract V getRawResult();

/**
 * Forces the given value to be returned as a result.  This method
 * is designed to support extensions, and should not in general be
 * called otherwise.
 *
 * @param value the value
 */
protected abstract void setRawResult(V value);

/**
 * Immediately performs the base action of this task and returns
 * true if, upon return from this method, this task is guaranteed
 * to have completed normally. This method may return false
 * otherwise, to indicate that this task is not necessarily
 * complete (or is not known to be complete), for example in
 * asynchronous actions that require explicit invocations of
 * completion methods. This method may also throw an (unchecked)
 * exception to indicate abnormal exit. This method is designed to
 * support extensions, and should not in general be called
 * otherwise.
 *
 * @return {@code true} if this task is known to have completed normally
 */
protected abstract boolean exec();

public final V get() throws InterruptedException, ExecutionException {
    //判断当前线程是否是工作线程
    int s = (Thread.currentThread() instanceof ForkJoinWorkerThread) ?
       //如果是工作线程，执行doJoin方法
        doJoin() :
        //反之，则调用外部中断等待方法，将外部调用的线程wait。
        externalInterruptibleAwaitDone();
    Throwable ex;
    //如果返回的s结果，为取消状态，抛出异常
    if ((s &= DONE_MASK) == CANCELLED)
        throw new CancellationException();
    //如果s为异常状态，则调用getThrowableException从异常表中获取一个异常，再new一个ExecutionException包裹之后抛出。 
    if (s == EXCEPTIONAL && (ex = getThrowableException()) != null)
        throw new ExecutionException(ex);
    //默认返回结果是getRawResult。
    return getRawResult();
}

/**
 * Waits if necessary for at most the given time for the computation
 * to complete, and then retrieves its result, if available.
 *
 * @param timeout the maximum time to wait
 * @param unit the time unit of the timeout argument
 * @return the computed result
 * @throws CancellationException if the computation was cancelled
 * @throws ExecutionException if the computation threw an
 * exception
 * @throws InterruptedException if the current thread is not a
 * member of a ForkJoinPool and was interrupted while waiting
 * @throws TimeoutException if the wait timed out
 */
public final V get(long timeout, TimeUnit unit)
    throws InterruptedException, ExecutionException, TimeoutException {
    int s;
    //时间通过unit的单位转换之后按纳秒获取
    long nanos = unit.toNanos(timeout);
    //如果线程被中断，则返回中断异常
    if (Thread.interrupted())
        throw new InterruptedException();
    //如果status大于0，且纳秒数也大于0
    if ((s = status) >= 0 && nanos > 0L) {
        //计算时间d和deadline
        long d = System.nanoTime() + nanos;
        long deadline = (d == 0L) ? 1L : d; // avoid 0
        Thread t = Thread.currentThread();
        //如果当前线程是工作线程，调用pool提供的awaitJoin方法。
        if (t instanceof ForkJoinWorkerThread) {
            ForkJoinWorkerThread wt = (ForkJoinWorkerThread)t;
            s = wt.pool.awaitJoin(wt.workQueue, this, deadline);
        }
        //反之，再判断当前task是否是子类CountedCompleter，则采用externalHelpComplete
        else if ((s = ((this instanceof CountedCompleter) ?
                       ForkJoinPool.common.externalHelpComplete(
                           (CountedCompleter<?>)this, 0) :
                      //tryExternalUnpush方法 ForkJoinPool.common.tryExternalUnpush(this) ?
                       doExec() : 0)) >= 0) {
            long ns, ms; // measure in nanosecs, but wait in millisecs
            //死循环的方式确保wait成功
            while ((s = status) >= 0 &&
                   (ns = deadline - System.nanoTime()) > 0L) {
                if ((ms = TimeUnit.NANOSECONDS.toMillis(ns)) > 0L &&
                    U.compareAndSwapInt(this, STATUS, s, s | SIGNAL)) {
                    synchronized (this) {
                       //如果status大于0则wait，反之则botifyAll
                        if (status >= 0)
                            wait(ms); // OK to throw InterruptedException
                        else
                            notifyAll();
                    }
                }
            }
        }
    }
    //如果状态大于0 s更新status的值
    if (s >= 0)
        s = status;
    //如果s超出范围，则抛出异常，进行异常处理
    if ((s &= DONE_MASK) != NORMAL) {
        Throwable ex;
        if (s == CANCELLED)
            throw new CancellationException();
        if (s != EXCEPTIONAL)
            throw new TimeoutException();
        if ((ex = getThrowableException()) != null)
            throw new ExecutionException(ex);
    }
    return getRawResult();
}

/**
 * Blocks a non-worker-thread until completion or interruption.
 */
private int externalInterruptibleAwaitDone() throws InterruptedException {
    int s; 
    //如果线程中断，则抛出异常
    if (Thread.interrupted())
        throw new InterruptedException();
    //如果状态大于0且s为CountedCompleter的实现
    if ((s = status) >= 0 &&
        (s = ((this instanceof CountedCompleter) ?
             //此时将采用pool的externalHelpComplete方法，此方法会触发窃取操作 ForkJoinPool.common.externalHelpComplete(
                  (CountedCompleter<?>)this, 0) :
              ForkJoinPool.common.tryExternalUnpush(this) ? doExec() :
              0)) >= 0) {
        while ((s = status) >= 0) {
            if (U.compareAndSwapInt(this, STATUS, s, s | SIGNAL)) {
             //加锁，等待，注意此时wait的参数为0。那么这个线程只能等待被唤醒
                synchronized (this) {
                    if (status >= 0)
                        wait(0L);
                    else
                        notifyAll();
                }
            }
        }
    }
    return s;
}

/**
 * Attempts to cancel execution of this task. This attempt will
 * fail if the task has already completed or could not be
 * cancelled for some other reason. If successful, and this task
 * has not started when {@code cancel} is called, execution of
 * this task is suppressed. After this method returns
 * successfully, unless there is an intervening call to {@link
 * #reinitialize}, subsequent calls to {@link #isCancelled},
 * {@link #isDone}, and {@code cancel} will return {@code true}
 * and calls to {@link #join} and related methods will result in
 * {@code CancellationException}.
 *
 * <p>This method may be overridden in subclasses, but if so, must
 * still ensure that these properties hold. In particular, the
 * {@code cancel} method itself must not throw exceptions.
 *
 * <p>This method is designed to be invoked by <em>other</em>
 * tasks. To terminate the current task, you can just return or
 * throw an unchecked exception from its computation method, or
 * invoke {@link #completeExceptionally(Throwable)}.
 *
 * @param mayInterruptIfRunning this value has no effect in the
 * default implementation because interrupts are not used to
 * control cancellation.
 *
 * @return {@code true} if this task is now cancelled
 */
public boolean cancel(boolean mayInterruptIfRunning) {
    return (setCompletion(CANCELLED) & DONE_MASK) == CANCELLED;
}

public final boolean isDone() {
    return status < 0;
}

public final boolean isCancelled() {
    return (status & DONE_MASK) == CANCELLED;
}

/**
 * Arranges to asynchronously execute this task in the pool the
 * current task is running in, if applicable, or using the {@link
 * ForkJoinPool#commonPool()} if not {@link #inForkJoinPool}.  While
 * it is not necessarily enforced, it is a usage error to fork a
 * task more than once unless it has completed and been
 * reinitialized.  Subsequent modifications to the state of this
 * task or any data it operates on are not necessarily
 * consistently observable by any thread other than the one
 * executing it unless preceded by a call to {@link #join} or
 * related methods, or a call to {@link #isDone} returning {@code
 * true}.
 *
 * @return {@code this}, to simplify usage
 */
public final ForkJoinTask<V> fork() {
    Thread t;
    //判断是否为工作线程
    if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)
        //加入工作队列
        ((ForkJoinWorkerThread)t).workQueue.push(this);
    else
        //外部执行push操作
        ForkJoinPool.common.externalPush(this);
    return this;
}

/**
 * Returns the result of the computation when it {@link #isDone is
 * done}.  This method differs from {@link #get()} in that
 * abnormal completion results in {@code RuntimeException} or
 * {@code Error}, not {@code ExecutionException}, and that
 * interrupts of the calling thread do <em>not</em> cause the
 * method to abruptly return by throwing {@code
 * InterruptedException}.
 *
 * @return the computed result
 */
public final V join() {
    int s;
    if ((s = doJoin() & DONE_MASK) != NORMAL)
        reportException(s);
    return getRawResult();
}

/**
 * Implementation for join, get, quietlyJoin. Directly handles
 * only cases of already-completed, external wait, and
 * unfork+exec.  Others are relayed to ForkJoinPool.awaitJoin.
 *
 * @return status upon completion
 */
private int doJoin() {
    int s; Thread t; ForkJoinWorkerThread wt; ForkJoinPool.WorkQueue w;
    //如果status小于0 则直接返回s 
    return (s = status) < 0 ? s :
        ((t = Thread.currentThread()) instanceof
        //判断是否是工作队列
        ForkJoinWorkerThread) ?
        (w = (wt = (ForkJoinWorkerThread)t).workQueue).
        tryUnpush(this) && (s = doExec()) < 0 ? s :
        //等待join
        wt.pool.awaitJoin(w, this, 0L) :
        //外部等待执行完毕
        externalAwaitDone();
}

/**
 * Commences performing this task, awaits its completion if
 * necessary, and returns its result, or throws an (unchecked)
 * {@code RuntimeException} or {@code Error} if the underlying
 * computation did so.
 *
 * @return the computed result
 */
public final V invoke() {
    int s;
    //执行doInvoke 并判断其状态
    if ((s = doInvoke() & DONE_MASK) != NORMAL)
        reportException(s);
    return getRawResult();
}

/**
 * Implementation for invoke, quietlyInvoke.
 *
 * @return status upon completion
 */
private int doInvoke() {
    int s; Thread t; ForkJoinWorkerThread wt;
    //执行doExec方法，并返回，此时就是当前线程，有可能不是工作线程
    return (s = doExec()) < 0 ? s :
        ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
        //如果不是工作线程，通过pool的awaitJoin方法。
        (wt = (ForkJoinWorkerThread)t).pool.
        awaitJoin(wt.workQueue, this, 0L) :
        //执行外部等待执行完成的方法externalAwaitDone
        externalAwaitDone();
}

/**
 * Blocks a non-worker-thread until completion.
 * @return status upon completion
 */
private int externalAwaitDone() {
    //判断s是否是CountedCompleter实例
    int s = ((this instanceof CountedCompleter) ? // try helping
             //如果是则执行comm的外部协助窃取方法
             ForkJoinPool.common.externalHelpComplete(
                 (CountedCompleter<?>)this, 0) :
             ForkJoinPool.common.tryExternalUnpush(this) ? doExec() : 0);
    //如果s大于0 
    if (s >= 0 && (s = status) >= 0) {
        boolean interrupted = false;
        //循环 采用cas的方式
        do {
            //cas方式修改状态
            if (U.compareAndSwapInt(this, STATUS, s, s | SIGNAL)) {
               //同步加锁
                synchronized (this) {
                    if (status >= 0) {
                        try {
                           //将当前线程置于wait状态。
                            wait(0L);
                        } catch (InterruptedException ie) {
                            interrupted = true;
                        }
                    }
                    else
                    //反之 唤醒全部线程
                        notifyAll();
                }
            }
        } while ((s = status) >= 0);
        if (interrupted)
            Thread.currentThread().interrupt();
    }
    return s;
}

/**
 * Forks the given tasks, returning when {@code isDone} holds for
 * each task or an (unchecked) exception is encountered, in which
 * case the exception is rethrown. If more than one task
 * encounters an exception, then this method throws any one of
 * these exceptions. If any task encounters an exception, the
 * other may be cancelled. However, the execution status of
 * individual tasks is not guaranteed upon exceptional return. The
 * status of each task may be obtained using {@link
 * #getException()} and related methods to check if they have been
 * cancelled, completed normally or exceptionally, or left
 * unprocessed.
 *
 * @param t1 the first task
 * @param t2 the second task
 * @throws NullPointerException if any task is null
 */
public static void invokeAll(ForkJoinTask<?> t1, ForkJoinTask<?> t2) {
    int s1, s2;
    //第二个任务调用fork，用工作线程执行
    t2.fork();
    //第一个任务调用doInvoke，当前线程执行
    if ((s1 = t1.doInvoke() & DONE_MASK) != NORMAL)
        t1.reportException(s1);
    //之后用异步执行的第二个任务的结果阻塞
    if ((s2 = t2.doJoin() & DONE_MASK) != NORMAL)
        t2.reportException(s2);
    }

/**
 * Forks the given tasks, returning when {@code isDone} holds for
 * each task or an (unchecked) exception is encountered, in which
 * case the exception is rethrown. If more than one task
 * encounters an exception, then this method throws any one of
 * these exceptions. If any task encounters an exception, others
 * may be cancelled. However, the execution status of individual
 * tasks is not guaranteed upon exceptional return. The status of
 * each task may be obtained using {@link #getException()} and
 * related methods to check if they have been cancelled, completed
 * normally or exceptionally, or left unprocessed.
 *
 * @param tasks the tasks
 * @throws NullPointerException if any task is null
 */
public static void invokeAll(ForkJoinTask<?>... tasks) {
    Throwable ex = null;
    int last = tasks.length - 1;
    //除第0个之外的任务都会调用fork进行处理，而第0个会用当前线程进行处理
    for (int i = last; i >= 0; --i) {
        ForkJoinTask<?> t = tasks[i];
        if (t == null) {
            if (ex == null)
                ex = new NullPointerException();
        }
        else if (i != 0)
            t.fork();
        else if (t.doInvoke() < NORMAL && ex == null)
            ex = t.getException();
    }
    //遍历，除第0个任务之外，所有任务都调用doJoin，也就是说当前线程必须等待其他任务都执行完成。
    for (int i = 1; i <= last; ++i) {
        ForkJoinTask<?> t = tasks[i];
        if (t != null) {
            if (ex != null)
                t.cancel(false);
            else if (t.doJoin() < NORMAL)
                ex = t.getException();
        }
    }
    if (ex != null)
        rethrow(ex);
}

// Exception table support

//执行任务的时候抛出的异常表，使调用者能够进行报告，由于异常很少见，因此我们不直接将其保存在任务对象中，而是使用弱引用表。请注意，取消异常未出现在表中，而是记录状态值。注意，这些静态变量在下面的静态块中初始化。
private static final ExceptionNode[] exceptionTable;
private static final ReentrantLock exceptionTableLock;
private static final ReferenceQueue<Object> exceptionTableRefQueue;

//固定长度为32
private static final int EXCEPTION_MAP_CAPACITY = 32;

static final class ExceptionNode extends WeakReference<ForkJoinTask<?>> {
    final Throwable ex;
    //链表结构
    ExceptionNode next;
    final long thrower;  // use id not ref to avoid weak cycles
    //hashCode
    final int hashCode;  // store task hashCode before weak ref disappears
    ExceptionNode(ForkJoinTask<?> task, Throwable ex, ExceptionNode next) {
       //初始化赋值
        super(task, exceptionTableRefQueue);
        this.ex = ex;
        //下一个记录
        this.next = next;
        //记录线程ID
        this.thrower =
        Thread.currentThread().getId();
        //赋值hashCode
        this.hashCode = System.identityHashCode(task);
    }
}

/**
 * Records exception and sets status.
 *
 * @return status on exit
 */
final int recordExceptionalCompletion(Throwable ex) {
    int s;
    //如果s的状态大于0 说明task没有执行完
    if ((s = status) >= 0) {
       //获取当前task的hashCode
        int h = System.identityHashCode(this);
        //定义锁
        final ReentrantLock lock = exceptionTableLock;
        //加锁
        lock.lock();
        try {
           //调用expungeStaleExceptions
            expungeStaleExceptions();
            //获取异常表引用
            ExceptionNode[] t = exceptionTable;
            int i = h & (t.length - 1);
            //循环遍历
            for (ExceptionNode e = t[i]; ; e = e.next) {
                //在为空的位置插入当前异常
                if (e == null) {
                    t[i] = new ExceptionNode(this, ex, t[i]);
                    break;
                }
                if (e.get() == this) // already present
                    break;
            }
        } finally {
            //解锁
            lock.unlock();
        }
        s = setCompletion(EXCEPTIONAL);
    }
    return s;
}

private void clearExceptionalCompletion() {
    //识别hashCode
    int h = System.identityHashCode(this);
    //获得锁
    final ReentrantLock lock = exceptionTableLock;
    //加锁
    lock.lock();
    try {
        //拿到异常表
        ExceptionNode[] t = exceptionTable;
        int i = h & (t.length - 1);
        ExceptionNode e = t[i];
        ExceptionNode pred = null;
        //循环
        while (e != null) {
            ExceptionNode next = e.next;
            //将当前null清理
            if (e.get() == this) {
                if (pred == null)
                    t[i] = next;
                else
                    pred.next = next;
                break;
            }
            pred = e;
            e = next;
        }
        expungeStaleExceptions();
        status = 0;
    } finally {
        lock.unlock();
    }
}

private static void expungeStaleExceptions() {
   //循环
    for (Object x; (x = exceptionTableRefQueue.poll()) != null;) {
        //判断 如果x是弱引用的异常节点
        if (x instanceof ExceptionNode) {
            //取得hashCode
            int hashCode = ((ExceptionNode)x).hashCode;
            //获得异常表
            ExceptionNode[] t = exceptionTable;
            int i = hashCode & (t.length - 1);
            ExceptionNode e = t[i];
            ExceptionNode pred = null;
            //循环 遍历清除
            while (e != null) {
                ExceptionNode next = e.next;
                if (e == x) {
                    if (pred == null)
                        t[i] = next;
                    else
                        pred.next = next;
                    break;
                }
                pred = e;
                e = next;
            }
        }
    }
}

private Throwable getThrowableException() {
    if ((status & DONE_MASK) != EXCEPTIONAL)
        return null;
    //拿到当前task的hashCode
    int h = System.identityHashCode(this);
    ExceptionNode e;
    //加锁
    final ReentrantLock lock = exceptionTableLock;
    lock.lock();
    try {
        //清理无效的引用
        expungeStaleExceptions();
        ExceptionNode[] t = exceptionTable;
        e = t[h & (t.length - 1)];
        while (e != null && e.get() != this)
            e = e.next;
    } finally {
        //解锁
        lock.unlock();
    }
    Throwable ex;
    if (e == null || (ex = e.ex) == null)
        return null;
    //如果异常不是当前线程引起的
    if (e.thrower != Thread.currentThread().getId()) {
        Class<? extends Throwable> ec = ex.getClass();
        try {
            Constructor<?> noArgCtor = null;
            Constructor<?>[] cs = ec.getConstructors();// public ctors only
            //循环
            for (int i = 0; i < cs.length; ++i) {
                Constructor<?> c = cs[i];
                Class<?>[] ps = c.getParameterTypes();
                if (ps.length == 0)
                    noArgCtor = c;
                else if (ps.length == 1 && ps[0] == Throwable.class) {
                   //返回一个新的异常实例
                    Throwable wx = (Throwable)c.newInstance(ex);
                    return (wx == null) ? ex : wx;
                }
            }
            //如果noArgCtor不为空
            if (noArgCtor != null) {
                Throwable wx = (Throwable)(noArgCtor.newInstance());
                if (wx != null) {
                    //初始化原因
                    wx.initCause(ex);
                    return wx;
                }
            }
        } catch (Exception ignore) {
        }
    }
    return ex;
}

/**
 * Adaptor for Runnables. This implements RunnableFuture
 * to be compliant with AbstractExecutorService constraints
 * when used in ForkJoinPool.
 */
static final class AdaptedRunnable<T> extends ForkJoinTask<T>
    implements RunnableFuture<T> {
    //需要适配的runnable
    final Runnable runnable;
    T result;
    AdaptedRunnable(Runnable runnable, T result) {
        if (runnable == null) throw new NullPointerException();
        this.runnable = runnable;
        this.result = result; // OK to set this even before completion
    }
    public final T getRawResult() { return result; }
    public final void setRawResult(T v) { result = v; }
    public final boolean exec() { runnable.run(); return true; }
    //调用invoke方法
    public final void run() { invoke(); }
    private static final long serialVersionUID = 5232453952276885070L;
}

static final class AdaptedRunnableAction extends ForkJoinTask<Void>
    implements RunnableFuture<Void> {
    final Runnable runnable;
    AdaptedRunnableAction(Runnable runnable) {
        if (runnable == null) throw new NullPointerException();
        this.runnable = runnable;
    }
    public final Void getRawResult() { return null; }
    public final void setRawResult(Void v) { }
    public final boolean exec() { runnable.run(); return true; }
    public final void run() { invoke(); }
    private static final long serialVersionUID = 5232453952276885070L;
}

static final class RunnableExecuteAction extends ForkJoinTask<Void> {
    final Runnable runnable;
    RunnableExecuteAction(Runnable runnable) {
        if (runnable == null) throw new NullPointerException();
        this.runnable = runnable;
    }
    public final Void getRawResult() { return null; }
    public final void setRawResult(Void v) { }
    public final boolean exec() { runnable.run(); return true; }
    void internalPropagateException(Throwable ex) {
        rethrow(ex); // rethrow outside exec() catches.
    }
    private static final long serialVersionUID = 5232453952276885070L;
}

static final class AdaptedCallable<T> extends ForkJoinTask<T>
    implements RunnableFuture<T> {
    final Callable<? extends T> callable;
    T result;
    AdaptedCallable(Callable<? extends T> callable) {
        if (callable == null) throw new NullPointerException();
        this.callable = callable;
    }
    public final T getRawResult() { return result; }
    public final void setRawResult(T v) { result = v; }
    public final boolean exec() {
        try {
            result = callable.call();
            return true;
        } catch (Error err) {
            throw err;
        } catch (RuntimeException rex) {
            throw rex;
        } catch (Exception ex) {
            throw new RuntimeException(ex);
        }
    }
    public final void run() { invoke(); }
    private static final long serialVersionUID = 2838392045355241008L;
}

public static <T> ForkJoinTask<T> adapt(Callable<? extends T> callable) {
    return new AdaptedCallable<T>(callable);
}

public static <T> ForkJoinTask<T> adapt(Runnable runnable, T result) {
    return new AdaptedRunnable<T>(runnable, result);
}

public static ForkJoinTask<?> adapt(Runnable runnable) {
    return new AdaptedRunnableAction(runnable);
}