如果把归并算法的递归过程，想象为一棵树，那么每个节点就是一个任务task，其中,父节点的结果需要子节点先计算了，才能计算出来。
好，说到树，树的遍历应该有所了解，必须先计算出子节点结果，再计算父节点，后序遍历，就可以做,但后序遍历还是在一个线程中运行，如何做到多线程并行遍历树呢？
这应该就是ForkJoinPool 以及 ForkJoinTask 等共同要解决的问题。
-
1.在ForkJoin中，Fork 意味着新的子节点出现，新的Task产生。如何让这些Task并发处理，没错偷窃算法，别的线程偷了该task,在别的线程运行,达到并行的效果。
- 1.为了避免疑惑,需要先解释一个概念,新的Task产生,会将其放在一个WorkQueue的队列中,而这个队列,可以是先进先出,也可以是先进后出.
-
2.Join 代表要执行该task
- Join 结束意味着拿到子节点task的结果, 反之, 子节点task还没计算完,Join是没法退出的,
- 这就意味着Join中要处理很多事,不然太浪费线程了. 它就不再像之前的get()方法,会一直循环等待结果.这里的Join像是一边等结果,一边做事,从自己对应的WorkQueue的队列里拿task 或者从别的Work Queue中偷task. 有点像,单线程的递归,没有结果就继续跑子节点的任务,直到任务有结果反馈,更何况这里还可以偷task.
- 这里的task中有一个status来表示自己状态, 当status<0,意味着运行结束或出了异常或取消了等,基本上就可以退出Join了
-
3.为什么可以偷?
- ForkJoinPool 有一个WorkQueues数组,单一的WorkQueue,有自己的对应的pool和线程及任务队列
也就是说,线程可以访问pool中的WorkQueue组,也就可以偷task. - 每个task 都status ,上面提及过,也方便了"偷",因为task运行在别的线程,join还是可以查看task 的status.
- ForkJoinPool 有一个WorkQueues数组,单一的WorkQueue,有自己的对应的pool和线程及任务队列
-
4.线程(线程与WorkQueue一一对应)的管理
- 一个新的线程的运行逻辑是什么?自己的Work Queue中queue中有任务就去执行,没有任务,要去偷.
- 这意味啥?1.run()程序中包含 是一个死循环,要时刻检测pool中有无任务执行.但是,没错,但是,这很蠢.因为很浪费.任务要是都运行完了,难道整个线程都要跑死循环.所以没有这么做.
//看下这里的run()程序中的核心部分
- 这意味啥?1.run()程序中包含 是一个死循环,要时刻检测pool中有无任务执行.但是,没错,但是,这很蠢.因为很浪费.任务要是都运行完了,难道整个线程都要跑死循环.所以没有这么做.
- 一个新的线程的运行逻辑是什么?自己的Work Queue中queue中有任务就去执行,没有任务,要去偷.
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;;) {
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
}
}
- 这就牵扯到线程的管理了,要新增多少个线程?何时park或unpark 线程.
当然,这只能算ForkJoin的主线工作,还有很多细节很多功能,并未补充.但也说明了一些问题,有时候,代码复杂,往往在于需要多个点配合完成. 甚至会杂糅在一起,难度升级,所以,还是需要拆解工作.
接下来是主线工作的源码分析工作.细节问题不再此处拓展,下次再说.
1.从Fork 出发
//fork,新增一个task,往线程的workQueue中新增任务 public final ForkJoinTaskfork() { Thread t; if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ((ForkJoinWorkerThread)t).workQueue.push(this); else ForkJoinPool.common.externalPush(this); return this; } 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 //这是找到top对应的起点位置offset U.putOrderedObject(a, ((m & s) << ASHIFT) + ABASE, task); //top处+1. U.putOrderedInt(this, QTOP, s + 1); if ((n = s - b) <= 1) { if ((p = pool) != null) // 这里就是在看pool中是否需要新增WorkQueue,并涉及线程池的逻辑,不展开了 p.signalWork(p.workQueues, this); } else if (n >= m) growArray(); } }
接下来看看Join
public final V join() {
int s;
if ((s = doJoin() & DONE_MASK) != NORMAL)
reportException(s);
return getRawResult();
}
//关键是doJoin
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).
//这里解释一下,是否从队列中看能否吐出该task,并且执行,否则,awaitJoin
tryUnpush(this) && (s = doExec()) < 0 ? s :
wt.pool.awaitJoin(w, this, 0L) :
externalAwaitDone();
}
// 接着看awaitJoin
// 直接看一个死循环,退出的条件只有一个states<0.
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);
// 如果队列中没有任务了,或者把任务队列的任务都运行完了
else if (w.base == w.top || w.tryRemoveAndExec(task))
//开始帮助偷任务了
helpStealer(w, task);
if ((s = task.status) < 0)
break;
long ms, ns;
//join中,这里的deadline=0,所以,不会因为时间退出
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;
}
//这里的tryRemoveAndExec 尝试能否将任务队列中的任务运行完
final boolean tryRemoveAndExec(ForkJoinTask task) {
ForkJoinTask[] a; int m, s, b, n;
if ((a = array) != null && (m = a.length - 1) >= 0 &&
task != null) {
// 只要top>base就说明还有任务,说明下,top ,base是WorkQueue的成员变量,值会改变的.
while ((n = (s = top) - (b = base)) > 0) {
//while下又有for循环,因为top,base 有变化可能
//其中的j,s 分别对应top的开始位置,和top索引.
//比如 top开始为0,当有一个值写入时,top为1,但是按照数组习惯索引为0.
for (ForkJoinTask t;;) { // traverse from s to b
long j = ((--s & m) << ASHIFT) + ABASE;
if ((t = (ForkJoinTask)U.getObject(a, j)) == null)
//top对应的任务为null,返回true
return s + 1 == top; // shorter than expected
//如果找到task,有可能删除并执行.
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());
if (removed)
// 找到了,就运行该task
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
}
//如果n 用完,长度跑完了,就返回false
if (--n == 0)
return false;
}
//task的status<0 ,返回false
if (task.status < 0)
return false;
}
}
return true;
}
从Join中可以发现,Join 不闲着.
在看看怎么偷,这里展示第一次WorkQueue的偷,而不是Join的helpStealer(…)
这里主要是逻辑流程.
(不关心就跳过,这里主要玩了一个不断尝试偷,但有可能没得偷,或者偷的时候线程本身被unpark了,然后怎么办的事,很细节,简单的过程,当然是偷到了,返回偷的任务即可.)
1.选一个随机数,从WorkQueues 数组中选一个WorkQueue
2.如果WorkQueue q !=null,checkSum +=b,checkSum是用来记WorkQueue的数量,再去看它是否有任务
3.如果有任务,且w的线程标识是Active,即ss>=0,则可以偷取q一个不为null的Base任务,返回
4.如果没有偷到任务,进入下一个判断,k用来控制整个循环的次数,因为 (k=(k+1)&m)==origin,若为true, 恰好是k 增长了m.相当于把WorkQueues数组转完了.
还没有发现可以偷的WorkQueue,就将自己设为unActive
还有可能是线程被激活时,没有偷成功,需要进入循环继续
还有可能ss<0 就直接退出循环
5.如果条件2的情况下,ss<0,会尝试释放idel线程,注意这里面首先释放的是 c = ctl, ws[m & (int)c] ctl 中对应的WorkQueue,而不是q,尝试后如果q的ss还是<0 就将K重置, 重新循环.
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;
//对应 逻辑步骤 2
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) {
//对应步骤3
if (ss >= 0) {
if (U.compareAndSwapObject(a, i, t, null)) {
q.base = b + 1;
if (n < -1) // signal others
signalWork(ws, q);
return t;
}
}
//步骤5
else if (oldSum == 0 && // try to activate
w.scanState < 0)
tryRelease(c = ctl, ws[m & (int)c], AC_UNIT);
}
//步骤5
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;
}
//步骤4
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;
}
内容已经够多了,但是,ForkJoin 还有很多没有涉及.
简单总结下,“偷” 确实实现了并行化,操作,由于,pool涉及到线程池,就无可厚非的涉及到线程的管理,而且,在细节部分又大量使用Unsafe 操作,以及位运算.所以,对源码感兴趣的可以探究一下,这是一个数据结构不算复杂,但是确实多条件控制下的程序.
最后的最后,为什么我把这个理解为树呢? 虽然,连树的结构都找不到,这是因为抽象的看Join,就是一个递归,它在不断的运行子任务直到返回结果.但这里的单线程的递归,被打散在其它线程中.树是为了理解吧.
