Java ForkJoinPool - 队列中的任务顺序

Question

我想了解在 Java fork-join 池中处理任务的顺序。

到目前为止，我在文档中找到的唯一相关信息是关于一个名为“asyncMode”的参数，“如果此池对 fork 任务使用本地先进先出调度模式，则该参数为真从未加入”。

我对这个说法的解释是每个worker都有自己的task queue； worker 从自己队列的前面接任务，或者如果他们自己的队列是空的，则从其他 worker 队列的后面偷走任务；如果 asyncMode 为 true(resp.false)，工作人员将新 fork 的任务添加到自己队列的后面(resp.front)。

如果我的理解有误，请指正!

现在，这提出了几个问题:

1) 加入的 fork 任务的顺序是什么？

我的猜测是，当任务被 fork 时，它会被添加到工作人员的队列中，如我在上面的解释中所述。现在，假设任务已加入...

• 如果在调用 join 时任务尚未开始，则调用 join 的 worker 会将任务从队列中拉出并立即开始处理。

• 如果在调用 join 时任务已经被另一个 worker 窃取，那么调用 join 的 worker 将同时处理其他任务(按照我上面解释中描述的获取任务的顺序)，直到它加入的任务被偷走它的工作人员完成。

这个猜测是基于用打印语句编写简单的测试代码，并观察改变连接调用顺序影响任务处理顺序的方式。有人可以告诉我我的猜测是否正确吗？

2)对外提交的任务顺序是什么？

根据 the answer to this question ，fork-join 池不使用外部队列。 (顺便说一句，我使用的是 Java 8。)

那么我是否理解当一个任务从外部提交时，该任务被添加到一个随机选择的工作队列中？

如果是，外部提交的任务是加到队尾还是队头？

最后，这是否取决于任务是通过调用pool.execute(task) 还是通过调用pool.invoke(task) 提交的？这是否取决于调用 pool.execute(task) 或 pool.invoke(task) 的线程是外部线程还是此 fork-join 池中的线程？

Answer 1

1. 你的猜测是正确的，你完全正确。正如您在“Implementation overview”中看到的那样。

 * Joining Tasks
 * =============
 *
 * Any of several actions may be taken when one worker is waiting
 * to join a task stolen (or always held) by another.  Because we
 * are multiplexing many tasks on to a pool of workers, we can't
 * just let them block (as in Thread.join).  We also cannot just
 * reassign the joiner's run-time stack with another and replace
 * it later, which would be a form of "continuation", that even if
 * possible is not necessarily a good idea since we may need both
 * an unblocked task and its continuation to progress.  Instead we
 * combine two tactics:
 *
 *   Helping: Arranging for the joiner to execute some task that it
 *      would be running if the steal had not occurred.
 *
 *   Compensating: Unless there are already enough live threads,
 *      method tryCompensate() may create or re-activate a spare
 *      thread to compensate for blocked joiners until they unblock.

2.ForkJoinPool.invoke和ForkJoinPool.join在任务提交方式上是完全一样的。代码中可以看到

    public <T> T invoke(ForkJoinTask<T> task) {
        if (task == null)
            throw new NullPointerException();
        externalPush(task);
        return task.join();
    }
    public void execute(ForkJoinTask<?> task) {
        if (task == null)
            throw new NullPointerException();
        externalPush(task);
    }

在 externalPush 中，您可以看到使用 ThreadLocalRandom 将任务添加到随机选择的工作队列中。而且，它是用push的方式进入到队头的。

    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)) {
            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);
                U.putIntVolatile(q, QLOCK, 0);
                if (n <= 1)
                    signalWork(ws, q);
                return;
            }
            U.compareAndSwapInt(q, QLOCK, 1, 0);
        }
        externalSubmit(task);
    }

我不知道你说的是什么意思:

And does this depend on whether the thread calling pool.execute(task) or pool.invoke(task) is an external thread or a thread within this fork-join pool?