談?wù)刦ork/join實(shí)現(xiàn)原理

import java.util.Arrays;import java.util.Random;import java.util.concurrent.ExecutionException;import java.util.concurrent.ForkJoinPool;import java.util.concurrent.ForkJoinTask;import java.util.concurrent.RecursiveTask;
/** * Fork/join框架測(cè)試 */public class TestForkJoinFramework {
    public static void main(String[] args) {        long beginTime = System.currentTimeMillis();        ForkJoinPool pool = new ForkJoinPool();        int mockArrLen = 1000_0000;        int[] arr = new int[mockArrLen];        Random r = new Random();        for (int index = 1; index <= mockArrLen; index++) {            arr[index - 1] = r.nextInt(1000_0000);        }        FJOrderTask task = new FJOrderTask(arr);        ForkJoinTask<int[]> taskResult = pool.submit(task);        try {            // 等待結(jié)果完成            taskResult.get();        } catch (InterruptedException | ExecutionException e) {            e.printStackTrace();        }        long endTime = System.currentTimeMillis();        System.out.println("耗時(shí)=" + (endTime - beginTime));    }
    /**     * 單個(gè)排序的子任務(wù)     */    private static class FJOrderTask extends RecursiveTask<int[]> {
        /**         * 當(dāng)前排序的數(shù)組值         */        private final int[] source;
        public FJOrderTask(int[] source) {            this.source = source;        }
        /**         * 真正的業(yè)務(wù)計(jì)算邏輯         *         * @see java.util.concurrent.RecursiveTask#compute()         */        @Override        protected int[] compute() {            int sourceLen = source.length;            // 如果條件成立，說(shuō)明任務(wù)中要進(jìn)行排序的集合還不夠小            System.out.println(Thread.currentThread());            if (sourceLen > 2) {                int midIndex = sourceLen / 2;                // 拆分成兩個(gè)子任務(wù), 0 -> mid - 1, mid -> len                FJOrderTask task1 = new FJOrderTask(                        Arrays.copyOf(source, midIndex));                task1.fork();                FJOrderTask task2 = new FJOrderTask(                        Arrays.copyOfRange(source, midIndex, sourceLen));                task2.fork();                // 將兩個(gè)有序的數(shù)組，合并成一個(gè)有序的數(shù)組                int[] result1 = task1.join();                int[] result2 = task2.join();                return insertMerge(result1, result2);            }            // 否則說(shuō)明集合中只有一個(gè)或者兩個(gè)元素，可以進(jìn)行這兩個(gè)元素的比較排序了            else {                // 如果條件成立，說(shuō)明數(shù)組中只有一個(gè)元素，或者是數(shù)組中的元素都已經(jīng)排列好位置了                if (sourceLen == 1                        || source[0] <= source[1]) {                    return source;                } else {                    int[] orderedArr = new int[sourceLen];                    orderedArr[0] = source[1];                    orderedArr[1] = source[0];                    return orderedArr;                }            }        }
        /**         * 使用插入排序，將兩個(gè)有序數(shù)組合并起來(lái)         *         * @param arr1 有序數(shù)組1         * @param arr2 有序數(shù)組2         * @return 合并后的有序數(shù)組         */        private int[] insertMerge(int[] arr1, int[] arr2) {            int[] result = new int[arr1.length + arr2.length];            int arr1Len = arr1.length;            int arr2Len = arr2.length;            int destLen = result.length;            // 簡(jiǎn)單插入排序            for (int i = 0, array1Index = 0, array2Index = 0; i < destLen; i++) {                int value1 = array1Index >= arr1Len                        ? Integer.MAX_VALUE : arr1[array1Index];                int value2 = array2Index >= arr2Len                        ? Integer.MAX_VALUE : arr2[array2Index];                if (value1 < value2) {                    array1Index++;                    result[i] = value1;                }                else {                    array2Index++;                    result[i] = value2;                }            }            return result;        }
    }}

// java.util.concurrent.ForkJoinPool#ForkJoinPool()    /**     * Creates a {@code ForkJoinPool} with parallelism equal to {@link     * java.lang.Runtime#availableProcessors}, using the {@linkplain     * #defaultForkJoinWorkerThreadFactory default thread factory},     * no UncaughtExceptionHandler, and non-async LIFO processing mode.     *     * @throws SecurityException if a security manager exists and     *         the caller is not permitted to modify threads     *         because it does not hold {@link     *         java.lang.RuntimePermission}{@code ("modifyThread")}     */    public ForkJoinPool() {        // 并行度默認(rèn)是cpu的核數(shù)        this(Math.min(MAX_CAP, Runtime.getRuntime().availableProcessors()),             defaultForkJoinWorkerThreadFactory, null, false);    }    /**     * Creates a {@code ForkJoinPool} with the given parameters.     *     * @param parallelism the parallelism level. For default value,     * use {@link java.lang.Runtime#availableProcessors}.     * @param factory the factory for creating new threads. For default value,     * use {@link #defaultForkJoinWorkerThreadFactory}.     * @param handler the handler for internal worker threads that     * terminate due to unrecoverable errors encountered while executing     * tasks. For default value, use {@code null}.     * @param asyncMode if true,     * establishes local first-in-first-out scheduling mode for forked     * tasks that are never joined. This mode may be more appropriate     * than default locally stack-based mode in applications in which     * worker threads only process event-style asynchronous tasks.     * For default value, use {@code false}.     * @throws IllegalArgumentException if parallelism less than or     *         equal to zero, or greater than implementation limit     * @throws NullPointerException if the factory is null     * @throws SecurityException if a security manager exists and     *         the caller is not permitted to modify threads     *         because it does not hold {@link     *         java.lang.RuntimePermission}{@code ("modifyThread")}     */    public ForkJoinPool(int parallelism,                        ForkJoinWorkerThreadFactory factory,                        UncaughtExceptionHandler handler,                        boolean asyncMode) {        this(checkParallelism(parallelism),             checkFactory(factory),             handler,             // FIFO_QUEUE = 1 << 16, LIFO_QUEUE = 0             asyncMode ? FIFO_QUEUE : LIFO_QUEUE,             "ForkJoinPool-" + nextPoolId() + "-worker-");        checkPermission();    }    /**     * Creates a {@code ForkJoinPool} with the given parameters, without     * any security checks or parameter validation.  Invoked directly by     * makeCommonPool.     */    private ForkJoinPool(int parallelism,                         ForkJoinWorkerThreadFactory factory,                         UncaughtExceptionHandler handler,                         int mode,                         String workerNamePrefix) {        this.workerNamePrefix = workerNamePrefix;        this.factory = factory;        this.ueh = handler;        this.config = (parallelism & SMASK) | mode;        long np = (long)(-parallelism); // offset ctl counts        this.ctl = ((np << AC_SHIFT) & AC_MASK) | ((np << TC_SHIFT) & TC_MASK);    }

// java.util.concurrent.ForkJoinPool#submit    /**     * Submits a ForkJoinTask for execution.     *     * @param task the task to submit     * @param <T> the type of the task's result     * @return the task     * @throws NullPointerException if the task is null     * @throws RejectedExecutionException if the task cannot be     *         scheduled for execution     */    public <T> ForkJoinTask<T> submit(ForkJoinTask<T> task) {        if (task == null)            throw new NullPointerException();        // submit主要是向pool中加入任務(wù)隊(duì)列        externalPush(task);        return task;    }    /**     * Tries to add the given task to a submission queue at     * submitter's current queue. Only the (vastly) most common path     * is directly handled in this method, while screening for need     * for externalSubmit.     *     * @param task the task. Caller must ensure non-null.     */    final void externalPush(ForkJoinTask<?> task) {        WorkQueue[] ws; WorkQueue q; int m;        int r = ThreadLocalRandom.getProbe();        int rs = runState;        // 如果線程不是第一次進(jìn)入，且獲得鎖，則直接放隊(duì)列即可        // 否則走普通加入隊(duì)列邏輯        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;                // 通過(guò) putOrderedObject 添加任務(wù)到隊(duì)列中                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);        }        // 初始化時(shí)的submit或者通用 submit        externalSubmit(task);    }
    /**     * Full version of externalPush, handling uncommon cases, as well     * as performing secondary initialization upon the first     * submission of the first task to the pool.  It also detects     * first submission by an external thread and creates a new shared     * queue if the one at index if empty or contended.     *     * @param task the task. Caller must ensure non-null.     */    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;            // 停止運(yùn)行            if ((rs = runState) < 0) {                tryTerminate(false, false);     // help terminate                throw new RejectedExecutionException();            }            // 未被初始化，先執(zhí)行初始化            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                        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;                        // 隊(duì)列數(shù)量初始化                        workQueues = new WorkQueue[n];                        ns = STARTED;                    }                } finally {                    unlockRunState(rs, (rs & ~RSLOCK) | ns);                }            }            // 當(dāng)前線程已添加過(guò)隊(duì)列            else if ((q = ws[k = r & m & SQMASK]) != null) {                // 上鎖添加到隊(duì)列中                if (q.qlock == 0 && U.compareAndSwapInt(q, QLOCK, 0, 1)) {                    ForkJoinTask<?>[] a = q.array;                    // 取出棧頂指針，向其中放入任務(wù)                    int s = q.top;                    boolean submitted = false; // initial submission or resizing                    try {                      // locked version of push                        if ((a != null && a.length > s + 1 - q.base) ||                            (a = q.growArray()) != null) {                            int j = (((a.length - 1) & s) << ASHIFT) + ABASE;                            U.putOrderedObject(a, j, task);                            U.putOrderedInt(q, QTOP, s + 1);                            submitted = true;                        }                    } finally {                        U.compareAndSwapInt(q, QLOCK, 1, 0);                    }                    // 如果隊(duì)列添加成功，則喚醒一個(gè) worker, 返回                    // 否則進(jìn)入下一次嘗試添加過(guò)程                    if (submitted) {                        signalWork(ws, q);                        return;                    }                }                move = true;                   // move on failure            }            else if (((rs = runState) & RSLOCK) == 0) { // create new queue                q = new WorkQueue(this, null);                q.hint = r;                q.config = k | SHARED_QUEUE;                q.scanState = INACTIVE;                rs = lockRunState();           // publish index                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            // 如有必要，為當(dāng)前線程生成新的標(biāo)識(shí)            if (move)                r = ThreadLocalRandom.advanceProbe(r);        }    }

// java.util.concurrent.ForkJoinPool#signalWork    /**     * Tries to create or activate a worker if too few are active.     *     * @param ws the worker array to use to find signallees     * @param q a WorkQueue --if non-null, don't retry if now empty     */    final void signalWork(WorkQueue[] ws, WorkQueue q) {        long c; int sp, i; WorkQueue v; Thread p;        while ((c = ctl) < 0L) {                       // too few active，一個(gè)標(biāo)識(shí)，分兩段使用，低位為0代表worker還可以添加            if ((sp = (int)c) == 0) {                  // no idle workers                if ((c & ADD_WORKER) != 0L)            // too few workers                    tryAddWorker(c);                break;            }            if (ws == null)                            // unstarted/terminated                break;            if (ws.length <= (i = sp & SMASK))         // terminated                break;            if ((v = ws[i]) == null)                   // terminating                break;            int vs = (sp + SS_SEQ) & ~INACTIVE;        // next scanState            int d = sp - v.scanState;                  // screen CAS            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                if ((p = v.parker) != null)                    U.unpark(p);                break;            }            if (q != null && q.base == q.top)          // no more work                break;        }    }
    /**     * Tries to add one worker, incrementing ctl counts before doing     * so, relying on createWorker to back out on failure.     *     * @param c incoming ctl value, with total count negative and no     * idle workers.  On CAS failure, c is refreshed and retried if     * this holds (otherwise, a new worker is not needed).     */    private void tryAddWorker(long c) {        boolean add = false;        do {            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;                // 添加標(biāo)識(shí)成功，再創(chuàng)建worker                if (add) {                    createWorker();                    break;                }            }        } while (((c = ctl) & ADD_WORKER) != 0L && (int)c == 0);    }
    /**     * Tries to construct and start one worker. Assumes that total     * count has already been incremented as a reservation.  Invokes     * deregisterWorker on any failure.     *     * @return true if successful     */    private boolean createWorker() {        ForkJoinWorkerThreadFactory fac = factory;        Throwable ex = null;        ForkJoinWorkerThread wt = null;        try {            // 調(diào)用線程工廠創(chuàng)建新的worker, 并立即啟動(dòng)worker            if (fac != null && (wt = fac.newThread(this)) != null) {                wt.start();                return true;            }        } catch (Throwable rex) {            ex = rex;        }        // 創(chuàng)建失敗，處理異常        deregisterWorker(wt, ex);        return false;    }    /**     * Default ForkJoinWorkerThreadFactory implementation; creates a     * new ForkJoinWorkerThread.     */    static final class DefaultForkJoinWorkerThreadFactory        implements ForkJoinWorkerThreadFactory {        public final ForkJoinWorkerThread newThread(ForkJoinPool pool) {            return new ForkJoinWorkerThread(pool);        }    }

// java.util.concurrent.ForkJoinWorkerThread#ForkJoinWorkerThread    /**     * Creates a ForkJoinWorkerThread operating in the given pool.     *     * @param pool the pool this thread works in     * @throws NullPointerException if pool is null     */    protected ForkJoinWorkerThread(ForkJoinPool pool) {        // Use a placeholder until a useful name can be set in registerWorker        super("aForkJoinWorkerThread");        this.pool = pool;        // workQueue 臨時(shí)向 pool 中進(jìn)行注冊(cè)所得        this.workQueue = pool.registerWorker(this);    }
    /**     * Callback from ForkJoinWorkerThread constructor to establish and     * record its WorkQueue.     *     * @param wt the worker thread     * @return the worker's queue     */    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 by approx half n                    int step = (n <= 4) ? 2 : ((n >>> 1) & EVENMASK) + 2;                    while (ws[i = (i + step) & m] != null) {                        if (++probes >= n) {                            workQueues = ws = Arrays.copyOf(ws, n <<= 1);                            m = n - 1;                            probes = 0;                        }                    }                }                w.hint = s;                           // use as random seed                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;    }

// java.util.concurrent.ForkJoinWorkerThread#run    /**     * This method is required to be public, but should never be     * called explicitly. It performs the main run loop to execute     * {@link ForkJoinTask}s.     */    public void run() {        if (workQueue.array == null) { // only run once            Throwable exception = null;            try {                onStart();                pool.runWorker(workQueue);            } catch (Throwable ex) {                exception = ex;            } finally {                try {                    onTermination(exception);                } catch (Throwable ex) {                    if (exception == null)                        exception = ex;                } finally {                    pool.deregisterWorker(this, exception);                }            }        }    }    // java.util.concurrent.ForkJoinPool#runWorker    /**     * Top-level runloop for workers, called by ForkJoinWorkerThread.run.     */    final void runWorker(WorkQueue w) {        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;;) {            // 取任務(wù)，執(zhí)行            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        }    }
        /**         * Executes the given task and any remaining local tasks.         */        final void runTask(ForkJoinTask<?> task) {            if (task != null) {                scanState &= ~SCANNING; // mark as busy                (currentSteal = task).doExec();                U.putOrderedObject(this, QCURRENTSTEAL, null); // release for GC                execLocalTasks();                ForkJoinWorkerThread thread = owner;                if (++nsteals < 0)      // collect on overflow                    transferStealCount(pool);                scanState |= SCANNING;                if (thread != null)                    thread.afterTopLevelExec();            }        }    // java.util.concurrent.ForkJoinTask#doExec    /**     * Primary execution method for stolen tasks. Unless done, calls     * exec and records status if completed, but doesn't wait for     * completion otherwise.     *     * @return status on exit from this method     */    final int doExec() {        int s; boolean completed;        if ((s = status) >= 0) {            try {                completed = exec();            } catch (Throwable rex) {                return setExceptionalCompletion(rex);            }            if (completed)                s = setCompletion(NORMAL);        }        return s;    }    // java.util.concurrent.RecursiveTask#exec    /**     * Implements execution conventions for RecursiveTask.     */    protected final boolean exec() {        // 即調(diào)用具體業(yè)務(wù)類的 compute 方法        result = compute();        return true;    }

// java.util.concurrent.ForkJoinPool#scan    /**     * Scans for and tries to steal a top-level task. Scans start at a     * random location, randomly moving on apparent contention,     * otherwise continuing linearly until reaching two consecutive     * empty passes over all queues with the same checksum (summing     * each base index of each queue, that moves on each steal), at     * which point the worker tries to inactivate and then re-scans,     * attempting to re-activate (itself or some other worker) if     * finding a task; otherwise returning null to await work.  Scans     * otherwise touch as little memory as possible, to reduce     * disruption on other scanning threads.     *     * @param w the worker (via its WorkQueue)     * @param r a random seed     * @return a task, or null if none found     */    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            for (int origin = r & m, k = origin, oldSum = 0, checkSum = 0;;) {                WorkQueue q; ForkJoinTask<?>[] a; ForkJoinTask<?> t;                int b, n; long c;                // 首次獲取時(shí)，是從自身隊(duì)列中獲取                if ((q = ws[k]) != null) {                    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) {                            if (ss >= 0) {                                if (U.compareAndSwapObject(a, i, t, null)) {                                    q.base = b + 1;                                    if (n < -1)       // signal others                                        signalWork(ws, q);                                    return t;                                }                            }                            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;                    }                    checkSum += b;                }                if ((k = (k + 1) & m) == origin) {    // continue until stable                    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                        long nc = ((SP_MASK & ns) |                                   (UC_MASK & ((c = ctl) - AC_UNIT)));                        w.stackPred = (int)c;         // hold prev stack top                        U.putInt(w, QSCANSTATE, ns);                        if (U.compareAndSwapLong(this, CTL, c, nc))                            ss = ns;                        else                            w.scanState = ss;         // back out                    }                    checkSum = 0;                }            }        }        return null;    }

// java.util.concurrent.ForkJoinTask#fork    /**     * 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;        // ForkJoinWorkerThread 中持有workQueue實(shí)例，可直接向其添加任務(wù)        if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)            ((ForkJoinWorkerThread)t).workQueue.push(this);        else            // 如果是外部線程，則添加到一共享pool中即可，后續(xù)將其各空閑線程處理            ForkJoinPool.common.externalPush(this);        return this;    }        // java.util.concurrent.ForkJoinPool.WorkQueue#push        /**         * Pushes a task. Call only by owner in unshared queues.  (The         * shared-queue version is embedded in method externalPush.)         *         * @param task the task. Caller must ensure non-null.         * @throws RejectedExecutionException if array cannot be resized         */        final void push(ForkJoinTask<?> task) {            ForkJoinTask<?>[] a; ForkJoinPool p;            int b = base, s = top, n;            if ((a = array) != null) {    // ignore if queue removed                int m = a.length - 1;     // fenced write for task visibility                U.putOrderedObject(a, ((m & s) << ASHIFT) + ABASE, task);                U.putOrderedInt(this, QTOP, s + 1);                if ((n = s - b) <= 1) {                    if ((p = pool) != null)                        p.signalWork(p.workQueues, this);                }                else if (n >= m)                    growArray();            }        }
/** * A thread managed by a {@link ForkJoinPool}, which executes * {@link ForkJoinTask}s. * This class is subclassable solely for the sake of adding * functionality -- there are no overridable methods dealing with * scheduling or execution.  However, you can override initialization * and termination methods surrounding the main task processing loop. * If you do create such a subclass, you will also need to supply a * custom {@link ForkJoinPool.ForkJoinWorkerThreadFactory} to * {@linkplain ForkJoinPool#ForkJoinPool use it} in a {@code ForkJoinPool}. * * @since 1.7 * @author Doug Lea */public class ForkJoinWorkerThread extends Thread {    /*     * ForkJoinWorkerThreads are managed by ForkJoinPools and perform     * ForkJoinTasks. For explanation, see the internal documentation     * of class ForkJoinPool.     *     * This class just maintains links to its pool and WorkQueue.  The     * pool field is set immediately upon construction, but the     * workQueue field is not set until a call to registerWorker     * completes. This leads to a visibility race, that is tolerated     * by requiring that the workQueue field is only accessed by the     * owning thread.     *     * Support for (non-public) subclass InnocuousForkJoinWorkerThread     * requires that we break quite a lot of encapsulation (via Unsafe)     * both here and in the subclass to access and set Thread fields.     */
    final ForkJoinPool pool;                // the pool this thread works in    final ForkJoinPool.WorkQueue workQueue; // work-stealing mechanics    ...}

// java.util.concurrent.ForkJoinTask#join    /**     * 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);        // 任務(wù)完成后，主動(dòng)獲取結(jié)果        return getRawResult();    }    /**     * Throws exception, if any, associated with the given status.     */    private void reportException(int s) {        if (s == CANCELLED)            throw new CancellationException();        if (s == EXCEPTIONAL)            rethrow(getThrowableException());    }    // java.util.concurrent.RecursiveTask#getRawResult    public final V getRawResult() {        return result;    }

    /**     * 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;        return (s = status) < 0 ? s :            ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?            // 取當(dāng)前任務(wù)執(zhí)行， doExec 執(zhí)行任務(wù)，awaitJoin 等待執(zhí)行完成            (w = (wt = (ForkJoinWorkerThread)t).workQueue).            tryUnpush(this) && (s = doExec()) < 0 ? s :            wt.pool.awaitJoin(w, this, 0L) :            externalAwaitDone();    }
    // java.util.concurrent.ForkJoinPool#awaitJoin    /**     * Helps and/or blocks until the given task is done or timeout.     *     * @param w caller     * @param task the task     * @param deadline for timed waits, if nonzero     * @return task status on exit     */    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);                // 遞歸添加任務(wù)等待完成                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;                else if ((ms = TimeUnit.NANOSECONDS.toMillis(ns)) <= 0L)                    ms = 1L;                if (tryCompensate(w)) {                    task.internalWait(ms);                    U.getAndAddLong(this, CTL, AC_UNIT);                }            }            U.putOrderedObject(w, QCURRENTJOIN, prevJoin);        }        return s;    }        // java.util.concurrent.ForkJoinPool.WorkQueue#tryRemoveAndExec        /**         * If present, removes from queue and executes the given task,         * or any other cancelled task. Used only by awaitJoin.         *         * @return true if queue empty and task not known to be done         */        final boolean tryRemoveAndExec(ForkJoinTask<?> task) {            ForkJoinTask<?>[] a; int m, s, b, n;            if ((a = array) != null && (m = a.length - 1) >= 0 &&                task != null) {                while ((n = (s = top) - (b = base)) > 0) {                    for (ForkJoinTask<?> t;;) {      // traverse from s to b                        long j = ((--s & m) << ASHIFT) + ABASE;                        if ((t = (ForkJoinTask<?>)U.getObject(a, j)) == null)                            return s + 1 == top;     // shorter than expected                        else if (t == task) {                            boolean removed = false;                            if (s + 1 == top) {      // pop                                if (U.compareAndSwapObject(a, j, task, null)) {                                    U.putOrderedInt(this, QTOP, s);                                    removed = true;                                }                            }                            else if (base == b)      // replace with proxy                                removed = U.compareAndSwapObject(                                    a, j, task, new EmptyTask());                            // 執(zhí)行子任務(wù)                            if (removed)                                task.doExec();                            break;                        }                        else if (t.status < 0 && s + 1 == top) {                            if (U.compareAndSwapObject(a, j, t, null))                                U.putOrderedInt(this, QTOP, s);                            break;                  // was cancelled                        }                        if (--n == 0)                            return false;                    }                    if (task.status < 0)                        return false;                }            }            return true;        }