ForkJoinPool源码阅读

并发源码阅读

ForkJoinPool内容

externalPush

    final void externalPush(ForkJoinTask<?> task)
    {
        WorkQueue[] ws;
        WorkQueue q;
        int m;
        int r = ThreadLocalRandom.getProbe();
        int rs = runState;
        if ((ws = workQueues) != null && (m = (ws.length - 1)) >= 0 && (q = ws[m & r & SQMASK]) != null && r != 0 && rs > 0 && U.compareAndSwapInt(q, QLOCK, 0, 1))
        {
            ForkJoinTask<?>[] a;
            int am, n, s;
            if ((a = q.array) != null && (am = a.length - 1) > (n = (s = q.top) - q.base))
            {
                int j = ((am & s) << ASHIFT) + ABASE;
                U.putOrderedObject(a, j, task);
                U.putOrderedInt(q, QTOP, s + 1);
                U.putIntVolatile(q, QLOCK, 0);
                if (n <= 1)
                    signalWork(ws, q);
                return;
            }
            U.compareAndSwapInt(q, QLOCK, 1, 0);
        }
        //全量的提交执行
        externalSubmit(task);
    }

externalSubmit

主要完成几个工作

1. 如果整个Pool尚未初始化，则尝试cas竞争来初始化Pool。初始化内容是创建WQ数组
2. 通过随机下标在共享WQ区域找到一个WQ，抢占其控制权，放入一个任务对象
3. 如果随机下标的槽位为空，则创建一个共享的WQ，放入其中。

private void externalSubmit(ForkJoinTask<?> task)
    {
        //获得线程内的随机数，这个没什么好讲的。
        int r; // initialize caller's probe
        if ((r = ThreadLocalRandom.getProbe()) == 0)
        {
            ThreadLocalRandom.localInit();
            r = ThreadLocalRandom.getProbe();
        }
        for (;;)
        {
            WorkQueue[] ws;
            WorkQueue q;
            int rs, m, k;
            boolean move = false;
            if ((rs = runState) < 0)
            {
                tryTerminate(false, false); // help terminate
                throw new RejectedExecutionException();
            }
            //如果发现
            //1.线程池尚未启动
            //2.工作队列尚未初始化
            //3.工作队列长度为0
            //三个条件任意存在，则尝试执行初始化工作
            else if ((rs & STARTED) == 0 || // initialize
                    ((ws = workQueues) == null || (m = ws.length - 1) < 0))
            {
                int ns = 0;
                rs = lockRunState();
                try
                {
                    if ((rs & STARTED) == 0)
                    {
                        U.compareAndSwapObject(this, STEALCOUNTER, null, new AtomicLong());
                        // create workQueues array with size a power of two
                        //config包含了ForkJoinPool的配置信息。是一个int数字。其中低16位bit用于表达系统整体的并发数。第17位bit用于表达模式。先进先出还是后进先出。先进先出是队列，后进先出是栈。
                        int p = config & SMASK; // ensure at least 2 slots
                        int n = (p > 1) ? p - 1 : 1;
                        n |= n >>> 1;
                        n |= n >>> 2;
                        n |= n >>> 4;
                        n |= n >>> 8;
                        n |= n >>> 16;
                        n = (n + 1) << 1;
                        //上面的算法是一个固定的快速算法，目的是找到比n大的同时是2的次方幂的一个整数
                        workQueues = new WorkQueue[n];
                        ns = STARTED;
                    }
                }
                finally
                {
                    //初始化完成后取消锁定并且更新初始化标识
                    unlockRunState(rs, (rs & ~RSLOCK) | ns);
                }
            }
            //经过上面的赋值，r是随机数，m是数组长度-1也就是对应的mask，SQMASK是共享WQ的长度范围
            //如果在对应的位置存在工作队列，则进入工作队列的竞争流程
            else if ((q = ws[k = r & m & SQMASK]) != null)
            {
                //竞争workqueue锁成功
                if (q.qlock == 0 && U.compareAndSwapInt(q, QLOCK, 0, 1))
                {
                    ForkJoinTask<?>[] a = q.array;
                    int s = q.top;
                    boolean submitted = false; // initial submission or resizing
                    try
                    { 
                        //如果task数组不为空且未满 
                        //或者task数组扩容成功
                        if ((a != null && a.length > s + 1 - q.base) || (a = q.growArray()) != null)
                        {
                            //计算得到task在数组要放入的偏移量
                            int j = (((a.length - 1) & s) << ASHIFT) + ABASE;
                            //只放入storestore屏障，这是为了性能考虑。
                            U.putOrderedObject(a, j, task);
                            U.putOrderedInt(q, QTOP, s + 1);
                            submitted = true;
                        }
                    }
                    finally
                    {
                        U.compareAndSwapInt(q, QLOCK, 1, 0);
                    }
                    if (submitted)
                    {
                        signalWork(ws, q);
                        return;
                    }
                }
                move = true; // move on failure
            }
            //如果队列不存在，并且整个Pool的锁尚未被持有。则尝试初始化一个工作队列
            else if (((rs = runState) & RSLOCK) == 0)
            { // create new queue
                //创建了一个用于共享的Workqueue
                q = new WorkQueue(this, null);
                q.hint = r;
                //k是上面流程发现的在workqueues中为null的slot下标，workQueue内的config意味着当前的workQueue的下标是工作模式。是独占，还是共享等。
                q.config = k | SHARED_QUEUE;
                q.scanState = INACTIVE;
                rs = lockRunState(); // publish index
                //rs大于0意味着线程池么有停止，剩余的条件均是判断对应的slot位置仍然是null
                if (rs > 0 && (ws = workQueues) != null && k < ws.length && ws[k] == null)
                    ws[k] = q; // else terminated
                unlockRunState(rs, rs & ~RSLOCK);
            }
            else
                move = true; // move if busy
            //这个写法看起来容易和上面混在一起。其实已经是一个新的if开头了。如果move为true，就意味着需要切换更新线程的随机值了
            if (move)
                r = ThreadLocalRandom.advanceProbe(r);
        }
    }

signalWork(WorkQueue[] ws, WorkQueue q)

尝试唤醒处于等待堆栈栈顶的工作线程来执行任务。如果没有空闲线程且整体线程数不足，则增加工作线程。

final void signalWork(WorkQueue[] ws, WorkQueue q) {
        long c; int sp, i; WorkQueue v; Thread p;
        //ctl小于0就是AC部分小于0，意味着没有足够的激活中的执行资源
        while ((c = ctl) < 0L) {                       // too few active
            //低32位为0，意味着当前没有正在等待执行的线程
            if ((sp = (int)c) == 0) {                  // no idle workers
                //(c & ADD_WORKER)！=0 推断出TC小于0，意味着整体执行资源不足。
                if ((c & ADD_WORKER) != 0L)            // too few workers
                    tryAddWorker(c);
                break;
            }
            if (ws == null)                            // unstarted/terminated
                break;
            //如果wq数组的长度小于等待线程持有的wq的下标，意味着pool已经终止了
            if (ws.length <= (i = sp & SMASK))         // terminated
                break;
            //如果等待线程持有的wq的下标在Pool的WQ数组中已经为空，意味着Pool开始执行终止流程了。
            if ((v = ws[i]) == null)                   // terminating
                break;
            //生成等待线程的下一个scanState的值。增加SS_SEQ的值为是为了保证每一次scanState都在变化，用于版本控制。避免CAS中的ABA问题。
            int vs = (sp + SS_SEQ) & ~INACTIVE;        // next scanState
            //d==0，意味着到这一步，之前的处于等待的worker的状态没有变化
            int d = sp - v.scanState;                  // screen CAS
            //只有先成功更新ctl才能更新自身的scanState，每个地方都遵循这个顺序
            //增加AC的数值，同时将SP部分恢复到上一个等待线程的数据
            long nc = (UC_MASK & (c + AC_UNIT)) | (SP_MASK & v.stackPred);
            if (d == 0 && U.compareAndSwapLong(this, CTL, c, nc)) {
                v.scanState = vs;                      // activate v
                //如果该worker线程处于休眠状态，唤醒。
                if ((p = v.parker) != null)
                    U.unpark(p);
                break;
            }
            //任务是放在q这个WQ里面的。如果q还有任务，意味着还可以继续唤醒等待线程，因为存在可以偷取的任务。否则的话，可能没有任务可以偷取了，直接退出循环
            if (q != null && q.base == q.top)          // no more work
                break;
        }
    }

lockRunState

获得运行状态的锁。

private int lockRunState()
    {
        int rs;
        return (
        (
        //如果当前运行状态的锁被其他线程得到
        ((rs = runState) & RSLOCK) != 0 
        //或者运行状态没有锁但是竞争锁失败
                || !U.compareAndSwapInt(this, RUNSTATE, rs, rs |= RSLOCK)//
        ) ? 
        //等待直到获取锁
                awaitRunStateLock()
                //之前的步骤成功得到锁，直接返回结果
                : rs
        );
    }

awaitRunStateLock

尝试竞争运行锁，直到成功

private int awaitRunStateLock()
    {
        Object lock;
        boolean wasInterrupted = false;
        for (int spins = SPINS, r = 0, rs, ns;;)
        {
            //在运行锁没有被其他线程得到的情况尝试cas竞争锁
            if (((rs = runState) & RSLOCK) == 0)
            {
                if (U.compareAndSwapInt(this, RUNSTATE, rs, ns = rs | RSLOCK))
                {
                    if (wasInterrupted)
                    {
                        try
                        {
                            Thread.currentThread().interrupt();
                        }
                        catch (SecurityException ignore)
                        {
                        }
                    }
                    return ns;
                }
            }
            //将r初始化为随机数种子
            else if (r == 0)
                r = ThreadLocalRandom.nextSecondarySeed();
            //如果spin大于0对r进行随机数演算。
            else if (spins > 0)
            {
                r ^= r << 6;
                r ^= r >>> 21;
                r ^= r << 7; // xorshift
                //根据随机数结果对spin进行递减操作
                if (r >= 0)
                    --spins;
            }
            //如果在上面的步骤都完成的情况下，并且forkjoinpool还没有初始化意味着较高的竞争，主动让出cpu
            else if ((rs & STARTED) == 0 || (lock = stealCounter) == null)
                Thread.yield(); // initialization race
            //如果走到这里意味着线程发现线程池已经初始化了，但是本线程竞争运行锁失败，此时在运行字段上cas放入需要通知的标识。
            else if (U.compareAndSwapInt(this, RUNSTATE, rs, rs | RSIGNAL))
            {
                //如果成功，就以stealCounter为对象锁执行等待。
                synchronized (lock)
                {
                    //如果通知标识存在，则线程阻塞等待其他线程完成工作
                    if ((runState & RSIGNAL) != 0)
                    {
                        try
                        {
                            lock.wait();
                        }
                        catch (InterruptedException ie)
                        {
                            if (!(Thread.currentThread() instanceof ForkJoinWorkerThread))
                                wasInterrupted = true;
                        }
                    }
                    //否则唤醒所有在这个对象锁上执行等待的线程
                    else
                        lock.notifyAll();
                }
            }
        }
    }

unlockRunState

解锁运行锁。如果发现通知标识，则复位该标识并且进行线程唤醒

private void unlockRunState(int oldRunState, int newRunState)
    {
        //如果cas失败意味着在该cas之前或者同时，运行锁被设置了通知标识
        if (!U.compareAndSwapInt(this, RUNSTATE, oldRunState, newRunState))
        {
            Object lock = stealCounter;
            //直接清除运行锁上的通知标识
            //这里不需要通过cas。因为当runState被更新后，其他线程想要放上通知标识是通过cas完成，而由于其他线程的预期值都是锁定的状态，与新值不符，都会失败。
            //所以这里直接更新是ok的
            runState = newRunState; // clears RSIGNAL bit
            //更新完成之后，由于需要获得lock的sync锁，因此无论何种时序，都可以处于等待状态的线程恢复过来。
            if (lock != null)
                synchronized (lock)
                {
                    lock.notifyAll();
                }
        }
    }

tryAddWorker(long c)

不断尝试增加一个worker线程直到以下情况的发生
+ 当前Pool正在尝试终止
+ 增加成功
+ 当前激活worker数量超过配置的并发数并且没有处于等待的worker线程

 private void tryAddWorker(long c) {
        boolean add = false;
        do {
            //增加一个执行资源数。得到的计算结果就是nc。
            long nc = ((AC_MASK & (c + AC_UNIT)) |
                       (TC_MASK & (c + TC_UNIT)));
            if (ctl == c) {
                int rs, stop;                 // check if terminating
                if ((stop = (rs = lockRunState()) & STOP) == 0)
                    add = U.compareAndSwapLong(this, CTL, c, nc);
                unlockRunState(rs, rs & ~RSLOCK);
                if (stop != 0)
                    break;
                //如果ctl更新成功，则确实的新增一个worker线程
                if (add) {
                    //如果增加worker的意图成功，则进入创建环节。
                    createWorker();
                    break;
                }
            }
        //如果激活线程数到达或者超过配置的并发数，并且当前不存在处于等待的worker线程。则退出循环。
        } while (((c = ctl) & ADD_WORKER) != 0L && (int)c == 0);
    }

createWorker()

创建一个worker线程。创建成功则返回true。创建失败则在pool中取消注册该worker，并且返回false

 private boolean createWorker() {
        ForkJoinWorkerThreadFactory fac = factory;
        Throwable ex = null;
        ForkJoinWorkerThread wt = null;
        try {
            if (fac != null && (wt = fac.newThread(this)) != null) {
                wt.start();
                return true;
            }
        } catch (Throwable rex) {
            ex = rex;
        }
        deregisterWorker(wt, ex);
        return false;
    }

registerWorker(ForkJoinWorkerThread wt)

为新生成的ForkJoinWorkerThread创建一个私有的非共享的WorkQueue。并且将该WQ放入到Pool的WQ数组中。该WQ的模式即为Pool的工作模式。

final WorkQueue registerWorker(ForkJoinWorkerThread wt) {
        UncaughtExceptionHandler handler;
        wt.setDaemon(true);                           // configure thread
        if ((handler = ueh) != null)
            wt.setUncaughtExceptionHandler(handler);
        WorkQueue w = new WorkQueue(this, wt);
        int i = 0;                                    // assign a pool index
        int mode = config & MODE_MASK;
        int rs = lockRunState();
        try {
            WorkQueue[] ws; int n;                    // skip if no array
            if ((ws = workQueues) != null && (n = ws.length) > 0) {
                //得到一个初始化的数值。该数值的含义在尽可能避免碰撞
                int s = indexSeed += SEED_INCREMENT;  // unlikely to collide
                int m = n - 1;
                //得到一个奇数下标
                i = ((s << 1) | 1) & m;               // odd-numbered indices
                //如果发生了碰撞，则不断移动下标。下标的移动步长为数组长度的一半。
                if (ws[i] != null) {                  // collision
                    int probes = 0;
                    //step是大于或等于WQ数组长度的一半的最小偶数。
                    int step = (n <= 4) ? 2 : ((n >>> 1) & EVENMASK) + 2;
                    while (ws[i = (i + step) & m] != null) {
                        //不断尝试移动下标，如果接近2次完整的遍历都未曾发现空的槽位，则2倍扩容WQ数组
                        if (++probes >= n) {
                            workQueues = ws = Arrays.copyOf(ws, n <<= 1);
                            m = n - 1;
                            probes = 0;
                        }
                    }
                }
                //workQueue数组的大小受限于TC。也就是受限于并发数限制。并发数则限制在0x7fff内。
                w.hint = s;                           // use as random seed
                //当前workqueue的模式是workpool的模式。且非共享。
                w.config = i | mode;
                w.scanState = i;                      // publication fence
                ws[i] = w;
            }
        } finally {
            unlockRunState(rs, rs & ~RSLOCK);
        }
        wt.setName(workerNamePrefix.concat(Integer.toString(i >>> 1)));
        return w;
    }

runWorker(WorkQueue w)

一个ForkjoinWorkerThread的主要循环。扫描WQ数组，偷取任务进行执行（有可能偷取的任务就是自身的任务）。没有任务时则进行等待。

runWorker(WorkQueue w) {
        //对WQ执行初始化和适当的扩容。
        w.growArray();                   // allocate queue
        int seed = w.hint;               // initially holds randomization hint
        int r = (seed == 0) ? 1 : seed;  // avoid 0 for xorShift
        for (ForkJoinTask<?> t;;) {
            if ((t = scan(w, r)) != null)
                w.runTask(t);
            else if (!awaitWork(w, r))
                break;
            r ^= r << 13; r ^= r >>> 17; r ^= r << 5; // xorshift
        }
    }

scan(WorkQueue w, int r)

扫描WQ数组并且尝试偷取一个任务。扫描起始于一个随机位置r，不断向前递增。如果一轮扫描中没有发现任何可以偷取的偷取，则尝试将自身设置为非激活状态，并且ctl的ac值-1。 如果再次扫描一轮，仍然处于非激活状态并且整个WQ数组状态稳定。则退出扫描并且返回null。
而如果扫描时发现了可以偷取的任务，自身处于激活状态则尝试偷取，否则尝试激活一个worker（可能是自己也可能是别人）。

private ForkJoinTask<?> scan(WorkQueue w, int r) {
        WorkQueue[] ws; int m;
        if ((ws = workQueues) != null && (m = ws.length - 1) > 0 && w != null) {
            int ss = w.scanState;                     // initially non-negative
            //计算初始随机位置origin
            for (int origin = r & m, k = origin, oldSum = 0, checkSum = 0;;) {
                WorkQueue q; ForkJoinTask<?>[] a; ForkJoinTask<?> t;
                int b, n; long c;
                //从Pool的WQ数组中挑选一个随机槽位上的WQ。该WQ不为空
                if ((q = ws[k]) != null) {
                    //挑选的WQ对象，存在着可以偷取的任务（top-base大于0时意味着存在任务）
                    if ((n = (b = q.base) - q.top) < 0 &&
                        (a = q.array) != null) {      // non-empty
                        long i = (((a.length - 1) & b) << ASHIFT) + ABASE;
                        //如果发现一个可以偷取的任务
                        if ((t = ((ForkJoinTask<?>)
                                  U.getObjectVolatile(a, i))) != null &&
                            q.base == b) {
                            //读取到该WQ队列头部的task并且此时该队列的base尚未变化
                            //如果当前worker处于激活状态则尝试偷取
                            if (ss >= 0) {
                                //将要偷取的task所在槽位cas方式置空。成功后则增加该WQ的base值并且返回偷取的任务。
                                if (U.compareAndSwapObject(a, i, t, null)) {
                                    q.base = b + 1;
                                    //n<-1，意味着当前WQ内的待提取任务超过1个。因此可以尝试唤醒该worker线程
                                    if (n < -1)       
                                        signalWork(ws, q);
                                    return t;
                                }
                            }
                       //这个条件意味着本次扫描是当前worker设置为非激活状态成功后的第一轮扫描。此时发现了可以偷取的任务，则尝试激活一个worker（可以是自己或者是别人）
                            else if (oldSum == 0 &&   // try to activate
                                     w.scanState < 0)
                                tryRelease(c = ctl, ws[m & (int)c], AC_UNIT);
                        }
                        //如果当前是非激活状态，则尝试刷新。有可能上面的操作中激活的是自身
                        if (ss < 0)                   // refresh
                            ss = w.scanState;
                        //如果走到这里意味着发现了可以偷取的任务，但是没有偷取到。因此移动到一个新的随机初始值。并重新开始整个扫描流程
                        r ^= r << 1; r ^= r >>> 3; r ^= r << 10;
                        origin = k = r & m;           // move and rescan
                        oldSum = checkSum = 0;
                        continue;
                    }
                    //选中的WQ没有可以偷取任务时增加checkSum
                    checkSum += b;
                }
                //让k+=1.如果循环完一个数组长度，则尝试执行一些操作
                if ((k = (k + 1) & m) == origin) {    
                    //ss>=0成立。意味着当前worker处于激活状态。
                    //ss == (ss = w.scanState)意味着一轮扫描结束，worker的状态没有发生改变并且处于非激活状态
                    //oldSum == (oldSum = checkSum) 意味着一轮扫描结束，整个WQ数组的内容没有发生改变
                    //后两个条件同时为真，意味着连续的两轮扫描中，整个WQ数组状态稳定。
                    if ((ss >= 0 || (ss == (ss = w.scanState))) &&
                        oldSum == (oldSum = checkSum)) {
                        //如果已经是非激活状态了，则直接退出
                        if (ss < 0 || w.qlock < 0)    // already inactive
                            break;
                        //将状态设置为负数，也就是非激活状态。
                        int ns = ss | INACTIVE;       // try to inactivate
                        //设定新的ctl值。首先是AC部分减去1.然后是SP部分更新为当前worker的内容。
                        long nc = ((SP_MASK & ns) |
                                   (UC_MASK & ((c = ctl) - AC_UNIT)));
                        //当前的worker保留了上一个留存于SP部分的worker信息。
                        w.stackPred = (int)c;         // hold prev stack top
                        //非volatile写的方式更新了scanState的值。因为下面还有cas操作，这里这样的更新操作可以节省性能。
                        U.putInt(w, QSCANSTATE, ns);
                        //如果cas方式更新ctl成功，则将ss设置为更新后的状态值，否则将scanState回退到更新前的状态值也就是ss的值。
                        if (U.compareAndSwapLong(this, CTL, c, nc))
                            ss = ns;
                        else
                            w.scanState = ss;         // back out
                    }
                    checkSum = 0;
                }
            }
        }
        return null;
    }

awaitWork(WorkQueue w, int r)

进入该方法的WQ，本身已经是非激活状态，并且将自身的scanState更新到了ctl中。
如果当前worker不是最后一个等待worker，则进入休眠状态等待被其他线程唤醒。
如果当前worker是最后一个worker，则区分不同情况处理。

• 总体并发数超过了最大并发数+2，则尝试立刻cas方式更新ctl。成功的话，则终止当前worker。
• 如果失败或者并发数尚未超过最大并发数+2，则尝试休眠一段时间。休眠完成后如果Pool的状态不变，则尝试终止该worker。

以下几种情况该方法返回false表达该worker需要终止

• worker的qlock被设置为负数。
• 该worker是等待堆栈中最后一个worker，并且更新ctl成功

/**
* 返回false意味着该worker线程需要终止了。也就是需要执行deregisterWorker方法
*/
private boolean awaitWork(WorkQueue w, int r) {
        if (w == null || w.qlock < 0)                 // w is terminating
            return false;
        for (int pred = w.stackPred, spins = SPINS, ss;;) {
            //能够进入该方法的worker都是处于非激活状态。如果发现再次被激活，则需要退出循环。
            if ((ss = w.scanState) >= 0)
                break;
            else if (spins > 0) {
                //类xorshift运算，得到下一个随机数
                r ^= r << 6; r ^= r >>> 21; r ^= r << 7;
                if (r >= 0 && --spins == 0) {         // randomize spins
                    WorkQueue v; WorkQueue[] ws; int s, j; AtomicLong sc;
                    //检查worker堆栈中的上一个worker是否已经进入休眠状态。如果未曾进入休眠状态的话，意味着该worker也有可能不进入休眠。因此这里会通过设置spins的值继续进行自旋。
                    if (pred != 0 && (ws = workQueues) != null &&
                        (j = pred & SMASK) < ws.length &&
                        (v = ws[j]) != null &&        // see if pred parking
                        (v.parker == null || v.scanState >= 0))
                        spins = SPINS;                // continue spinning
                }
            }
            //worker线程被终止，则直接返回
            else if (w.qlock < 0)                     // recheck after spins
                return false;
            else if (!Thread.interrupted()) {
                long c, prevctl, parkTime, deadline;
                int ac = (int)((c = ctl) >> AC_SHIFT) + (config & SMASK);
                //计算结果ac的值含义是当前激活的worker数量
                if ((ac <= 0 && tryTerminate(false, false)) ||
                    (runState & STOP) != 0)           // pool terminating
                    return false;
                //如果总的资源数已经超过了并发需求并且该worker是最后一个等待worker，则尝试终止该worker。如果通过ctl替换的方式无法终止，则通过超时的方式进行终止
                if (ac <= 0 && ss == (int)c) {        // is last waiter
                    //计算得到前一个ctl的值
                    prevctl = (UC_MASK & (c + AC_UNIT)) | (SP_MASK & pred);
                    //得到tc部分的值
                    int t = (short)(c >>> TC_SHIFT);  // shrink excess spares
                    //如果总资源数超过并发数+2，则意味着worker数量太多，终止本worker。
                    //如果ctl更新成功，则直接返回false
                    if (t > 2 && U.compareAndSwapLong(this, CTL, c, prevctl))
                        return false;                
                    //如果ctl更新失败则休眠一段时间后再次尝试
                    parkTime = IDLE_TIMEOUT * ((t >= 0) ? 1 : 1 - t);
                    deadline = System.nanoTime() + parkTime - TIMEOUT_SLOP;
                }
                else
                    prevctl = parkTime = deadline = 0L;
                Thread wt = Thread.currentThread();
                U.putObject(wt, PARKBLOCKER, this);   // emulate LockSupport
                w.parker = wt;
                if (w.scanState < 0 && ctl == c)      // recheck before park
                    U.park(false, parkTime);
                U.putOrderedObject(w, QPARKER, null);
                U.putObject(wt, PARKBLOCKER, null);
                //被其他线程唤醒了
                if (w.scanState >= 0)
                    break;
                //休眠时间到达，并且ctl更新成功，则返回false。需要停止该worker
                if (parkTime != 0L && ctl == c &&
                    deadline - System.nanoTime() <= 0L &&
                    U.compareAndSwapLong(this, CTL, c, prevctl))
                    return false;                     // shrink pool
            }
        }
        return true;
    }

awaitJoin(WorkQueue w, ForkJoinTask task, long deadline)

等待给定的任务直到该任务完成。在等待的过程中，等待的worker并非直接进入阻塞状态。而是采取几个策略推进任务的整体执行

1. 如果给定任务是本地任务，则尝试直接获取该任务，并且立刻执行
2. 如果给定任务是其他worker的任务，则尝试偷取该worker的任务执行，以加快该worker完成给定任务的可能性。
3. 没有任务可以执行则尝试创建一个代替worker或者恢复一个处于休眠状态的worker。如果创建或者恢复成功，则自身陷入阻塞。

    final int awaitJoin(WorkQueue w, ForkJoinTask<?> task, long deadline)
    {
        int s = 0;
        if (task != null && w != null)
        {
            ForkJoinTask<?> prevJoin = w.currentJoin;
            U.putOrderedObject(w, QCURRENTJOIN, task);
            CountedCompleter<?> cc = (task instanceof CountedCompleter) ? (CountedCompleter<?>) task : null;
            for (;;)
            {
                if ((s = task.status) < 0)
                    break;
                if (cc != null)
                    helpComplete(w, cc, 0);
                //如果没有本地任务，或者无法执行给定任务。则尝试将偷取了给定任务的worker中的本地任务偷取来自己执行。
                else if (w.base == w.top || w.tryRemoveAndExec(task))
                    helpStealer(w, task);
                if ((s = task.status) < 0)
                    break;
                long ms, ns;
                if (deadline == 0L)
                    ms = 0L;
                //如果最后期限已过，则直接返回
                else if ((ns = deadline - System.nanoTime()) <= 0L)
                    break;
                //等待时间最少是1ms。
                else if ((ms = TimeUnit.NANOSECONDS.toMillis(ns)) <= 0L)
                    ms = 1L;
                //如果补偿成功，则本worker阻塞一段时间。
                if (tryCompensate(w))
                {
                    task.internalWait(ms);
                    U.getAndAddLong(this, CTL, AC_UNIT);
                }
            }
            U.putOrderedObject(w, QCURRENTJOIN, prevJoin);
        }
        return s;
    }

helpStealer(WorkQueue w, ForkJoinTask task)

只有当worker w没有本地任务的时候才会执行这个方法。
worker v 尝试从偷取了task的work stealer的本地任务中偷取一个任务来执行。

private void helpStealer(WorkQueue w, ForkJoinTask<?> task) {
        WorkQueue[] ws = workQueues;
        int oldSum = 0, checkSum, m;
        if (ws != null && (m = ws.length - 1) >= 0 && w != null &&
            task != null) {
            do {                                       // restart point
                checkSum = 0;                          // for stability check
                ForkJoinTask<?> subtask;
                WorkQueue j = w, v;                    // v is subtask stealer
                descent: for (subtask = task; subtask.status >= 0; ) {
                    //使用随机数初始化一个奇数值。然后遍历WQ数组。只遍历其中的worker部分也就是奇数部分。
                    for (int h = j.hint | 1, k = 0, i; ; k += 2) {
                        //没有可以偷取任务的worker，退出整个大的循环
                        if (k > m)                   
                            break descent;
                        if ((v = ws[i = (h + k) & m]) != null) {
                            //发现了一个worker v，该worker偷取的任务就是给定的任务task。
                            if (v.currentSteal == subtask) {
                                j.hint = i;
                                break;
                            }
                            checkSum += v.base;
                        }
                    }
                    for (;;) {                       
                        ForkJoinTask<?>[] a; int b;
                        checkSum += (b = v.base);
                        ForkJoinTask<?> next = v.currentJoin;
                        //1. 如果给定的任务已经完成
                        //2. 如果当前的worker v没有在等待给定任务
                        //3. 如果偷取给定任务的worker又偷取了别的任务
                        //意味着之前的数据已经过期，放弃
                        if (subtask.status < 0 || j.currentJoin != subtask ||
                            v.currentSteal != subtask) // stale
                            break descent;
                        //如果偷取任务的worker没有本地任务
                        if (b - v.top >= 0 || (a = v.array) == null) {
                            //偷取任务的worker没有在join的任务要完成
                            if ((subtask = next) == null)
                                break descent;
                            j = v;
                            break;
                        }
                        int i = (((a.length - 1) & b) << ASHIFT) + ABASE;
                        ForkJoinTask<?> t = ((ForkJoinTask<?>)
                                             U.getObjectVolatile(a, i));
                        if (v.base == b) {
                            if (t == null)             // stale
                                break descent;
                            //从偷取给定任务的worker处偷取了一个任务来执行
                            if (U.compareAndSwapObject(a, i, t, null)) {
                                v.base = b + 1;
                                ForkJoinTask<?> ps = w.currentSteal;
                                int top = w.top;
                                do {
                                    U.putOrderedObject(w, QCURRENTSTEAL, t);
                                    t.doExec();       
                                //如果本worker偷取的任务执行的过程中，分裂了。那么继续获取分裂后的本地任务执行，直到本地任务都被完成
                                } while (task.status >= 0 &&
                                         w.top != top &&
                                         (t = w.pop()) != null);
                                U.putOrderedObject(w, QCURRENTSTEAL, ps);
                                //不相等意味着本地任务在分裂的时候，被其他的线程也偷取了一部分
                                if (w.base != w.top)
                                    return;            // can't further help
                            }
                        }
                    }
                }
                //循环两次，整个Pool的状态都是稳定的，没有新增任务。
            } while (task.status >= 0 && oldSum != (oldSum = checkSum));
        }
    }

helpComplete(WorkQueue w, CountedCompleter task,int maxTasks)

final int helpComplete(WorkQueue w, CountedCompleter<?> task,
                           int maxTasks) {
        WorkQueue[] ws; int s = 0, m;
        if ((ws = workQueues) != null && (m = ws.length - 1) >= 0 &&
            task != null && w != null) {
            int mode = w.config;                 // for popCC
            int r = w.hint ^ w.top;              // arbitrary seed for origin
            int origin = r & m;                  // first queue to scan
            int h = 1;                           // 1:ran, >1:contended, <0:hash
            for (int k = origin, oldSum = 0, checkSum = 0;;) {
                CountedCompleter<?> p; WorkQueue q;
                if ((s = task.status) < 0)
                    break;
                if (h == 1 && (p = w.popCC(task, mode)) != null) {
                    p.doExec();                  // run local task
                    if (maxTasks != 0 && --maxTasks == 0)
                        break;
                    origin = k;                  // reset
                    oldSum = checkSum = 0;
                }
                else {                           // poll other queues
                    if ((q = ws[k]) == null)
                        h = 0;
                    else if ((h = q.pollAndExecCC(task)) < 0)
                        checkSum += h;
                    if (h > 0) {
                        if (h == 1 && maxTasks != 0 && --maxTasks == 0)
                            break;
                        r ^= r << 13; r ^= r >>> 17; r ^= r << 5; // xorshift
                        origin = k = r & m;      // move and restart
                        oldSum = checkSum = 0;
                    }
                    else if ((k = (k + 1) & m) == origin) {
                        if (oldSum == (oldSum = checkSum))
                            break;
                        checkSum = 0;
                    }
                }
            }
        }
        return s;
    }

tryCompensate(WorkQueue w)

有意向阻塞worker v一段时间。因此这里尝试做出某种补偿策略。返回true意味着本worker可以阻塞

• 如果存在休眠的worker并且唤醒该worker成功则返回true
• 如果没有休眠的worker
  
  • 如果ctl发生了变化或者有worker正在处理任务则返回false
  • 如果总资源数大于并发配置，超过1个激活worker，本worker没有本地任务了尝试减少AC的值。减少成功返回true
  • 如果总资源数大于系统并发限制上限。则抛出异常。静止该worker继续创建新的worker
  • 如果上面情况都没有发生，则尝试增加TC的值。增加成功则创建一个worker并且返回true

    private boolean tryCompensate(WorkQueue w)
    {
        boolean canBlock;
        WorkQueue[] ws;
        long c;
        int m, pc, sp;
        if (w == null || w.qlock < 0 || // caller terminating
                (ws = workQueues) == null || (m = ws.length - 1) <= 0 || (pc = config & SMASK) == 0) 
            canBlock = false;
        //如果存在休眠的worker则尝试唤醒。如果尝试唤醒成功，则本worker可以阻塞。
        //外部代码中，当前worker如果从休眠中恢复，会增加AC的值。而在这里，因为唤醒成功的情况下本worker会阻塞。所以这里就不增加AC的值了。一增一减就归零了。
        else if ((sp = (int) (c = ctl)) != 0) 
            canBlock = tryRelease(c, ws[sp & m], 0L);
        //没有休眠中的worker
        else
        {
            int ac = (int) (c >> AC_SHIFT) + pc;
            int tc = (short) (c >> TC_SHIFT) + pc;
            int nbusy = 0; // validate saturation
            //检查所有的奇数位置，也就是检查所有的worker的状态。
            //如果worker处于扫描任务的状态，标记为nbusy。遍历数组直到发现一个处于工作状态的线程
            //如果都没有过于工作状态的worker，那么遍历结束时nbusy应该是总资源数的2倍。（2倍是因为其选择奇数的算法造成）
            for (int i = 0; i <= m; ++i)
            {
                WorkQueue v;
                if ((v = ws[((i << 1) | 1) & m]) != null)
                {
                    if ((v.scanState & SCANNING) != 0)
                        break;
                    ++nbusy;
                }
            }
            //有worker在处理任务或者ctl发生了变化。
            if (nbusy != (tc << 1) || ctl != c)
                canBlock = false; // unstable or stale
            //如果总资源大于配置的并发，激活worker数量大于1，本worker也没有任务
            //则尝试在不唤醒休眠线程的情况下，单纯让自身陷入等待状态。
            //所以这里让AC-1
            else if (tc >= pc && ac > 1 && w.isEmpty())
            {
                long nc = ((AC_MASK & (c - AC_UNIT)) | (~AC_MASK & c)); // uncompensated
                canBlock = U.compareAndSwapLong(this, CTL, c, nc);
            }
            //如果总worker数量超过了最大的限定worker数量，抛出异常
            else if (tc >= MAX_CAP || (this == common && tc >= pc + commonMaxSpares))
                throw new RejectedExecutionException("Thread limit exceeded replacing blocked worker");
            //尝试增加一个worker线程。这里只是增加了TC部分的值。
            //AC部分的值在外部方法中增加
            else
            { // similar to tryAddWorker
                boolean add = false;
                int rs; // CAS within lock
                long nc = ((AC_MASK & c) | (TC_MASK & (c + TC_UNIT)));
                if (((rs = lockRunState()) & STOP) == 0)
                    add = U.compareAndSwapLong(this, CTL, c, nc);
                unlockRunState(rs, rs & ~RSLOCK);
                canBlock = add && createWorker(); // throws on exception
            }
        }
        return canBlock;
    }

deregisterWorker(ForkJoinWorkerThread wt, Throwable ex)

在Pool中取消该WQ的注册。主要内容是

• 从WQ数组中设置自身槽位为空
• 更新ctl的值，将其中AC和TC部分-1
• 如果有需要，则唤醒等待堆栈栈顶的等待worker，或者是增加一个新的执行资源

worker会退出有两种情况。

• worker运行过程中因为资源过剩而退出。此时该worker应该是处于等待堆栈最顶端的worker，并且来执行该方法前已经从等待堆栈弹出。此时退出顺序很重要，必须让等待堆栈的栈顶的先退出，逐步唤醒才能保证顺序不乱。
• 因为整个Pool都要关闭了。所有的worker都会被唤醒然后执行退出流程。此时因为worker是被外在因素全部唤醒，不再需要存在顺序了。

final void deregisterWorker(ForkJoinWorkerThread wt, Throwable ex) {
        WorkQueue w = null;
        //将当前的workqueue从WQ数组中删除
        if (wt != null && (w = wt.workQueue) != null) {
            WorkQueue[] ws;                           // remove index from array
            int idx = w.config & SMASK;
            int rs = lockRunState();
            if ((ws = workQueues) != null && ws.length > idx && ws[idx] == w)
                ws[idx] = null;
            unlockRunState(rs, rs & ~RSLOCK);
        }
        //更新ctl。删除一个资源数。主要就是在AC和TC部分-1
        long c;                                      
        do {} while (!U.compareAndSwapLong
                     (this, CTL, c = ctl, ((AC_MASK & (c - AC_UNIT)) |
                                           (TC_MASK & (c - TC_UNIT)) |
                                           (SP_MASK & c))));
        if (w != null) {
            //设置qlock为-1，表明该worker已经终止。
            w.qlock = -1;                             
            w.transferStealCount(this);
            w.cancelAll();                            // cancel remaining tasks
        }
        //本worker线程退出，因此确认下是否需要唤醒休眠线程，或者是在资源不足的情况下增加一个worker线程。
        for (;;) {                                    // possibly replace
            WorkQueue[] ws; int m, sp;
            //如果当前pool已经处于终止过程中，说明不需要新增worker资源，退出。
            if (tryTerminate(false, false) || w == null || w.array == null ||
                (runState & STOP) != 0 || (ws = workQueues) == null ||
                (m = ws.length - 1) < 0)              
                break;
            //如果存在处于休眠等待的worker线程，则直接唤醒
            if ((sp = (int)(c = ctl)) != 0) {     
                if (tryRelease(c, ws[sp & m], AC_UNIT))
                    break;
            }
            //如果是发生了异常导致了worker线程的终止。并且当前激活的worker资源少于预定的并发数，则尝试增加一个worker线程用于弥补本worker线程退出造成的影响。
            else if (ex != null && (c & ADD_WORKER) != 0L) {
                tryAddWorker(c);                     
                break;
            }
            else                                      
                break;
        }
        if (ex == null)                               // help clean on way out
            ForkJoinTask.helpExpungeStaleExceptions();
        else                                          // rethrow
            ForkJoinTask.rethrow(ex);
    }