Jdk1.7 JUC源码增量解析(4)-ForkJoin-ForkJoin任务的fork/join过程

Jdk1.7 JUC源码增量解析(4)-ForkJoin-ForkJoin任务的fork/join过程

作者：大飞

 

概述：

• 这篇通过分析一个ForkJoin任务的执行过程来分析ForkJoin的相关代码，主要侧重于分裂(fork)/合并(join)过程。

 

源码分析：

 

       还是先看一个代码示例，这个示例介绍篇出现过，这里只贴出任务代码：

public class SumTask extends RecursiveTask<Long>{
	private static final int THRESHOLD = 10;
	
	private long start;
	private long end;
	
	public SumTask(long n) {
		this(1,n);
	}
	
	private SumTask(long start, long end) {
		this.start = start;
		this.end = end;
	}
	@Override
	protected Long compute() {
		long sum = 0;
		if((end - start) <= THRESHOLD){
			for(long l = start; l <= end; l++){
				sum += l;
			}
		}else{
			long mid = (start + end) >>> 1;
		    SumTask left = new SumTask(start, mid);
		    SumTask right = new SumTask(mid + 1, end);
			left.fork();
			right.fork();
			sum = left.join() + right.join();
		}
		return sum;
	}
	private static final long serialVersionUID = 1L;
}

 

 

• 这里略过任务提交、执行的一些过程，上篇都分析过了。任务执行过程中，新建了子任务，然后进行fork操作，看下fork的源码：

    public final ForkJoinTask<V> fork() {
        ((ForkJoinWorkerThread) Thread.currentThread())
            .pushTask(this);
        return this;
    }

       fork里面其实是将任务放到当前工作线程的任务队列里面了，看下pushTask方法细节：

    final void pushTask(ForkJoinTask<?> t) {
        ForkJoinTask<?>[] q; int s, m;
        if ((q = queue) != null) {    // ignore if queue removed
            //这里首先根据当前的queueTop对队列(数组)长度取模来算出放置任务的下标
            //然后再通过下标算出偏移地址，提供给Unsafe使用。
            long u = (((s = queueTop) & (m = q.length - 1)) << ASHIFT) + ABASE;
            //设置任务。
            UNSAFE.putOrderedObject(q, u, t);
            //修改queueTop
            queueTop = s + 1;         // or use putOrderedInt
            if ((s -= queueBase) <= 2)
                pool.signalWork();
            else if (s == m)
                growQueue(); //如果队列满了，扩展一下队列容量。
        }
    }

        将任务push到任务队列，如果任务队列满了，扩展一下。

        看下扩展队列调用的growQueue方法：

    private void growQueue() {
        ForkJoinTask<?>[] oldQ = queue;
        int size = oldQ != null ? oldQ.length << 1 : INITIAL_QUEUE_CAPACITY;
        if (size > MAXIMUM_QUEUE_CAPACITY)
            throw new RejectedExecutionException("Queue capacity exceeded");
        if (size < INITIAL_QUEUE_CAPACITY)
            size = INITIAL_QUEUE_CAPACITY;
        ForkJoinTask<?>[] q = queue = new ForkJoinTask<?>[size];
        int mask = size - 1;
        int top = queueTop;
        int oldMask;
        if (oldQ != null && (oldMask = oldQ.length - 1) >= 0) {
            for (int b = queueBase; b != top; ++b) {
                long u = ((b & oldMask) << ASHIFT) + ABASE;
                Object x = UNSAFE.getObjectVolatile(oldQ, u);
                if (x != null && UNSAFE.compareAndSwapObject(oldQ, u, x, null))
                    UNSAFE.putObjectVolatile
                        (q, ((b & mask) << ASHIFT) + ABASE, x);
            }
        }
    }

        容量为原来的2倍，不超过MAXIMUM_QUEUE_CAPACITY(1 << 24)，最小为INITIAL_QUEUE_CAPACITY(1 << 13)。

 

• 完事儿了，fork过程就这么简单。fork出子任务后，当前任务的计算可能会需要子任务的结果，需要join子任务：

    sum = left.join() + right.join();

 

       看下join的源码：

    public final V join() {
        if (doJoin() != NORMAL)
            return reportResult();
        else
            return getRawResult();
    }

        join中会先调用doJoin方法，如果doJoin方法返回NORMAL，那么通过getRawResult方法来获取结果；否则会调用reportResult来处理和获取结果。

        先看下doJoin方法：

    private int doJoin() {
        Thread t; ForkJoinWorkerThread w; int s; boolean completed;
        if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) {
            if ((s = status) < 0)
                return s; //如果当前任务已经完成，直接返回状态。
            if ((w = (ForkJoinWorkerThread)t).unpushTask(this)) {
                //如果当前任务恰好是当前工作线程的队列顶端的第一个任务
                //那么将该任务出队，然后执行。
                try {
                    completed = exec();
                } catch (Throwable rex) {
                    return setExceptionalCompletion(rex);
                }
                if (completed)
                    return setCompletion(NORMAL);
            }
            //否则调用当前工作线程的joinTask方法。
            return w.joinTask(this);
        }
        else
            //如果当前线程不是ForkJoin工作线程，那么调用externalAwaitDone
            return externalAwaitDone();
    }

        doJoin方法中，如果要join的任务正好在当前任务队列的顶端，那么pop出这个任务，然后执行任务。

        注意这里调用的是ForkJoinWorkerThread的unpushTash方法，这是另一个版本的pop，看下实现：

    final boolean unpushTask(ForkJoinTask<?> t) {
        ForkJoinTask<?>[] q;
        int s;
        if ((q = queue) != null && (s = queueTop) != queueBase &&
            UNSAFE.compareAndSwapObject
            (q, (((q.length - 1) & --s) << ASHIFT) + ABASE, t, null)) {
            queueTop = s; // or putOrderedInt
            return true;
        }
        return false;
    }

        unpushTash方法中仅当给定的t是队列顶端的任务，才会返回并移除t。

 

        继续doJoin方法，如果unpushTask方法失败，就会调用ForkJoinWorkerThread的joinTask方法，看下这个方法： 

    private static final int MAX_HELP = 16;
    ...
    final int joinTask(ForkJoinTask<?> joinMe) {
        //记录之前的合并任务。
        ForkJoinTask<?> prevJoin = currentJoin;
        //设置当前工作线程的合并任务。
        currentJoin = joinMe;
        for (int s, retries = MAX_HELP;;) {
            if ((s = joinMe.status) < 0) {
                //如果合并任务已经完成，恢复之前的合并任务。
                currentJoin = prevJoin;
                return s; //返回任务状态。
            }
            if (retries > 0) {
                if (queueTop != queueBase) {
                    /*
                     * 如果当前任务队列中有任务，尝试从当前队列顶端获取给定任务
                     * (如果给定任务恰好在当前任务队列顶端的话)或者其他一个已经
                     * 被取消的任务。
                     */
                    if (!localHelpJoinTask(joinMe))
                        retries = 0;           // cannot help
                }
                else if (retries == MAX_HELP >>> 1) {
                    --retries;                 // check uncommon case
                    /*
                     * 这里尝试一种特殊情况：如果给定的任务正好在其他工作线程的
                     * 队列的底部，那么尝试窃取这个任务并执行。
                     */
                    if (tryDeqAndExec(joinMe) >= 0)
                        Thread.yield();        // 如果没成功，这里出让一下CPU。
                }
                else
                    retries = helpJoinTask(joinMe) ? MAX_HELP : retries - 1;
            }
            else {
                retries = MAX_HELP;           // restart if not done
                pool.tryAwaitJoin(joinMe);
            }
        }
    }

        joinTask方法中的逻辑也有些复杂，总结一下：

        1.joinTask方法中主体是一个无限循环，里面会先尝试帮助合并的一些操作，失败的话会继续重试，最多尝试MAX_HELP次。超过了MAX_HELP无法继续尝试的话，就会调用tryAwaitJoin方法等待合并任务。

        2.尝试过程中，如果当前任务队列中有任务，会调用localHelpJoinTask方法，如果方法调用失败会直接进入合并等待；否则会先进行helpJoinTask的尝试，尝试MAX_HELP/2次，成功的话会一直尝试，直到给定的任务完成；如果helpJoinTask尝试了MAX_HELP/2次都没有成功过，且本地队列一直没有任务，那么会进行一个特殊的尝试，会假设给定的任务在其他工作线程的任务队列的底部，然后去窃取这个任务，也会尝试MAX_HELP/2次。

 

        看下localHelpJoinTask方法： 

    private boolean localHelpJoinTask(ForkJoinTask<?> joinMe) {
        int s, i; ForkJoinTask<?>[] q; ForkJoinTask<?> t;
        if ((s = queueTop) != queueBase && (q = queue) != null &&
            (i = (q.length - 1) & --s) >= 0 &&
            (t = q[i]) != null) {
            if (t != joinMe && t.status >= 0)
                //如果当前工作线程的任务队列顶端的任务不是给定任务，
                //且任务的状态是未取消(这里如果<0，一定是取消的任务)，返回false。
                return false; 
            if (UNSAFE.compareAndSwapObject
                (q, (i << ASHIFT) + ABASE, t, null)) {
                //取出给定任务或者一个被取消的任务。
                queueTop = s;           // or putOrderedInt
                t.doExec();
            }
        }
        return true;
    }

        localHelpJoinTask方法中的逻辑是：如果当前任务队列顶端的任务是要合并的任务，或者是一个被取消的任务，那么尝试处理这个任务，返回成功；否则失败。

 

        再看下joinTask方法中调用的tryDeqAndExec方法：

    private int tryDeqAndExec(ForkJoinTask<?> t) {
        int m = pool.scanGuard & SMASK;
        ForkJoinWorkerThread[] ws = pool.workers;
        //扫描所有工作线程
        if (ws != null && ws.length > m && t.status >= 0) {
            for (int j = 0; j <= m; ++j) {
                ForkJoinTask<?>[] q; int b, i;
                ForkJoinWorkerThread v = ws[j];
                if (v != null &&
                    (b = v.queueBase) != v.queueTop &&
                    (q = v.queue) != null &&
                    (i = (q.length - 1) & b) >= 0 &&
                    q[i] ==  t) {
                    //如果有工作线程的任务队列的底部正好是给定任务t。
                    //尝试窃取t后执行。
                    long u = (i << ASHIFT) + ABASE;
                    if (v.queueBase == b &&
                        UNSAFE.compareAndSwapObject(q, u, t, null)) {
                        v.queueBase = b + 1;
                        v.stealHint = poolIndex;
                        ForkJoinTask<?> ps = currentSteal;
                        currentSteal = t;
                        t.doExec();
                        currentSteal = ps;
                    }
                    break;
                }
            }
        }
        return t.status;
    }

        tryDeqAndExec方法中的逻辑就是扫描所有工作线程，如果有工作线程的队列底部的任务正好是要合并的任务，那么窃取该任务然后处理之。

 

        最后看下joinTask方法中调用的helpJoinTask方法：

    private boolean helpJoinTask(ForkJoinTask<?> joinMe) {
        boolean helped = false;
        int m = pool.scanGuard & SMASK;
        ForkJoinWorkerThread[] ws = pool.workers;
        if (ws != null && ws.length > m && joinMe.status >= 0) {
            int levels = MAX_HELP;              // remaining chain length
            ForkJoinTask<?> task = joinMe;      // base of chain
            outer:for (ForkJoinWorkerThread thread = this;;) {
                // 找到线程thread的窃取者v
                ForkJoinWorkerThread v = ws[thread.stealHint & m];
                if (v == null || v.currentSteal != task) {
                    //如果thread没有窃取者或者v当前窃取的任务不是task，扫描工作线程数组。
                    for (int j = 0; ;) {        // search array
                        if ((v = ws[j]) != null && v.currentSteal == task) {
                            //如果找到了窃取线程，将其设置为thread的窃取线程。
                            thread.stealHint = j;
                            break;              // save hint for next time
                        }
                        if (++j > m)
                            break outer;        // 没找到的话，直接跳出outer循环。
                    }
                }
                // 找到了窃取者v。
                for (;;) {
                    ForkJoinTask<?>[] q; int b, i;
                    if (joinMe.status < 0)
                        break outer; //如果joinMe任务已经完成，跳出outer循环。
                    if ((b = v.queueBase) == v.queueTop ||
                        (q = v.queue) == null ||
                        (i = (q.length-1) & b) < 0)
                        break;                  //如果v的队列是空的，跳出当前循环。
                    long u = (i << ASHIFT) + ABASE;
                    ForkJoinTask<?> t = q[i]; 
                    if (task.status < 0)
                        break outer;            //如果task任务已经完成，跳出outer循环。
                    //尝试窃取v的任务队列底部的任务。
                    if (t != null && v.queueBase == b &&
                        UNSAFE.compareAndSwapObject(q, u, t, null)) {
                        //窃取成功后，执行任务。
                        v.queueBase = b + 1;
                        v.stealHint = poolIndex;
                        ForkJoinTask<?> ps = currentSteal;
                        currentSteal = t;
                        t.doExec();
                        currentSteal = ps;
                        helped = true;
                    }
                }
                // 再去找v的窃取者，注意这里是一个链。
                ForkJoinTask<?> next = v.currentJoin;
                if (--levels > 0 && task.status >= 0 &&
                    next != null && next != task) {
                    task = next;
                    thread = v;
                }
                else
                    break;  // 如果超过最大深度(MAX_HELP) 或者 task已经执行完成 或者 找到了头(next==null) 或者 出现循环  退出。
            }
        }
        return helped;
    }

        细节：

        这个方法的前提是当前线程需要join给定的任务joinMe，但是这个任务被其他线程(窃取者)窃取了。所以方法中首先找到窃取joinMe任务的工作线程v，如果找到了窃取者v，就会从v的任务队列中窃取任务来完成(帮助v完成任务)。但也有可能v也在join其他的任务(比如当前线程执行任务过程中，分裂出一个子任务A，工作线程v窃取了A，然后执行，执行过程中A由分裂出子任务A1，A1又被另一个工作线程v1给窃取了...是一个链)，所以方法中要顺着这个链一直找下去，目的就是能尽快的合并joinMe任务。为了避免一些情况，这里尝试的最大链深度限定为MAX_HELP。

 

        接上面joinTask方法，如果尝试不成功，会调用Pool的tryAwaitJoin方法：

    final void tryAwaitJoin(ForkJoinTask<?> joinMe) {
        int s;
        Thread.interrupted(); // clear interrupts before checking termination
        //如果joinMe未完成
        if (joinMe.status >= 0) {
            //尝试阻塞等待之前的预操作
            if (tryPreBlock()) {
                //在joinMe任务上阻塞等待
                joinMe.tryAwaitDone(0L);
                //被唤醒后的操作
                postBlock();
            }
            else if ((ctl & STOP_BIT) != 0L)
                //如果Pool关闭了，取消任务。
                joinMe.cancelIgnoringExceptions();
        }
    }

        先看下tryAwaitJoin方法中调用的tryPreBlock方法：

    /**
     * Tries to increment blockedCount, decrement active count
     * (sometimes implicitly) and possibly release or create a
     * compensating worker in preparation for blocking. Fails
     * on contention or termination.
     *
     * @return true if the caller can block, else should recheck and retry
     */
    private boolean tryPreBlock() {
        int b = blockedCount;
        //累加等待join任务的计数。
        if (UNSAFE.compareAndSwapInt(this, blockedCountOffset, b, b + 1)) {
            int pc = parallelism;
            do {
                ForkJoinWorkerThread[] ws; ForkJoinWorkerThread w;
                int e, ac, tc, rc, i;
                long c = ctl;
                int u = (int)(c >>> 32);
                if ((e = (int)c) < 0) {
                                                 // 如果Pool关闭了，跳过。
                }
                else if ((ac = (u >> UAC_SHIFT)) <= 0 && e != 0 &&
                         (ws = workers) != null &&
                         (i = ~e & SMASK) < ws.length &&
                         (w = ws[i]) != null) {
                    //如果当前活动的工作线程不大于cpu核数，且有线程在等待任务(处于空闲状态)。
                    //那么唤醒这个工作线程。
                    long nc = ((long)(w.nextWait & E_MASK) |
                               (c & (AC_MASK|TC_MASK)));
                    if (w.eventCount == e &&
                        UNSAFE.compareAndSwapLong(this, ctlOffset, c, nc)) {
                        w.eventCount = (e + EC_UNIT) & E_MASK;
                        if (w.parked)
                            UNSAFE.unpark(w);
                        return true;             
                    }
                }
                else if ((tc = (short)(u >>> UTC_SHIFT)) >= 0 && ac + pc > 1) {
                    //如果总的工作线程数量不少于cpu核心数量，且至少有一个活动的工作线程。
                    //尝试在总控信息上将AC递减。
                    long nc = ((c - AC_UNIT) & AC_MASK) | (c & ~AC_MASK);
                    if (UNSAFE.compareAndSwapLong(this, ctlOffset, c, nc))
                        return true;             
                }
                else if (tc + pc < MAX_ID) {
                    //如果不满足上面条件，这里会增加一个工作线程。
                    long nc = ((c + TC_UNIT) & TC_MASK) | (c & ~TC_MASK);
                    if (UNSAFE.compareAndSwapLong(this, ctlOffset, c, nc)) {
                        addWorker();
                        return true;            
                    }
                }
                //如果失败，这里会把刚才对b增加的1给减回去。
            } while (!UNSAFE.compareAndSwapInt(this, blockedCountOffset,
                                               b = blockedCount, b - 1));
        }
        return false;
    }

        继续看下tryAwaitJoin方法中调用的ForkJoinTask的tryAwaitDone方法：

    final void tryAwaitDone(long millis) {
        int s;
        try {
            if (((s = status) > 0 ||
                 (s == 0 &&
                  UNSAFE.compareAndSwapInt(this, statusOffset, 0, SIGNAL))) &&
                status > 0) {
                synchronized (this) {
                    if (status > 0)
                        wait(millis);
                }
            }
        } catch (InterruptedException ie) {
            // caller must check termination
        }
    }

        细节：

               阻塞等待(join任务)时，首先会检查join任务的状态，如果join任务未完成的话，才可以在join任务上等待。也就是说，join的运行状态(再回顾一下task的运行状态定义)必须大于等于0。如果join的status大于0，说明join任务上已经有其他工作线程等待了，当前线程直接等待就可以了；如果join的status等于0，说明当前线程是第一个要在join任务上阻塞等待的线程，那么会尝试将join的status改为SIGNAL(1)，然后进行阻塞等待工作。注意方法中不会处理中断异常，需要外部来处理。

 

        在看下tryAwaitJoin方法中调用的postBlock方法：

    private void postBlock() {
        long c;
        do {} while (!UNSAFE.compareAndSwapLong(this, ctlOffset,  // 累加活动线程计数
                                                c = ctl, c + AC_UNIT));
        int b;
        do {} while (!UNSAFE.compareAndSwapInt(this, blockedCountOffset, // 递减等待join任务的计数。
                                               b = blockedCount, b - 1));
    }

          

        最后，tryAwaitJoin方法中如果发现Pool关闭，会取消joinMe任务，调用其cancelIgnoringExceptions方法：

    final void cancelIgnoringExceptions() {
        try {
            cancel(false);
        } catch (Throwable ignore) {
        }
    }
    private static final int CANCELLED   = -2;
    public boolean cancel(boolean mayInterruptIfRunning) {
        return setCompletion(CANCELLED) == CANCELLED;
    }
    private int setCompletion(int completion) {
        for (int s;;) {
            if ((s = status) < 0)
                return s;
            if (UNSAFE.compareAndSwapInt(this, statusOffset, s, completion)) {
                if (s != 0)
                    synchronized (this) { notifyAll(); }
                return completion;
            }
        }
    }

        可见，cancelIgnoringExceptions中的逻辑就是将任务运行状态设置为CANCELLED，然后唤醒在任务上等待的线程(如果有的话)。

 

        最后回到join方法，如果正常完成会调用getRawResult方法：

    public abstract V getRawResult();

        ForkJoinTask中的getRawResult方法未实现，交由子类去实现，比如在RecursiveTask中：

    V result;
    ...
    public final V getRawResult() {
        return result;
    }
    ...

    protected final boolean exec() {
        result = compute();
        return true;
    }

      

 

        如果join方法中，任务非正常结束，会调用reportResult方法：

    private V reportResult() {
        int s; Throwable ex;
        if ((s = status) == CANCELLED)
            throw new CancellationException();
        if (s == EXCEPTIONAL && (ex = getThrowableException()) != null)
            UNSAFE.throwException(ex);
        return getRawResult();
    }

        方法中，如果任务状态为取消，抛出取消异常；如果任务状态是异常结束，会从异常表中获取异常，获取到的话，抛出异常。

 

 

• 简单总结一下ForkJoinPool中的ForkJoinTask的fork/join流程：

          1.fork时，会将任务添加到当前工作线程的任务队列的里面。

          2.join某个任务后，当前线程要做的首先是想办法完成这个任务，或者帮助加快这个任务的完成，如果这些尝试失败，当前线程就会在要join的任务(等待队列)上进行阻塞等待，等任务完成后被唤醒。

 

 

 

最后：

       本篇通过一个ForkJoin任务的fork/join过程来分析代码，结合上一篇的话，已经涵盖了ForkJoinTask一个完整执行过程的相关代码了。下篇会做一个收尾工作，将本篇和上篇未涉及到的ForkJoin框架源码分析一下。 

 

 

评论

1 楼 在世界的中心呼喚愛 2016-02-06  

好文章，自己看源码遇到好多不理解的地方。这里都能找到。。。

源码分析从例子上出发，才好理解。。。
单个类分析，太晕了！