volatile int status;

// 完成状态（高 4 位，均为负值）
static final int NORMAL      = 0xf0000000;  // 正常完成
static final int CANCELLED   = 0xc0000000;  // 取消
static final int EXCEPTIONAL = 0x80000000;  // 异常

// 信号标记（低位，正值）
static final int SIGNAL      = 0x00010000;  // 有线程在等待
static final int SMASK       = 0x0000ffff;  // 低位掩码（取标记用）

volatile long ctl;

// 子域位偏移
// |----------|----------|----------|----------|
//  63      48 47      32 31      16 15       0
//      AC         TC         SS        ID

private static final long SP_MASK    = 0xffffffffL;      // 低 32 位掩码
private static final long UC_MASK    = ~SP_MASK;          // 高 32 位掩码

private static final long AC_UNIT    = 0x0001L << 48;     // AC 增量
private static final long TC_UNIT    = 0x0001L << 32;     // TC 增量
private static final long ADD_WORKER = 0x0001L << (32+15); // 需添加工作线标志

static final class WorkQueue {
    volatile int scanState;     // 扫描状态：<0 失活(INACTIVE)，奇数=SCANNING
    int stackPred;              // 前一个栈顶的 ctl 低 32 位值
    int nsteals;                // 窃取任务计数
    int hint;                   // 窃取者随机索引提示
    int config;                 // 池索引(低16位) + 模式(高16位)
    volatile int qlock;         // 1:锁定中，<0:终止，0:正常
    volatile int base;          // poll端索引（队列头，FIFO出队）
    int top;                    // push端索引（栈顶，LIFO入队）
    ForkJoinTask<?>[] array;    // 任务数组，容量为 2 的幂
    final ForkJoinWorkerThread owner; // 所属线程，SHARED_QUEUE 模式下为 null
    volatile Thread parker;     // 阻塞期间设为 owner
    volatile ForkJoinTask<?> currentJoin;   // 正在被 join 的任务
    volatile ForkJoinTask<?> currentSteal;  // 当前正在执行的窃取任务
}

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)) { // ② CAS 锁定随机偶数槽位
        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);    // ④ 更新 top
            U.putOrderedInt(q, QLOCK, 0);       // ⑤ 解锁
            if (n <= 1)
                signalWork(ws, q);              // ⑥ 通知线程来执行
            return;
        }
        U.compareAndSwapInt(q, QLOCK, 1, 0);    // ⑦ 失败则回退
    }
    externalSubmit(task);                       // ⑧ 完整路径（首次或特殊场景）
}

final void signalWork(WorkQueue[] ws, WorkQueue q) {
    long c; int sp, i; WorkQueue v; Thread p;
    while ((c = ctl) < 0L) {                        // ① AC < 0，活跃线程不足
        if ((sp = (int)c) == 0) {                   // ② 无空闲线程
            if ((c & ADD_WORKER) != 0L)             // ③ 允许添加线程
                tryAddWorker(c);                    // ④ 创建新线程
            break;
        }
        if (ws == null)
            break;
        if (ws.length <= (i = sp & SMASK))          // ⑤ 栈顶索引越界
            break;
        if ((v = ws[i]) == null)
            break;
        int vs = (sp + SS_SEQ) & ~INACTIVE;         // ⑥ 计算新scanState（加版本号）
        int d = sp - v.scanState;                   // ⑦ 版本号差值
        long nc = (UC_MASK & (c + AC_UNIT)) | (SP_MASK & v.stackPred);
        if (d == 0 && U.compareAndSwapLong(this, CTL, c, nc)) {  // ⑧ CAS 更新 ctl
            v.scanState = vs;
            if ((p = v.parker) != null)
                U.unpark(p);                        // ⑨ 唤醒阻塞线程
            break;
        }
    }
}

private boolean createWorker() {
    ForkJoinWorkerThreadFactory fac = factory;
    Throwable ex = null;
    ForkJoinWorkerThread wt = null;
    try {
        if (fac != null && (wt = fac.newThread(this)) != null) {
            wt.start();                          // 启动线程 → 调用 ForkJoinWorkerThread.run()
            return true;
        }
    } catch (Throwable rex) {
        ex = rex;
    }
    deregisterWorker(wt, ex);                    // 创建失败，清理
    return false;
}

final WorkQueue registerWorker(ForkJoinWorkerThread wt) {
    // ...
    WorkQueue w = new WorkQueue(this, wt);
    int i = 0;
    int mode = config & MODE_MASK;
    int rs = lockRunState();
    try {
        WorkQueue[] ws; int n;
        if ((ws = workQueues) != null && (n = ws.length) > 0) {
            int s = indexSeed += SEED_INCREMENT;
            int m = n - 1;
            i = ((s << 1) | 1) & m;             // 计算奇数索引
            if (ws[i] != null) {                 // 槽位冲突
                int probes = 0;
                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;
                    }
                }
            }
            // ...
            ws[i] = w;
        }
    } finally { unlockRunState(rs, rs); }
    wt.setName(workerNamePrefix.concat(Integer.toString(i >>> 1)));
    return w;
}

public void run() {
    if (workQueue.array == null) {       // 首次运行才初始化
        Throwable exception = null;
        try {
            onStart();
            pool.runWorker(workQueue);   // 核心循环
        } catch (Throwable ex) {
            exception = ex;
        } finally {
            try { onTermination(exception); }
            catch (Throwable ex) { ... }
            finally {
                pool.deregisterWorker(this, exception);
            }
        }
    }
}

final void runWorker(WorkQueue w) {
    w.growArray();                       // 分配任务数组
    int seed = w.hint;
    int r = (seed == 0) ? 1 : seed;
    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;                     // ⑤ 更新随机种子
    }
}

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;
        for (int origin = r & m, k = origin, oldSum = 0, checkSum = 0;;) {
            WorkQueue q; ForkJoinTask<?>[] a; ForkJoinTask<?> t;
            int b, n; long c;
            if ((q = ws[k]) != null) {                       // ① 当前槽位有队列
                if ((n = (b = q.base) - q.top) < 0 &&        // ② 队列非空
                    (a = q.array) != null) {
                    long i = (((a.length - 1) & b) << ASHIFT) + ABASE;
                    if ((t = ((ForkJoinTask<?>)
                              U.getObjectVolatile(a, i))) != null && // ③ volatile 读 base 位置
                        q.base == b) {
                        if (ss >= 0) {
                            if (U.compareAndSwapObject(a, i, t, null)) {  // ④ CAS 取走任务
                                q.base = b + 1;              // ⑤ 更新 base
                                if (n < -1)
                                    signalWork(ws, q);       // ⑥ 还剩任务，通知其他线程
                                return t;
                            }
                        }
                        else if (oldSum == 0 &&
                                 w.scanState < 0)
                            tryRelease(c = ctl, ws[m & (int)c], AC_UNIT); // ⑦ 唤醒失活线程
                    }
                    if (ss < 0)
                        ss = w.scanState;
                }
            }
            if (--checkSum == 0) {
                if (ss < 0) {                                 // ⑧ 当前线程已失活
                    if (oldSum == 0 && (oldSum = checkSum = 1) == 1) {
                        // ... 尝试再次扫描 (双重检查)
                        continue;
                    }
                }
                // ... oldSum/checkSum 递增逻辑，防止无限循环
            }
            if ((k = (k + 1) & m) == origin) {               // ⑨ 扫描完一整圈
                if (ss >= 0)
                    return null;
                break;
            }
        }
        // ⑩ 失活——CAS 将 scanState 设为 INACTIVE，更新 ctl
        int ns = ss | INACTIVE;
        long nc = ((SP_MASK & ns) | (UC_MASK & ((c = ctl) - AC_UNIT)));
        w.stackPred = (int) c;
        U.putInt(w, QSCANSTATE, ns);
        if (U.compareAndSwapLong(this, CTL, c, nc)) {
            ss = ns;
            // ... 更新 w.hint
        }
        // ⑪ 灭活后再次扫描一圈
        // ... 如果仍然无任务，返回 null（最终进入 awaitWork）
    }
    return null;
}

private boolean tryRelease(long c, WorkQueue v, long inc) {
    int sp = (int) c, vs = (sp + SS_SEQ) & ~INACTIVE;
    if (v != null && v.scanState == sp) {
        long nc = (UC_MASK & (c + inc)) | (SP_MASK & v.stackPred);
        if (U.compareAndSwapLong(this, CTL, c, nc)) {
            v.scanState = vs;
            if ((p = v.parker) != null)
                U.unpark(p);
            return true;
        }
    }
    return false;
}

public final ForkJoinTask<V> fork() {
    Thread t;
    if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)
        ((ForkJoinWorkerThread)t).workQueue.push(this);  // ① Worker 线程：push 到自身队列
    else
        ForkJoinPool.common.externalPush(this);           // ② 外部线程：走 externalPush
    return this;
}

final void push(ForkJoinTask<?> task) {
    ForkJoinTask<?>[] a; ForkJoinPool p;
    int b = base, s = top, n;
    if ((a = array) != null) {
        int m = a.length - 1;
        U.putOrderedObject(a, ((m & s) << ASHIFT) + ABASE, task);  // ① 放入 (m & top) 位置
        U.putOrderedInt(this, QTOP, s + 1);                        // ② top + 1
        if ((n = s - b) <= 1) {
            if ((p = pool) != null)
                p.signalWork(p.workQueues, this);                  // ③ 队列原为空，通知
        }
        else if (n >= m)
            growArray();                                           // ④ 满则扩容
    }
}

public final V join() {
    int s;
    if ((s = doJoin() & DONE_MASK) != NORMAL)
        reportException(s);
    return getRawResult();
}

private int doJoin() {
    int s; Thread t; ForkJoinWorkerThread wt; ForkJoinPool.WorkQueue w;
    return (s = status) < 0 ? s :    // ① 检查是否已经完成
        ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
        (w = (wt = (ForkJoinWorkerThread)t).workQueue).
        tryUnpush(this) && (s = doExec()) < 0 ? s :  // ② 尝试从本地 top 弹出并执行
        wt.pool.awaitJoin(w, this, 0L) :             // ③ 否则进入等待
        externalAwaitDone();                          // ④ 外部线程的等待路径
}

final int awaitJoin(WorkQueue w, ForkJoinTask<?> task, long deadline) {
    int s = 0;
    if (task != w.currentJoin)
        w.currentJoin = task;                        // ① 记录当前 join 的任务
    ForkJoinTask<?> prevJoin = w.currentJoin;        //    供 helpStealer 定位
    for (int k = 0, side = 0;;) {
        if ((s = task.status) < 0)                   // ② 任务已完成
            break;
        if (side == 0) {
            ForkJoinTask<?> subtask = w.currentSteal; // ③ 当前窃取的任务
            if (subtask != null)
                ForkJoinTask.helpExpungeStaleExceptions();
            if (subtask instanceof CountedCompleter)
                helpComplete(w, (CountedCompleter<?>) subtask, k, side);
            else if (w.base == w.top ||              // ④ 本地队列为空
                     w.tryRemoveAndExec(task))       // ⑤ 尝试找到并执行
                helpStealer(w, task, k);             // ⑥ 核心：帮助窃取者执行
        }
        // ... 更多侧循环逻辑
        if ((s = task.status) < 0)                   // ⑦ 再次检查
            break;
        // ⑧ tryCompensate：补偿机制
        Thread.interrupted();
        if (deadline == 0L)
            ForkJoinPool.tryCompensate(w);
        // ⑨ 内部等待（自旋 + timed park）
        long ms = 1L;
        task.internalWait(ms);
    }
    w.currentJoin = prevJoin;
    return s;
}

final void helpStealer(WorkQueue w, ForkJoinTask<?> task, int k) {
    WorkQueue[] ws = workQueues;
    int n = ws.length;
    if (ws != null && n > 0) {
        for (int m = n - 1, j = -n - m;;) {          // ① 从后往前扫描
            WorkQueue q = ws[j & m & 0xffff];
            if (q != null) {
                ForkJoinTask<?> t = q.currentSteal;   // ② 获取该队列当前执行的窃取任务
                if (t == task) {                       // ③ 找到窃取者
                    if (w.tryRemoveAndExec(task))      // ④ 尝试执行目标任务
                        break;
                    do {
                        if ((t = q.currentSteal) == task) {
                            if (w.base - w.top >= 0)
                                return;
                            do {} while (!w.tryRemoveAndExec(task));
                        }
                        // ⑤ 从窃取者队列 poll 任务执行
                        while (q.base != q.top) {
                            ForkJoinTask<?> o = q.poll();
                            if (o != null) {
                                w.runSubtask(o);       // ⑥ 帮助执行窃取者的任务
                            }
                        }
                    } while (w.base - w.top < 0);      // ⑦ 本地有新任务产生
                    break;
                }
            }
            if (--j == 0)
                break;
        }
    }
}

final boolean tryCompensate(WorkQueue w) {
    boolean canBlock;
    WorkQueue[] ws; long c; int m, pc, sp;
    if (w == null || w.qlock < 0 || (ws = workQueues) == null ||
        (m = ws.length - 1) < 0 || (pc = config & SMASK) == 0)
        canBlock = false;
    else if ((sp = (int)(c = ctl)) != 0)
        canBlock = tryRelease(c, ws[sp & 0x7fffffff], TC_UNIT); // ① 有空闲线程，唤醒
    else {
        int ac = (int)(c >> AC_SHIFT) + pc;
        int tc = (short)(c >> TC_SHIFT) + pc;
        int nbusy = 0;
        for (int i = 0; i <= m; ++i) {
            WorkQueue v = ws[i];
            if (v != null && v.scanState >= 0) ++nbusy;        // ② 统计忙碌线程
        }
        if (nbusy != (tc + ac) || c != ctl)
            canBlock = false;
        else if (tc >= pc && ac > 1 && w.isEmpty())
            canBlock = false;                                   // ③ 不需要补偿
        else
            canBlock = tryAddWorker(c);                         // ④ 创建补偿线程
    }
    return canBlock;
}

final void runTask(ForkJoinTask<?> task) {
    if (task != null) {
        scanState &= ~SCANNING;                    // ① 清除 SCANNING，标记忙碌
        (currentSteal = task).doExec();            // ② 执行窃取任务
        U.putOrderedObject(this, QCURRENTSTEAL, null);
        execLocalTasks();                          // ③ 消费本地队列中的所有任务
        ForkJoinTask<?>[] a = array;
        if (a != null && a.length > 0)
            reinitialize();                        // ④ 重置 WorkQueue 内部状态
        if (++nsteals < 0)
            transferStealCount(pool);              // ⑤ 窃取计数溢出处理
        scanState |= SCANNING;                     // ⑥ 恢复 SCANNING 状态
        if (owner != null)
            owner.afterTopLevelExec();             // ⑦ 顶层任务执行后的钩子
    }
}

final void execLocalTasks() {
    int b = base, m, s;
    ForkJoinTask<?>[] a = array;
    if (b - (s = top - 1) <= 0 && a != null && (m = a.length - 1) >= 0) {
        if ((config & FIFO_QUEUE) == 0) {          // LIFO 模式（默认）
            for (ForkJoinTask<?> t;;) {
                if ((t = (ForkJoinTask<?>)
                     U.getObject(a, ((m & s) << ASHIFT) + ABASE)) == null)
                    break;
                U.putOrderedInt(this, QTOP, s);
                t.doExec();
                if (base - (s = top - 1) > 0)
                    break;                         // 队列为空则退出
            }
        }
        else                                       // FIFO 模式
            pollAndExecAll();                      // 从 base 端取，类似窃取的顺序
    }
}

// 方式一：自定义并行度
ForkJoinPool pool = new ForkJoinPool(4);  // 4 个工作线程

// 方式二：完整参数
ForkJoinPool pool = new ForkJoinPool(
    4,                    // parallelism：并行度
    ForkJoinPool.defaultForkJoinWorkerThreadFactory,
    null,                 // handler：异常处理器
    false                 // asyncMode：false=LIFO(默认)，true=FIFO
);

// 方式三：使用公共池（JDK 8+）
ForkJoinPool common = ForkJoinPool.commonPool();
// 默认并行度 = Runtime.getRuntime().availableProcessors() - 1

// 方式四：通过 Executors
ExecutorService pool = Executors.newWorkStealingPool();
// 等同于 new ForkJoinPool(parallelism, factory, handler, true)

public class ArraySum {
    static class SumTask extends RecursiveTask<Long> {
        private final int[] arr;
        private final int start, end;
        static final int THRESHOLD = 1000;

        SumTask(int[] arr, int start, int end) {
            this.arr = arr;
            this.start = start;
            this.end = end;
        }

        @Override
        protected Long compute() {
            if (end - start <= THRESHOLD) {
                long sum = 0;
                for (int i = start; i < end; i++) sum += arr[i];
                return sum;
            }
            int mid = (start + end) >>> 1;
            SumTask left = new SumTask(arr, start, mid);
            SumTask right = new SumTask(arr, mid, end);
            invokeAll(left, right);   // 推荐：框架自动优化并行
            return left.join() + right.join();
        }
    }

    public static void main(String[] args) {
        int[] arr = new int[10_000_000];
        Arrays.parallelSetAll(arr, i -> i);  // 并行填充
        ForkJoinPool pool = new ForkJoinPool();
        long result = pool.invoke(new SumTask(arr, 0, arr.length));
        System.out.println("Sum: " + result);
    }
}

public class QuickSortParallel {
    static class SortTask extends RecursiveAction {
        private final int[] arr;
        private final int lo, hi;
        static final int THRESHOLD = 1000;

        SortTask(int[] arr, int lo, int hi) {
            this.arr = arr;
            this.lo = lo;
            this.hi = hi;
        }

        @Override
        protected void compute() {
            if (hi - lo <= THRESHOLD) {
                Arrays.sort(arr, lo, hi);
                return;
            }
            int pivot = partition(arr, lo, hi);
            SortTask left = new SortTask(arr, lo, pivot);
            SortTask right = new SortTask(arr, pivot + 1, hi);
            invokeAll(left, right);
        }

        private int partition(int[] arr, int lo, int hi) {
            // 标准快排分区逻辑
            int pivot = arr[lo];
            int i = lo, j = hi;
            while (i < j) {
                while (i < j && arr[--j] >= pivot);
                arr[i] = arr[j];
                while (i < j && arr[++i] <= pivot);
                arr[j] = arr[i];
            }
            arr[i] = pivot;
            return i;
        }
    }
}

// 错误：过度 fork
left.fork();
right.fork();
long result = left.join() + right.join();  // 两个 join 都阻塞

// 正确：一个当前线程执行，一个 fork
left.fork();
long rightResult = right.compute();        // 当前线程直接计算
long leftResult = left.join();             // 等待异步的那个
return leftResult + rightResult;

// 最佳：使用 invokeAll
invokeAll(left, right);
return left.join() + right.join();

SumTask task = new SumTask(arr, 0, arr.length);
pool.execute(task);  // 用 execute 而非 invoke
// task 内部的异常被捕获并保存在 status 中
// 必须显式调用 join()/get() 才会抛出异常
try {
    task.join();
} catch (RuntimeException e) {
    // 处理异常
}

if (task.isCompletedAbnormally()) {
    // 通过 getException() 获取原始异常
    Throwable ex = task.getException();
}

// 错误：在 commonPool 中执行耗时任务会影响全局
IntStream.range(0, 100).parallel().forEach(i -> {
    Thread.sleep(10000);  // 阻塞 commonPool 线程，影响其他所有使用它的功能
});

// 正确：耗时/阻塞任务使用自定义线程池
ForkJoinPool customPool = new ForkJoinPool(4);
customPool.submit(() -> {
    // 耗时操作
}).get();