loom的jtreg测试YieldQueueing.java无法移植到KonaFiber-8的原因分析

[quadhier]

YieldQueueing.java 移植情况说明

本人在将 loom 中的 jtreg 测试用例 YieldQueueing.java 移植到 KonaFiber-8 的过程中，发现由于两个 JDK 的 ForkJoinPool 和 VirtualThread 类的实现不同，移植似乎无法完成。下面详细分析一下无法移植的原因。

为了简洁性，这里只展示 jtreg 的 test description 部分和两个测试函数的代码，完整的测试用例可以在这里找到。

1. jtreg 的 test description 部分

/**
 * @test
 * @summary Test Thread.yield submits the virtual thread task to the expected queue
 * @requires vm.continuations
 * @run junit/othervm -Djdk.virtualThreadScheduler.maxPoolSize=1 YieldQueuing
 */

通过这里可以看出，在 loom 中使用 java 命令运行该测试用例时，需要添加 -Djdk.virtualThreadScheduler.maxPoolSize=1 属性，使得 VirtualThread 使用的默认调度器 ForkJoinPool 中只有一个 worker thread（也就是 VirtualThread 的 carrier thread 只有一个）。YieldQueuing.java 包含了两个测试函数 testYieldWithEmptyLocalQueue() 和 testYieldWithNonEmptyLocalQueue()，下面分别分析他们在 loom 和 KonaFiber-8 上的执行结果。

2. 对于 testYieldWithEmptyLocalQueue() 的分析

@Test
void testYieldWithEmptyLocalQueue() throws Exception {
    var list = new CopyOnWriteArrayList<String>();

    var threadsStarted = new AtomicBoolean();

    var threadA = Thread.ofVirtual().unstarted(() -> {
        // pin thread until task for B is in submission queue
        while (!threadsStarted.get()) {
            Thread.onSpinWait();
        }

        list.add("A");
        Thread.yield();      // push task for A to submission queue, B should run
        list.add("A");
    });

    var threadB = Thread.ofVirtual().unstarted(() -> {
        list.add("B");
    });

    // push tasks for A and B to submission queue
    threadA.start();
    threadB.start();

    // release A
    threadsStarted.set(true);

    // wait for result
    threadA.join();
    threadB.join();
    assertEquals(list, List.of("A", "B", "A"));
}

2.1 在 loom 和 KonaFiber-8 上的执行结果差异

这个测试用例是在用两个协程向一个列表中添加对象，预期最终列表中的元素总是依次为：“A”，“B”，“A”。在 loom 中确实是这样，但是在 KonaFiber-8 中我们观察到，第一元素总是 A，但是后面的两个元素 A 和 B 的相对顺序不固定，即可能为 AAB 也有可能是 ABA。

2.2 该测试在 loom 中的执行过程分析

2.2节的分析均基于 loom 的源码

VirtualThread 默认使用 ForkJoinPool 作为调度器。通过阅读 VirtualThread 的源码可以看出，这里的代码等价于向 ForkJoinPool 提交了三个任务（分别在 threadA.start()，threadB.start() 和 Thread.yield() 处）。我们用表格来记录（表格中的行顺序与任务的提交顺序没有必然联系）：

任务名称	任务内容	提交时所在的线程
A1	list.add("A")	main thread
B1	list.add("B")	main thread
A2	list.add("A")	worker thread

ForkJoinPool 中为了保存已经提交的任务，使用了一个属性 queues，它是一个以 WorkQueue 类型对象为元素的数组，其中每个 WorkQueue 都是一个任务的队列。数组中偶数下标的任务队列为公共的 "submission queue"，为所有线程“共享”；奇数下标的任务队列为 "local queue"，为一些线程“独享”。

Thread.start() 在提交任务时，会通过 VirtualThread.submitRunContinuation() 去最终调用到 ForkJoinPool.poolSubmit() 来提交任务。

ForkJoinPool.poolSubmit() 在处理任务提交时，会做一个判断，如果当前提交任务的线程是 worker thread，则会尝试把这个任务提交到它的 local queue 中，否则提交到某个 submission queue 中。这段逻辑对应的代码如下：（相应的GitHub 链接在这里）

Code 1. ForkJoinPool 任务提交过程

/**
 * Pushes a submission to the pool, using internal queue if called
 * from ForkJoinWorkerThread, else external queue.
 */
private <T> ForkJoinTask<T> poolSubmit(boolean signalIfEmpty,
                                       ForkJoinTask<T> task) {
    WorkQueue q; Thread t; ForkJoinWorkerThread wt;
    U.storeStoreFence();  // ensure safely publishable
    if (task == null) throw new NullPointerException();
    if (((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) &&
        (wt = (ForkJoinWorkerThread)t).pool == this)
        q = wt.workQueue;
    else {
        task.markPoolSubmission();
        q = submissionQueue(true);
    }
    q.push(task, this, signalIfEmpty);
    return task;
}

在这个测试函数中，由于任务 A1 和 B1 从 main thread 中提交，他们会被提交到同一个 submission queue 中。

Thread.yield() 在提交任务时，它会判断当前提交任务的线程是否是 worker thread，如果是，并且所有的 local queue 都为空，则直接提交到 ForkJoinPool 的某个 submission queue。如果上述情况不满足，则也通过 VirtualThread.submitRunContinuation() 去最终调用到 ForkJoinPool.poolSubmit() 去处理这个任务的提交。这段逻辑在 VirtualThread.afterYield() 函数中：（相应的GitHub 链接在这里）

Code 2. VirtualThread 在 yield 后提交任务的过程

// external submit if there are no tasks in the local task queue
if (currentThread() instanceof CarrierThread ct && ct.getQueuedTaskCount() == 0) {
    externalSubmitRunContinuation(ct.getPool());
} else {
    submitRunContinuation();
}

在这个测试函数中，A2 在 worker thread 被提交，但由于所有的 local queue 都是空的，A2 会被提交到某个 submission queue中。

在 worker thread 尝试取任务和执行任务时，他会先扫描 submission queue，找到一个任务并执行后，会将自己的 local queue 中的任务全部执行，再将刚扫描到的任务所在的 submission queue 中的任务全部执行。这段逻辑对应的代码如下：（相应的GitHub 链接在这里）

Code 3. Worker thread 取到任务后执行过程

/**
 * Runs the given (stolen) task if nonnull, as well as
 * remaining local tasks and/or others available from its
 * source queue, if any.
 */
final void topLevelExec(ForkJoinTask<?> task, WorkQueue src) {
    int cfg = config, fifo = cfg & FIFO, nstolen = 1;
    while (task != null) {
        task.doExec();
        if ((task = nextLocalTask(fifo)) == null &&
            src != null && (task = src.tryPoll()) != null)
            ++nstolen;
    }
    nsteals += nstolen;
    source = 0;
    if ((cfg & CLEAR_TLS) != 0)
        ThreadLocalRandom.eraseThreadLocals(Thread.currentThread());
}

在这个测试函数中，A1 被从 submission queue 中取到和执行后，worker thread 会继续找到 A1 所在的 submission queue 中的 B1 并执行，接着再去扫描所有 submission queue 并找到和执行 A2。

所以这个测试函数的执行过程是这样的：

1. 任务 A1 和 B1 在主线程中被提交，他们被提交到同一个 submission queue 中（见 Code 1）
2. 然后 worker thread 去执行，先扫描 queues 数组，找到 A1 和 B1 所在的 submission queue，取到任务 A1，执行它的过程中会提交 A2。因为此时所有的 local queue 都是空的，根据上面的分析，A2 被提交到一个某个 submission queue 中（见 Code 2）
3. 接着 worker thread 继续先尝试从自己的 local queue 中取任务；没有，继续从取到 A1 的这个 submission queue 中去取任务，取到了 B1 并执行（见 Code 3）
4. 最后 worker thread 继续扫描整个 queues 数组，找到 A2 并执行

所以这三个任务的执行顺序是 A1 -> B1 -> A2，即最终 list 中的元素依次是：“A”，“B”，“A”。

loom 中的这个测试用例就是想验证在上述的第 2 步中，A2 的确被提交到了 submission queue 中。

2.3 该测试在 KonaFiber-8 中的执行情况分析

KonaFiber-8 中的 ForkJoinPool 在处理提交的任务时，总是将任务提交到 submission queue 中。这样一来，A1 和 B1 会被提交到同一个 submission queue 中，并且 A1 的确最先得到执行。但是执行 A1 过程中提交的 A2 可能会被提交到另一个 submission queue 中。由于 worker thread 扫描 submission queue 的起点是随机的，可能先找到并执行 B1 或 A2 中的任意一个，导致这两个任务的相对执行顺序不确定。

3. 对于 testYieldWithNonEmptyLocalQueue() 的分析

@Test
void testYieldWithNonEmptyLocalQueue() throws Exception {
    var list = new CopyOnWriteArrayList<String>();

    var threadsStarted = new AtomicBoolean();

    var threadA = Thread.ofVirtual().unstarted(() -> {
        // pin thread until tasks for B and C are in submission queue
        while (!threadsStarted.get()) {
            Thread.onSpinWait();
        }

        list.add("A");
        LockSupport.park();   // B should run
        list.add("A");
    });

    var threadB = Thread.ofVirtual().unstarted(() -> {
        list.add("B");
        LockSupport.unpark(threadA);  // push task for A to local queue
        Thread.yield();               // push task for B to local queue, A should run
        list.add("B");
    });

    var threadC = Thread.ofVirtual().unstarted(() -> {
        list.add("C");
    });

    // push tasks for A, B and C to submission queue
    threadA.start();
    threadB.start();
    threadC.start();

    // release A
    threadsStarted.set(true);

    // wait for result
    threadA.join();
    threadB.join();
    threadC.join();
    assertEquals(list, List.of("A", "B", "A", "B", "C"));
}

3.1 在 loom 和 KonaFiber-8 上的执行结果差异

这个测试用例是在用三个协程向一个列表中添加对象，预期最终列表中的元素总是依次为：“A”，“B”，“A”，“B”，“C”。在 loom 中确实是这样，但是在 KonaFiber-8 中我们观察到，前两个元素总是AB，但是后面三个元素的顺序出现了多种情况。

3.2 该测试在 loom 中的执行过程分析

3.2节的分析均基于 loom 的源码

通过阅读 VirtualThread 的源码可以看出，这里的代码等价于向 ForkJoinPool 提交了五个任务（分别在 threadA.start()，threadB.start()，threadC.start()，LockSupport.unpark(threadA) 和 Thread.yield() 处）。我们用表格来记录（表格中的行顺序与任务的提交顺序没有必然联系）：

任务名称	任务内容	提交线程
A1	list.add("A")	main thread
B1	list.add("B")	main thread
C1	list.add("C")	main thread
A2	list.add("A")	worker thread
B2	list.add("B")	worker thread

在 LockSupport.unpark() 会调用到 VirtualThread.unpark()，后者会通过 VirtualThread.submitRunContinuation() 最终调用到 ForkJoinPool.poolSubmit() 方法来提交任务。

所以这个函数的执行过程是这样的：

1. 前三次任务是在主线程中提交的，他们被提交到同一个 submission queue 中（见 Code 1）
2. A2 在 worker thread 中提交，它通过被 LockSupport.unpark() 间接调用的 ForkJoinPool.poolSubmit() 方法提交，被直接提交到 worker thread 的 local queue 中（见 Code 1）
3. B2 也在 worker thread 中提交，由于 local queue 不为空，通过被 VirtualThread.submitRunContinuation() 间接调用的 ForkJoinPool.poolSubmit() 提交，被放入同样的 local queue 中（见 Code 2 和 Code 1）

所以这五个任务执行的顺序是 A1 -> B1 -> A2 -> B2 -> C1，即最终 list 中的元素依次是：“A”，“B”，“A”，“B”，“C”。

loom 中的这个测试用例就是想验证在上述的第 2 和 3 步中，任务 A2 和 B2 的确被提交到了 local queue 中。

3.3 该测试在 KonaFiber-8 中的执行情况分析

KonaFiber-8 中的 ForkJoinPool 在处理提交的任务时，总是将任务提交到 submission queue 中。这样一来，前三个任务的确被提交到同一个 submission queue 中，并且 A1，B1 总是最先被按顺序执行。但是在执行 B1 过程中提交的 A2 和 B2，可能会被提交到其他的 submission queue 中。又因为 worker thread 扫描和执行所有的 submission queue 的顺序是不固定的，导致后三个任务即 C1，A2 和 B2 的执行顺序不固定。