ForkjoinPool -1

Posted 2023-04-28

参考技术A

ForkJoin是用于并行执行任务的框架， 是一个把大任务分割成若干个小任务，最终汇总每个小任务结果后得到大任务结果的框架。Fork就是把一个大任务切分为若干子任务并行的执行，Join就是合并这些子任务的执行结果，最后得到这个大任务的结果。

下面是一个是一个简单的Join/Fork计算过程，将1—1001数字相加

通常这样个模型，你们会想到什么？

Release Framework ？ 常见的处理模型是什么？ task pool - worker pool的模型。 但是Forkjoinpool 采取了完全不同的模型。
ForkJoinPool一种ExecutorService的实现，运行ForkJoinTask任务。ForkJoinPool区别于其它ExecutorService，主要是因为它采用了一种工作窃取(work-stealing)的机制。所有被ForkJoinPool管理的线程尝试窃取提交到池子里的任务来执行，执行中又可产生子任务提交到池子中。
ForkJoinPool维护了一个WorkQueue的数组(数组长度是2的整数次方，自动增长)。每个workQueue都有任务队列(ForkJoinTask的数组)，并且用base、top指向任务队列队尾和队头。work-stealing机制就是工作线程挨个扫描任务队列，如果队列不为空则取队尾的任务并执行。示意图如下

流程图：

pool属性

workQueues是pool的属性，它是WorkQueue类型的数组。externalPush和externalSubmit所创建的workQueue没有owner(即不是worker)，且会被放到workQueues的偶数位置；而createWorker创建的workQueue（即worker）有owner，且会被放到workQueues的奇数位置。

WorkQueue的几个重要成员变量说明如下：

这是WorkQueue的config，高16位跟pool的config值保持一致，而低16位则是workQueue在workQueues数组的位置。
从workQueues属性的介绍中，我们知道，不是所有workQueue都有worker，没有worker的workQueue称为公共队列（shared queue），config的第32位就是用来判断是否是公共队列的。在externalSubmit创建工作队列时，有：
q.config = k | SHARED_QUEUE;
其中q是新创建的workQueue，k就是q在workQueues数组中的位置，SHARED_QUEUE=1<<31，注意这里config没有保留mode的信息。
而在registerWorker中，则是这样给workQueue的config赋值的：
w.config = i | mode;
w是新创建的workQueue，i是其在workQueues数组中的位置，没有设置SHARED_QUEUE标记位

scanState是workQueue的属性，是int类型的。scanState的低16位可以用来定位当前worker处于workQueues数组的哪个位置。每个worker在被创建时会在其构造函数中调用pool的registerWorker，而registerWorker会给scanState赋一个初始值，这个值是奇数，因为worker是由createWorker创建，并会被放到WorkQueues的奇数位置，而createWorker创建worker时会调用registerWorker。
简言之，worker的scanState初始值是奇数，非worker的scanstate初始值=INACTIVE=1<<31，小于0（非worker的workQueue在externalSubmit中创建）。
当每次调用signalWork（或tryRelease）唤醒worker时，worker的高16位就会加1
另外，scanState<0表示worker未激活，当worker调用runtask执行任务时，scanState会被置为偶数，即设置scanState的最右边一位为0。

worker休眠时，是这样存储的

worker的唤醒类似这样：

在worker休眠的4行伪码中，让ctl的低32位的值变为worker.scanState，这样下次就可以通过scanState唤醒该worker。唤醒该worker时，把该worker的preStack设置为ctl低32位的值，这样下下次唤醒的worker就是scanState等于该preStack的worker。
这里通过preStack保存下一个worker，这个worker比当前worker更早地在等待，所以形成一个后进先出的栈。

runState是int类型的值，控制整个pool的运行状态和生命周期，有下面几个值（可以好几个值同时存在）：

如果runState值为0，表示pool尚未初始化。
RSLOCK表示锁定pool，当添加worker和pool终止时，就要使用RSLOCK锁定整个pool。如果由于runState被锁定，导致其他操作等待runState解锁（通常用wait进行等待），当runState设置了RSIGNAL，表示runState解锁，并通知（notifyAll）等待的操作。
剩下4个值都跟runState生命周期有关，都可以顾名思义：
当需要停止时，设置runState的STOP值，表示准备关闭，这样其他操作看到这个标记位，就不会继续操作，比如tryAddWorker看到STOP就不会再创建worker：

而tryTerminate对这些生命周期状态的处理则是这样的：

当前top和base的初始值为 INITIAL_QUEUE_CAPACITY >>>1= (1 << 13)>>>1 = 8192/2。然后push一个task之后，top+=1，也就是说，top对应的位置是没有task的，最近push进来的task在top-1的位置。而base的位置则能对应到task，base对应最先放进队列的task，top-1对应最后放进队列的task。

qlock值含义：1: locked, < 0: terminate; else 0
即当qlock值位0时，可以正常操作，值=1时，表示锁定

int SQMASK=0x007e，则任何整数跟SQMASK位与后，得到的数就是偶数。
证明：
注意这里化为二进制是0111 1110，尤其注意最右边第一位是0，任何数跟最右边第一位是0的数位与后，得到的数就是偶数，因为位与之后，第一位就是0，比如s=A&SQMASK，A可以是任意整数，然后把s按二进制进行多项式展开，则有s=2 ⁿ¹⁺² n2 ……+2^nn，这里n≥1，所以s可以被2整除，即s是偶数。
所以一个数是奇数还是偶数，看其最右边第一位即可。

我们知道workQueue有externalPush创建的和createWorker创建的worker，两种方式创建的workQueue，其放置到workQueues的位置是不同的，前者放到workQueue的偶数位置，而后者则放到奇数位置。不同workQueue找到自己在workQueues的位置的算法有点不同。
下面看一下forkjoin框架获取workQueues中的偶数位置的workQueue的算法：

这样就能获取workQueues的偶数位置的workQueue。m保证m & r & SQMASK这整个运算结果不会超出workQueues的下标，SQMASK保证取到的是偶数位置的workQueue。这里有一个有趣的现象，假设0到workQueues.length-1之间有n个偶数，m & r & SQMASK每次都能取到其中一个偶数，而且连续n次取到的偶数不会出现重复值，散列性非常好。而且是循环的，即1到n次取n个不同偶数，n+1到2n也是取n次不同偶数，此时n个偶数每个都被重新取一次。下面分析下r值有什么秘密，为何能保证这样的散列性
ThreadLocalRandom内有一常量PROBE_INCREMENT = 0x9e3779b9，以及一个静态的probeGenerator =new AtomicInteger() ，然后每个线程的probe= probeGenerator.addAndGet(PROBE_INCREMENT)所以第一个线程的probe值是0x9e3779b9，第二个线程的值就是0x9e3779b9+0x9e3779b9，第三个线程的值就是0x9e3779b9+0x9e3779b9+0x9e3779b9以此类推，整个值是线性的，可以用y=kx表示，其中k=0x9e3779b9，x表示第几个线程。这样每个线程的probe可以保证不一样，而且具有很好的离散性。
实际上，可以不用0x9e3779b9这个值，用任意一个奇数都是可以的，比如1。如果用1的话，probe+=1，这样每个线程的probe就都是不同的，而且具有很好的离散性。也就是说，假设有限制条件probe<n，超过n则产生溢出。则probe自加n次后才会开始出现重复值，n次前probe每次自加的值都不同。实际上用任意一个奇数，都可以保证probe自加n次后才会开始出现重复值，有兴趣可看本文最后附录部分。由于奇数的离散性，所以只要线程数小于m或者SQMASK两者中的最小值，则每个线程都能唯一地占据一个ws中的一个位置

当一个操作是在非ForkjoinThread的线程中进行的，则称该操作为外部操作。比如我们前面执行pool.invoke，invoke内又执行externalPush。由于invoke是在非ForkjoinThread线程中进行的（这里是在main线程中进行），所以是一个外部操作，调用的是externalPush。之后task的执行是通过ForkJoinThread来执行的，所以task中的fork就是内部操作，调用的是push，把任务提交到工作队列。其实fork的实现是类似下面这样的：

即fork会根据执行自身的线程是否是ForkJoinThread的实例来判断是处于外部还是内部。那为何要区分内外部？
任何线程都可以使用ForkJoin框架，但是对于非ForkJoinThread的线程，它到底是怎样的，ForkJoin无法控制，也无法对其优化。因此区分出内外部，这样方便ForkJoin框架对任务的执行进行控制和优化
forkJoinPool.invoke(task)是把任务放入工作队列，并等待任务执行。源码如下

这里externalPush负责任务提交，externalPush源码如下：

ForkJoinPool.commonPool() 和 new ForkJoinPool(availableCPU - 1) 有啥区别？

【中文标题】ForkJoinPool.commonPool() 和 new ForkJoinPool(availableCPU - 1) 有啥区别？

【英文标题】：What is the difference between ForkJoinPool.commonPool() and new ForkJoinPool(availableCPU - 1)?ForkJoinPool.commonPool() 和 new ForkJoinPool(availableCPU - 1) 有什么区别？

【发布时间】：2019-02-04 22:38:46

【问题描述】：

在我的代码中，我有一个包含静态最终变量的类

private static final ForkJoinPool pool = new ForkJoinPool(availableCPUs - 1);

我有一个长时间运行的任务提交到池中，这将占用所有 CPU 资源。提交的任何其他任务都将挂起。 但是，当我切换到创建一个公共池时

private static final ForkJoinPool pool = ForkJoinPool.commonPool();

所有任务都可以提交执行。

我只是想知道这两段代码之间有什么区别。 commonPool() 仍然调用new ForkJoinPool() 并传递availableCPUs - 1

我还注意到commonPool() 使用SafeForkJoinWorkerThreadFactory 类型的工厂，而new ForkJoinPool() 使用ForkJoinPool$DefaultForkJoinWorkerThreadFactory。这有关系吗？

非常感谢！

【参考方案1】：

我想我明白了。

ForkJoin 维护两种类型的队列：一种是通用入站队列，另一种是每个工作线程的工作线程队列。所有工作线程将首先从一般入站队列中获取并填充它们的工作线程。在一个工作线程完成其工作队列中的所有任务后，它将尝试从其他工作线程中窃取。如果没有其他任务可以从其他工作线程中窃取，工作线程将再次从通用入站队列中获取。

但是，使用公共池，主线程也将有助于处理任务。主线程虽然没有工作队列。因此，在完成一项任务后，主线程将能够从一般入站队列中获取。

由于默认情况下，ForkJoin 队列是 LIFO，因此主线程将能够获取最后提交的任务。

【参考方案2】：

Documentation 说：

默认情况下，公共池是使用默认参数构造的。

ForkJoinPool()

使用default thread factory、无 UncaughtExceptionHandler 和非异步 LIFO 处理模式创建一个并行度等于 Runtime.availableProcessors() 的 ForkJoinPool。

那么是什么让您认为 new ForkJoinPool(availableCPUs - 1) 和 ForkJoinPool.commonPool() 会是相同大小的池？

如果您只有 2 个 CPU，那么availableCPUs - 1 表示您正在创建一个包含 1 个线程的池，即一次只能处理一个任务，因此长时间运行的任务会阻塞所有其他任务。

但是对于 2 个 CPU，availableProcessors() 意味着您将获得一个具有 2 个线程的公共池，即它可以在处理单个长时间运行的任务时处理其他任务。

【讨论】：

如果你查看 ForkJoinPool.java 的 makeCommonPool() 方法的源代码，你会注意到它实际上是使用 availableCPUs - 1 来创建公共池。此外，我的初始代码使用了没有参数的 ForkJoinPool()，它也有同样的问题。