Two operating modes are supported: + * + *
Note on keepAliveTime: non-core threads are kept alive for + * {@value #KEEP_ALIVE_SECONDS} seconds after becoming idle. This window must be long enough + * for the {@link ThreadLocal} reuse benefit to materialise. A value of {@code 0} would cause + * threads to terminate immediately and eliminate any reuse benefit. + *
The benchmark in issue #16042 / PR #16055 contained a synchronization bug: the timing code + * awaited {@code countDownLatch1} (the start gate) rather than {@code countDownLatch2} + * (the completion latch). This meant the elapsed time was measured before all tasks had + * finished, making the pooled executor appear faster than the non-pooled one even though both + * modes complete all tasks in roughly the same wall-clock time. + * + *
{@code
+ * CountDownLatch startGate = new CountDownLatch(1); // latch 1 - release all tasks together
+ * CountDownLatch completionLatch = new CountDownLatch(N); // latch 2 - await ALL task completions
+ *
+ * for (int i = 0; i < N; i++) {
+ * executor.execute(() -> {
+ * startGate.await(); // wait until everyone is ready
+ * doWork();
+ * completionLatch.countDown(); // signal completion
+ * });
+ * }
+ *
+ * long t0 = System.nanoTime();
+ * startGate.countDown(); // release all tasks simultaneously
+ * completionLatch.await(); // MUST await the COMPLETION latch, NOT the start gate
+ * long elapsed = System.nanoTime() - t0;
+ * }
+ *
+ * These tests verify that: + *
The start gate ({@code startGate}) releases all submitted tasks at the same time so they + * compete for the scheduler simultaneously. The completion latch ({@code completionLatch}) is + * decremented by every task when it finishes. Only after completionLatch reaches zero + * do we stop the clock, ensuring the elapsed time reflects actual end-to-end execution. + */ + @Test + @EnabledForJreRange(min = JRE.JAVA_21) + void unpooledExecutor_allTasksComplete_withCorrectTwoLatchPattern() throws InterruptedException { + URL url = URL.valueOf("dubbo://10.20.130.230:20880/context/path"); + ThreadPool threadPool = new VirtualThreadPool(); + Executor executor = threadPool.getExecutor(url); + + runBenchmark(executor, TASK_COUNT, "unpooled"); + } + + /** + * Verifies that the pooled executor also runs all tasks to completion when measured with the + * corrected two-latch pattern. + * + *
Previously, a flawed benchmark awaited the start gate a second time instead of the + * completion latch, returning immediately after releasing tasks. This test would have caught + * that bug because {@code completedTasks} would be far less than {@code TASK_COUNT}. + */ + @Test + @EnabledForJreRange(min = JRE.JAVA_21) + void pooledExecutor_allTasksComplete_withCorrectTwoLatchPattern() throws InterruptedException { + URL url = URL.valueOf("dubbo://10.20.130.230:20880/context/path?" + THREADS_VIRTUAL_CORE + "=" + + Runtime.getRuntime().availableProcessors()); + ThreadPool threadPool = new VirtualThreadPool(); + Executor executor = threadPool.getExecutor(url); + + runBenchmark(executor, TASK_COUNT, "pooled"); + } + + /** + * Validates the two-latch timing pattern itself: awaiting the completion latch means elapsed + * time is always >= the time to complete all tasks (trivially verifiable because the task + * counter equals {@code taskCount} when the method returns). + * + * @param executor the executor under test + * @param taskCount number of tasks to submit + * @param executorLabel human-readable label for assertion messages + */ + private static void runBenchmark(Executor executor, int taskCount, String executorLabel) + throws InterruptedException { + // latch 1: start gate - holds all tasks until released together (simulates concurrent load) + CountDownLatch startGate = new CountDownLatch(1); + // latch 2: completion latch - counts down when each task finishes + CountDownLatch completionLatch = new CountDownLatch(taskCount); + + AtomicInteger completedTasks = new AtomicInteger(0); + + for (int i = 0; i < taskCount; i++) { + executor.execute(() -> { + try { + // Wait until all tasks are queued and the start gate opens. + startGate.await(); + // Simulate a minimal unit of work (e.g. an RPC handler body). + simulateWork(); + completedTasks.incrementAndGet(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } finally { + // Always signal completion so the benchmark can drain. + // IMPORTANT: this must countDown on completionLatch (latch 2), + // NOT on startGate (latch 1). Decrementing latch 1 here was the + // bug in the original benchmark reported in issue #16174. + completionLatch.countDown(); + } + }); + } + + long startNanos = System.nanoTime(); + // Release all tasks simultaneously. + startGate.countDown(); + // Correct: await completionLatch (latch 2), NOT startGate (latch 1). + // Awaiting startGate here would return immediately (it is already at 0) and make + // the measurement appear artificially fast - exactly the flaw in #16174. + completionLatch.await(); + long elapsedNanos = System.nanoTime() - startNanos; + + // All tasks must have finished before we get here. + assertEquals( + taskCount, + completedTasks.get(), + executorLabel + " executor: expected all " + taskCount + + " tasks to complete before completionLatch.await() returned, " + + "but only " + completedTasks.get() + " finished. " + + "This indicates the benchmark awaited the wrong latch."); + + assertTrue(elapsedNanos > 0, executorLabel + " executor: elapsed time must be positive"); + } + + /** + * Simulates a lightweight unit of work inside each task. + * Intentionally minimal to keep the test fast while still being non-trivial. + */ + private static void simulateWork() { + // Trivial CPU work: a small loop that the JIT won't optimise away entirely. + int sum = 0; + for (int i = 0; i < 1000; i++) { + sum += i; + } + // Prevent dead-code elimination. + if (sum < 0) { + throw new IllegalStateException("impossible"); + } + } +} diff --git a/dubbo-plugin/dubbo-plugin-loom/src/test/java/org/apache/dubbo/common/threadpool/support/loom/VirtualThreadPoolTest.java b/dubbo-plugin/dubbo-plugin-loom/src/test/java/org/apache/dubbo/common/threadpool/support/loom/VirtualThreadPoolTest.java index 2f0ac0d916ea..3141033c6733 100644 --- a/dubbo-plugin/dubbo-plugin-loom/src/test/java/org/apache/dubbo/common/threadpool/support/loom/VirtualThreadPoolTest.java +++ b/dubbo-plugin/dubbo-plugin-loom/src/test/java/org/apache/dubbo/common/threadpool/support/loom/VirtualThreadPoolTest.java @@ -23,6 +23,9 @@ import java.util.concurrent.Executor; import java.util.concurrent.SynchronousQueue; import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; import org.hamcrest.Matchers; import org.junit.jupiter.api.Test; @@ -36,6 +39,7 @@ import static org.hamcrest.Matchers.instanceOf; import static org.hamcrest.Matchers.is; import static org.hamcrest.Matchers.startsWith; +import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertTrue; public class VirtualThreadPoolTest { @@ -83,6 +87,15 @@ void getExecutor3() throws Exception { assertThat(tpe.getMaximumPoolSize(), is(Integer.MAX_VALUE)); assertThat(tpe.getQueue(), instanceOf(SynchronousQueue.class)); + // Regression guard for issue #16174: keepAliveTime must be > 0 so that non-core + // virtual threads stay alive long enough to reuse ThreadLocal-cached state. + assertTrue( + tpe.getKeepAliveTime(TimeUnit.SECONDS) > 0, + "keepAliveTime must be > 0 to enable ThreadLocal reuse; " + + "a value of 0 causes threads to die immediately after each task, " + + "defeating the purpose of the pooled mode."); + assertEquals(VirtualThreadPool.KEEP_ALIVE_SECONDS, tpe.getKeepAliveTime(TimeUnit.SECONDS)); + final CountDownLatch latch = new CountDownLatch(1); executor.execute(() -> { Thread thread = Thread.currentThread(); @@ -94,4 +107,63 @@ void getExecutor3() throws Exception { latch.await(); assertThat(latch.getCount(), is(0L)); } + + /** + * Verifies that in pooled mode, a warm virtual thread can be reused for a second task, + * making ThreadLocal-cached values visible across consecutive submissions. + * + *
This is the key property that justifies the pooled mode (see issue #16042): libraries + * like FastJSON and Aerospike Java client store large byte-buffers in ThreadLocals. If threads + * are reused, those buffers survive across requests (reducing GC pressure). If every task + * gets a fresh thread the cache is useless. + * + *
The test submits two tasks sequentially to a pooled executor with corePoolSize=1.
+ * The first task sets a ThreadLocal value; the second task checks whether the same thread
+ * ran it and whether the ThreadLocal value is still present.
+ */
+ @Test
+ @EnabledForJreRange(min = JRE.JAVA_21)
+ void getExecutor4_threadLocalReuseInPooledMode() throws Exception {
+ URL url = URL.valueOf("dubbo://10.20.130.230:20880/context/path?" + THREADS_VIRTUAL_CORE + "=1&"
+ + THREAD_NAME_KEY + "=pool-reuse-test");
+ ThreadPool threadPool = new VirtualThreadPool();
+ Executor executor = threadPool.getExecutor(url);
+
+ // A ThreadLocal that the first task populates and the second task reads.
+ ThreadLocal