(juc系列)forkjoin框架源码学习

        volatile int source;       // source queue id, or sentinel 源队列ID
        int id;                    // pool index, mode, tag 池ID
        int base;                  // index of next slot for poll // 下一个拿的index
        int top;                   // index of next slot for push // 下一个放的index
        volatile int phase;        // versioned, negative: queued, 1: locked // 1是锁定. 负数是有队列
        int stackPred;             // pool stack (ctl) predecessor link // 
        int nsteals;               // number of steals // 偷取任务数量
        ForkJoinTask<?>[] array;   // the queued tasks; power of 2 size // 队列中的任务
        final ForkJoinPool pool;   // the containing pool (may be null) // 池子
        final ForkJoinWorkerThread owner; // owning thread or null if shared // 所属线程

        final void push(ForkJoinTask<?> task) {
            ForkJoinTask<?>[] a;
            int s = top, d = s - base, cap, m;
            ForkJoinPool p = pool;
            // 已有队列
            if ((a = array) != null && (cap = a.length) > 0) {
                // CAS更新任务
                QA.setRelease(a, (m = cap - 1) & s, task);
                // 下标+1
                top = s + 1;
                // 数组满了,扩容
                if (d == m)
                    growArray(false);
                else if (QA.getAcquire(a, m & (s - 1)) == null && p != null) {
                    VarHandle.fullFence();  // was empty
                    // 新搞一个线程过来? TODO
                    p.signalWork(null);
                }
            }
        }

        final ForkJoinTask<?> poll() {
            int b, k, cap; ForkJoinTask<?>[] a;
            // 队列中有值，
            while ((a = array) != null && (cap = a.length) > 0 &&
                   top - (b = base) > 0) {
                // 从数组中获取一个任务
                ForkJoinTask<?> t = (ForkJoinTask<?>)
                    QA.getAcquire(a, k = (cap - 1) & b);
                if (base == b++) {
                    if (t == null)
                        Thread.yield(); // await index advance
                    // 置为空
                    else if (QA.compareAndSet(a, k, t, null)) {
                        BASE.setOpaque(this, b);
                        // 返回任务
                        return t;
                    }
                }
            }
            return null;
        }

        // 获取第一个任务
        final ForkJoinTask<?> peek() {
        int cap; ForkJoinTask<?>[] a;
        return ((a = array) != null && (cap = a.length) > 0) ?
        a[(cap - 1) & ((id & FIFO) != 0 ? base : top - 1)] : null;
        }

// 权限
static final RuntimePermission modifyThreadPermission;

// common pool
static final ForkJoinPool common;

// 并发度
static final int COMMON_PARALLELISM;

// 偷取数量
volatile long stealCount;            // collects worker nsteals
// 保持活跃的时间
final long keepAlive;                // milliseconds before dropping if idle
// 下一个工作线程的下标
int indexSeed;                       // next worker index
// 最小最大线程
final int bounds;                    // min, max threads packed as shorts
// 并发度
volatile int mode;                   // parallelism, runstate, queue mode
// 工作队列
WorkQueue[] workQueues;              // main registry
// 工作线程的前缀
final String workerNamePrefix;       // for worker thread string; sync lock
// 线程工厂
final ForkJoinWorkerThreadFactory factory;
// 异常处理器
final UncaughtExceptionHandler ueh;  // per-worker UEH
// 是否饱和的判断方法
final Predicate<? super ForkJoinPool> saturate;

// 核心的状态控制
@jdk.internal.vm.annotation.Contended("fjpctl") // segregate
volatile long ctl;                   // main pool control

    public ForkJoinPool() {
        this(Math.min(MAX_CAP, Runtime.getRuntime().availableProcessors()),
             defaultForkJoinWorkerThreadFactory, null, false,
             0, MAX_CAP, 1, null, DEFAULT_KEEPALIVE, TimeUnit.MILLISECONDS);
    }

    public ForkJoinPool(int parallelism) {
        this(parallelism, defaultForkJoinWorkerThreadFactory, null, false,
             0, MAX_CAP, 1, null, DEFAULT_KEEPALIVE, TimeUnit.MILLISECONDS);
    }

    public ForkJoinPool(int parallelism,
                        ForkJoinWorkerThreadFactory factory,
                        UncaughtExceptionHandler handler,
                        boolean asyncMode) {
        this(parallelism, factory, handler, asyncMode,
             0, MAX_CAP, 1, null, DEFAULT_KEEPALIVE, TimeUnit.MILLISECONDS);
    }

    public ForkJoinPool(int parallelism,
                        ForkJoinWorkerThreadFactory factory,
                        UncaughtExceptionHandler handler,
                        boolean asyncMode,
                        int corePoolSize,
                        int maximumPoolSize,
                        int minimumRunnable,
                        Predicate<? super ForkJoinPool> saturate,
                        long keepAliveTime,
                        TimeUnit unit) {
        // check, encode, pack parameters
        // 并行度
        if (parallelism <= 0 || parallelism > MAX_CAP ||
            maximumPoolSize < parallelism || keepAliveTime <= 0L)
            throw new IllegalArgumentException();
        // 工厂
        if (factory == null)
            throw new NullPointerException();
        // 活跃时间
        long ms = Math.max(unit.toMillis(keepAliveTime), TIMEOUT_SLOP);

        // 线程数量
        int corep = Math.min(Math.max(corePoolSize, parallelism), MAX_CAP);
        // ctl变量的值
        long c = ((((long)(-corep)       << TC_SHIFT) & TC_MASK) |
                  (((long)(-parallelism) << RC_SHIFT) & RC_MASK));
        // mode
        int m = parallelism | (asyncMode ? FIFO : 0);
        
        int maxSpares = Math.min(maximumPoolSize, MAX_CAP) - parallelism;
        int minAvail = Math.min(Math.max(minimumRunnable, 0), MAX_CAP);
        
        // bounds
        int b = ((minAvail - parallelism) & SMASK) | (maxSpares << SWIDTH);
        int n = (parallelism > 1) ? parallelism - 1 : 1; // at least 2 slots
        
        // 初始工作队列的数量
        n |= n >>> 1; n |= n >>> 2; n |= n >>> 4; n |= n >>> 8; n |= n >>> 16;
        n = (n + 1) << 1; // power of two, including space for submission queues
        this.workQueues = new WorkQueue[n];
        // 线程池前缀
        this.workerNamePrefix = "ForkJoinPool-" + nextPoolId() + "-worker-";
        
        // 赋值
        this.factory = factory;
        this.ueh = handler;
        this.saturate = saturate;
        this.keepAlive = ms;
        this.bounds = b;
        this.mode = m;
        this.ctl = c;
        checkPermission();
    }

    public void execute(Runnable task) {
        if (task == null)
            throw new NullPointerException();
        ForkJoinTask<?> job;
        if (task instanceof ForkJoinTask<?>) // avoid re-wrap
            job = (ForkJoinTask<?>) task;
        else
            job = new ForkJoinTask.RunnableExecuteAction(task);
        // 核心的外部提交方法
        externalSubmit(job);
    }

    private <T> ForkJoinTask<T> externalSubmit(ForkJoinTask<T> task) {
        Thread t; ForkJoinWorkerThread w; WorkQueue q;
        if (task == null)
            throw new NullPointerException();
        // 当前线程就是一个`ForkJoin`类型的线程，直接调用该线程的队列进行push, 说明是内部分裂开的任务,直接入队当前线程的队列
        if (((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) &&
            (w = (ForkJoinWorkerThread)t).pool == this &&
            (q = w.workQueue) != null)
            // 调用上方介绍过的`workQueue.push`方法
            q.push(task);
        else
            // 调用externalPush进行提交任务
            externalPush(task);
        return task;
    }

    final void externalPush(ForkJoinTask<?> task) {
        int r;                                // initialize caller's probe
        // 随机一个探针
        if ((r = ThreadLocalRandom.getProbe()) == 0) {
            ThreadLocalRandom.localInit();
            r = ThreadLocalRandom.getProbe();
        }
        for (;;) {
            WorkQueue q;
            int md = mode, n;
            WorkQueue[] ws = workQueues;
            if ((md & SHUTDOWN) != 0 || ws == null || (n = ws.length) <= 0)
                throw new RejectedExecutionException();
            // 该位置为空. 新建一个工作队列，加锁入队.
            else if ((q = ws[(n - 1) & r & SQMASK]) == null) { // add queue
                int qid = (r | QUIET) & ~(FIFO | OWNED);
                Object lock = workerNamePrefix;
                // 新建一个工作队列的操作
                ForkJoinTask<?>[] qa =
                    new ForkJoinTask<?>[INITIAL_QUEUE_CAPACITY];
                q = new WorkQueue(this, null);
                q.array = qa;
                q.id = qid;
                q.source = QUIET;
                if (lock != null) {     // unless disabled, lock pool to install
                    synchronized (lock) {
                        // 放到对应位置上.
                        WorkQueue[] vs; int i, vn;
                        if ((vs = workQueues) != null && (vn = vs.length) > 0 &&
                            vs[i = qid & (vn - 1) & SQMASK] == null)
                            vs[i] = q;  // else another thread already installed
                    }
                }
            }
            // 如果工作队列的当前位置在忙，重新随机一个位置.
            else if (!q.tryLockPhase()) // move if busy
                r = ThreadLocalRandom.advanceProbe(r);
            else {
                // 该位置不为空,且不忙,就唤醒来干活了.
                if (q.lockedPush(task))
                    signalWork(null);
                return;
            }
        }
    }

    final int doExec() {
        int s; boolean completed;
        // 当前任务状态正常
        if ((s = status) >= 0) {
            try {
                // 调用抽象方法,进行任务的实际执行过程.
                completed = exec();
            } catch (Throwable rex) {
                completed = false;
                // 遇到异常了
                s = setExceptionalCompletion(rex);
            }
            if (completed)
                // 任务完成
                s = setDone();
        }
        return s;
    }

    public final ForkJoinTask<V> fork() {
        Thread t;
        // 如果是工作线程的子任务切分,直接调用之前的`workQueue.push`将任务添加到当前线程的任务队列中去
        if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)
            ((ForkJoinWorkerThread)t).workQueue.push(this);
        else
            // 如果是外部服务,直接调用fork,则调用之前的`externalPush`进行一个任务的提交
            ForkJoinPool.common.externalPush(this);
        return this;
    }

    public final V join() {
        int s;
        // 调用doJoin如果出错,报告异常.
        if (((s = doJoin()) & ABNORMAL) != 0)
            reportException(s);
        // 拿到结果
        return getRawResult();
    }

public abstract class RecursiveAction extends ForkJoinTask<Void> {
    private static final long serialVersionUID = 5232453952276485070L;

    /**
     * The main computation performed by this task.
     */
    protected abstract void compute();

    /**
     * Always returns {@code null}.
     *
     * @return {@code null} always
     */
    // 返回值永远是空
    public final Void getRawResult() { return null; }

    /**
     * Requires null completion value.
     */
    protected final void setRawResult(Void mustBeNull) { }

    /**
     * Implements execution conventions for RecursiveActions.
     */
    protected final boolean exec() {
        compute();
        return true;
    }
}

 static class SortTask extends RecursiveAction {
   final long[] array; final int lo, hi;
   
   SortTask(long[] array, int lo, int hi) {
     this.array = array; this.lo = lo; this.hi = hi;
   }
   
   SortTask(long[] array) { this(array, 0, array.length); }
   
   // 实现计算接口
   protected void compute() {
     // 排序范围不大,就排序
     if (hi - lo < THRESHOLD)
       sortSequentially(lo, hi);
     else {
         // 排序范围太大,就切分成两个任务,进行任务的提交
       int mid = (lo + hi) >>> 1;
       invokeAll(new SortTask(array, lo, mid),
                 new SortTask(array, mid, hi));
       merge(lo, mid, hi);
     }
   }
   // implementation details follow:
   static final int THRESHOLD = 1000;
   void sortSequentially(int lo, int hi) {
     Arrays.sort(array, lo, hi);
   }
   
   // 合并结果集
   void merge(int lo, int mid, int hi) {
     long[] buf = Arrays.copyOfRange(array, lo, mid);
     for (int i = 0, j = lo, k = mid; i < buf.length; j++)
       array[j] = (k == hi || buf[i] < array[k]) ?
         buf[i++] : array[k++];
   }
 }

 class IncrementTask extends RecursiveAction {
   final long[] array; final int lo, hi;
   IncrementTask(long[] array, int lo, int hi) {
     this.array = array; this.lo = lo; this.hi = hi;
   }
   protected void compute() {
     if (hi - lo < THRESHOLD) {
       for (int i = lo; i < hi; ++i)
         array[i]++;
     }
     else {
       int mid = (lo + hi) >>> 1;
       invokeAll(new IncrementTask(array, lo, mid),
                 new IncrementTask(array, mid, hi));
     }
   }
 }

 double sumOfSquares(ForkJoinPool pool, double[] array) {
   int n = array.length;
   Applyer a = new Applyer(array, 0, n, null);
   pool.invoke(a);
   return a.result;
 }

 class Applyer extends RecursiveAction {
   final double[] array;
   final int lo, hi;
   double result;
   Applyer next; // keeps track of right-hand-side tasks
   // 初始化
   Applyer(double[] array, int lo, int hi, Applyer next) {
     this.array = array; this.lo = lo; this.hi = hi;
     this.next = next;
   }

   // 叶子节点
   // 叶子节点不再继续分治,而是真的执行对应的计算
   double atLeaf(int l, int h) {
     double sum = 0;
     for (int i = l; i < h; ++i) // perform leftmost base step
       sum += array[i] * array[i];
     return sum;
   }

   protected void compute() {
     int l = lo;
     int h = hi;
     Applyer right = null;
     // 根据getSurplusQueuedTaskCount结果,动态的调整是否继续分治下去
     while (h - l > 1 && getSurplusQueuedTaskCount() <= 3) {
       int mid = (l + h) >>> 1;
       right = new Applyer(array, mid, h, right);
       right.fork();
       h = mid;
     }
     double sum = atLeaf(l, h);
     while (right != null) {
         // 如果右边的节点没有被偷, 继续计算
       if (right.tryUnfork()) // directly calculate if not stolen
         sum += right.atLeaf(right.lo, right.hi);
       else {
         right.join();
         sum += right.result;
       }
       right = right.next;
     }
     result = sum;
   }
 }

public abstract class RecursiveTask<V> extends ForkJoinTask<V> {
    private static final long serialVersionUID = 5232453952276485270L;

    /**
     * The result of the computation.
     */
    V result;

    /**
     * The main computation performed by this task.
     * @return the result of the computation
     */
    protected abstract V compute();

    // 返回结果
    public final V getRawResult() {
        return result;
    }

    // 设置结果
    protected final void setRawResult(V value) {
        result = value;
    }

    /**
     * Implements execution conventions for RecursiveTask.
     */
    protected final boolean exec() {
        result = compute();
        return true;
    }

}

 class Fibonacci extends RecursiveTask<Integer> {
   final int n;
   
   Fibonacci(int n) { this.n = n; }
   
   protected Integer compute() {
     // 小于1,返回
     if (n <= 1)
       return n;
     // 分别计算n-1和n-2.
     Fibonacci f1 = new Fibonacci(n - 1);
     f1.fork();
     Fibonacci f2 = new Fibonacci(n - 2);
     // join返回
     return f2.compute() + f1.join();
   }
 }

  
 public static <E> void forEach(E[] array, Consumer<E> action) {
   class Task extends CountedCompleter<Void> {
     final int lo, hi;
     Task(Task parent, int lo, int hi) {
       super(parent); this.lo = lo; this.hi = hi;
     }

     public void compute() {
       if (hi - lo >= 2) {
         int mid = (lo + hi) >>> 1;
         // must set pending count before fork
         // 分解任务后, 需要先设置需要等待的子任务的数量
         setPendingCount(2);
         // 然后调用子任务的fork
         new Task(this, mid, hi).fork(); // right child
         new Task(this, lo, mid).fork(); // left child
       }
       else if (hi > lo)
           // 执行操作
         action.accept(array[lo]);
       // 尝试完成整个任务.
       tryComplete();
     }
   }
   new Task(null, 0, array.length).invoke();
 }

    final ForkJoinPool pool;                // the pool this thread works in
    final ForkJoinPool.WorkQueue workQueue; // work-stealing mechanics

    protected ForkJoinWorkerThread(ForkJoinPool pool) {
        // Use a placeholder until a useful name can be set in registerWorker
        super("aForkJoinWorkerThread");
        this.pool = pool;
        this.workQueue = pool.registerWorker(this);
    }

    ForkJoinWorkerThread(ForkJoinPool pool, ClassLoader ccl) {
        super("aForkJoinWorkerThread");
        super.setContextClassLoader(ccl);
        ThreadLocalRandom.setInheritedAccessControlContext(this, INNOCUOUS_ACC);
        this.pool = pool;
        this.workQueue = pool.registerWorker(this);
    }

    ForkJoinWorkerThread(ForkJoinPool pool,
                         ClassLoader ccl,
                         ThreadGroup threadGroup,
                         AccessControlContext acc) {
        super(threadGroup, null, "aForkJoinWorkerThread");
        super.setContextClassLoader(ccl);
        ThreadLocalRandom.setInheritedAccessControlContext(this, acc);
        ThreadLocalRandom.eraseThreadLocals(this); // clear before registering
        this.pool = pool;
        this.workQueue = pool.registerWorker(this);
    }

    final WorkQueue registerWorker(ForkJoinWorkerThread wt) {
        UncaughtExceptionHandler handler;
        wt.setDaemon(true);                             // configure thread
        if ((handler = ueh) != null)
            wt.setUncaughtExceptionHandler(handler);
        int tid = 0;                                    // for thread name
        int idbits = mode & FIFO;
        String prefix = workerNamePrefix;
        // 以当前线程创建一个工作队列,
        WorkQueue w = new WorkQueue(this, wt);
        if (prefix != null) {
            synchronized (prefix) {
                WorkQueue[] ws = workQueues; int n;
                int s = indexSeed += SEED_INCREMENT;
                idbits |= (s & ~(SMASK | FIFO | DORMANT));
                if (ws != null && (n = ws.length) > 1) {
                    int m = n - 1;
                    tid = m & ((s << 1) | 1);           // odd-numbered indices
                    // 找个空的位置
                    for (int probes = n >>> 1;;) {      // find empty slot
                        WorkQueue q;
                        if ((q = ws[tid]) == null || q.phase == QUIET)
                            break;
                        else if (--probes == 0) {
                            tid = n | 1;                // resize below
                            break;
                        }
                        else
                            tid = (tid + 2) & m;
                    }
                    w.phase = w.id = tid | idbits;      // now publishable

                    if (tid < n)
                        // 创建的工作队列,放入到线程池的工作队列数组中去.
                        ws[tid] = w;
                    else {                              // expand array
                        // 重新创建工作队列的数组,
                        int an = n << 1;
                        WorkQueue[] as = new WorkQueue[an];
                        // 当前工作队列放进去
                        as[tid] = w;
                        int am = an - 1;
                        // 复制原有的所有任务过来
                        for (int j = 0; j < n; ++j) {
                            WorkQueue v;                // copy external queue
                            if ((v = ws[j]) != null)    // position may change
                                as[v.id & am & SQMASK] = v;
                            if (++j >= n)
                                break;
                            as[j] = ws[j];              // copy worker
                        }
                        workQueues = as;
                    }
                }
            }
            wt.setName(prefix.concat(Integer.toString(tid)));
        }
        return w;
    }

    public void run() {
        // 线程启动时,当前线程需要执行的任务必须为空.
        if (workQueue.array == null) { // only run once
            Throwable exception = null;
            try {
                // 调用启动前的hook
                onStart();
                // 启动当前工作线程
                pool.runWorker(workQueue);
            } catch (Throwable ex) {
                exception = ex;
            } finally {
                try {
                    // 调用终止时的hook.
                    onTermination(exception);
                } catch (Throwable ex) {
                    if (exception == null)
                        exception = ex;
                } finally {
                    // 注销当前工作线程
                    pool.deregisterWorker(this, exception);
                }
            }
        }
    }

    final void runWorker(WorkQueue w) {
        int r = (w.id ^ ThreadLocalRandom.nextSecondarySeed()) | FIFO; // rng
        // 初始化任务数组
        w.array = new ForkJoinTask<?>[INITIAL_QUEUE_CAPACITY]; // initialize
        for (;;) {
            int phase;
            // 扫描到任务了.
            if (scan(w, r)) {                     // scan until apparently empty
                r ^= r << 13; r ^= r >>> 17; r ^= r << 5; // move (xorshift)
            }
            else if ((phase = w.phase) >= 0) {    // enqueue, then rescan
                long np = (w.phase = (phase + SS_SEQ) | UNSIGNALLED) & SP_MASK;
                long c, nc;
                do {
                    w.stackPred = (int)(c = ctl);
                    nc = ((c - RC_UNIT) & UC_MASK) | np;
                } while (!CTL.weakCompareAndSet(this, c, nc));
            }
            else {                                // already queued
                // 没任务了,休眠一阵时间
                int pred = w.stackPred;
                Thread.interrupted();             // clear before park
                w.source = DORMANT;               // enable signal
                long c = ctl;
                int md = mode, rc = (md & SMASK) + (int)(c >> RC_SHIFT);
                if (md < 0)                       // terminating
                    break;
                else if (rc <= 0 && (md & SHUTDOWN) != 0 &&
                         tryTerminate(false, false))
                    break;                        // quiescent shutdown
                else if (w.phase < 0) {
                    if (rc <= 0 && pred != 0 && phase == (int)c) {
                        long nc = (UC_MASK & (c - TC_UNIT)) | (SP_MASK & pred);
                        long d = keepAlive + System.currentTimeMillis();
                        LockSupport.parkUntil(this, d);
                        if (ctl == c &&           // drop on timeout if all idle
                            d - System.currentTimeMillis() <= TIMEOUT_SLOP &&
                            CTL.compareAndSet(this, c, nc)) {
                            w.phase = QUIET;
                            break;
                        }
                    }
                    else {
                        LockSupport.park(this);
                        if (w.phase < 0)          // one spurious wakeup check
                            LockSupport.park(this);
                    }
                }
                w.source = 0;                     // disable signal
            }
        }
    }

    private boolean scan(WorkQueue w, int r) {
        WorkQueue[] ws; int n;
        // 检查参数, 当前线程池必须有工作队列的数组,且要扫描的当前工作队列不为空
        if ((ws = workQueues) != null && (n = ws.length) > 0 && w != null) {
            for (int m = n - 1, j = r & m;;) {
                WorkQueue q; int b;
                if ((q = ws[j]) != null && q.top != (b = q.base)) {
                    int qid = q.id;
                    ForkJoinTask<?>[] a; int cap, k; ForkJoinTask<?> t;
                    if ((a = q.array) != null && (cap = a.length) > 0) {
                        t = (ForkJoinTask<?>)QA.getAcquire(a, k = (cap - 1) & b);
                        if (q.base == b++ && t != null &&
                            QA.compareAndSet(a, k, t, null)) {
                            q.base = b;
                            w.source = qid;
                            // 如果任务比较多,唤醒其他工作线程
                            if (a[(cap - 1) & b] != null)
                                signalWork(q);    // help signal if more tasks
                            // 当前工作线程干活
                            w.topLevelExec(t, q,  // random fairness bound
                                           (r | (1 << TOP_BOUND_SHIFT)) & SMASK);
                        }
                    }
                    return true;
                }
                else if (--n > 0)
                    j = (j + 1) & m;
                else
                    break;
            }
        }
        return false;
    }

    final void signalWork(WorkQueue q) {
        for (;;) {
            long c; int sp; WorkQueue[] ws; int i; WorkQueue v;
        // 有足够多的工作线程,说明不需要唤醒了，退出
        if ((c = ctl) >= 0L)                      // enough workers
                break;
            // 没有空闲线程
            else if ((sp = (int)c) == 0) {            // no idle workers
                // 线程数很少，添加一个线程
                if ((c & ADD_WORKER) != 0L)           // too few workers
                    tryAddWorker(c);
                break;
            }
            // 线程池终止了,退出
            else if ((ws = workQueues) == null)
                break;                                // unstarted/terminated
            else if (ws.length <= (i = sp & SMASK))
                break;                                // terminated
            else if ((v = ws[i]) == null)
                break;                                // terminating
            else {
                // 唤醒一个其他的线程
                int np = sp & ~UNSIGNALLED;
                int vp = v.phase;
                long nc = (v.stackPred & SP_MASK) | (UC_MASK & (c + RC_UNIT));
                Thread vt = v.owner;
                if (sp == vp && CTL.compareAndSet(this, c, nc)) {
                    v.phase = np;
                    if (vt != null && v.source < 0)
                        LockSupport.unpark(vt);
                    break;
                }
                else if (q != null && q.isEmpty())     // no need to retry
                    break;
            }
        }
    }

    // 尝试添加一个工作线程
    private void tryAddWorker(long c) {
        do {
            long nc = ((RC_MASK & (c + RC_UNIT)) |
                       (TC_MASK & (c + TC_UNIT)));
            if (ctl == c && CTL.compareAndSet(this, c, nc)) {
                // 创建线程
                createWorker();
                break;
            }
        } while (((c = ctl) & ADD_WORKER) != 0L && (int)c == 0);
    }
    
    // 创建工作线程
    private boolean createWorker() {
        ForkJoinWorkerThreadFactory fac = factory;
        Throwable ex = null;
        ForkJoinWorkerThread wt = null;
        try {
            // 创建一个线程并运行
            if (fac != null && (wt = fac.newThread(this)) != null) {
                // 默认的实现是下方的Thread.
                wt.start();
                return true;
            }
        } catch (Throwable rex) {
            ex = rex;
        }
        // 注销一个工作线程
        deregisterWorker(wt, ex);
        return false;
    }

        final void topLevelExec(ForkJoinTask<?> t, WorkQueue q, int n) {
            int nstolen = 1;
            for (int j = 0;;) {
                // 调用`doExec`执行一个`ForkJoinTask`.
                if (t != null)
                    t.doExec();
                if (j++ <= n)
                    t = nextLocalTask();
                else {
                    j = 0;
                    t = null;
                }
                if (t == null) {
                    // 从q里面偷一个任务过来执行
                    if (q != null && (t = q.poll()) != null) {
                        ++nstolen;
                        j = 0;
                    }
                    else if (j != 0)
                        break;
                }
            }
            ForkJoinWorkerThread thread = owner;
            nsteals += nstolen;
            source = 0;
            if (thread != null)
                thread.afterTopLevelExec();
        }

    task -> task1 + task2
    task1.fork();
    task2.compute();

    task -> task1 + task2
    task1.fork();
    task1.join();
    task2.fork();
    task2.join();