Fix VirtualThreadPool benchmark methodology issue (#16174)

dubbo-plugin/dubbo-plugin-loom/src/main/java/org/apache/dubbo/common/threadpool/support/loom/VirtualThreadPool.java

@@ -23,28 +23,69 @@
 import java.util.concurrent.Executors;
 import java.util.concurrent.SynchronousQueue;
 import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
 
 import static org.apache.dubbo.common.constants.CommonConstants.DEFAULT_THREAD_NAME;
 import static org.apache.dubbo.common.constants.CommonConstants.THREADS_VIRTUAL_CORE;
 import static org.apache.dubbo.common.constants.CommonConstants.THREAD_NAME_KEY;
 
 /**
- * Creates a thread pool that use virtual thread
+ * Creates a thread pool that uses virtual threads (Project Loom).
+ *
+ * <p>Two operating modes are supported:
+ *
+ * <ul>
+ *   <li><b>Unpooled mode (default)</b>: When {@code threads.virtual.core} is not set (or {@code
+ *       <= 0}), a new virtual thread is created for every submitted task via {@link
+ *       Executors#newThreadPerTaskExecutor(java.util.concurrent.ThreadFactory)}. This is the
+ *       simplest and most scalable mode.
+ *
+ *   <li><b>Pooled mode</b>: When {@code threads.virtual.core} is set to a positive value, a
+ *       {@link ThreadPoolExecutor} backed by virtual threads is used with {@code corePoolSize =
+ *       threads.virtual.core} and an unbounded maximum. This mode is beneficial when downstream
+ *       libraries (e.g. FastJSON, Aerospike Java client) store large byte-buffers in
+ *       {@link ThreadLocal} caches: by keeping a pool of warm virtual threads alive, those cached
+ *       buffers can be reused across consecutive tasks instead of being re-allocated on every
+ *       call.
+ *
+ *       <p><b>Note on keepAliveTime</b>: 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.
+ * </ul>
  *
  * @see Executors#newVirtualThreadPerTaskExecutor()
  */
 public class VirtualThreadPool implements ThreadPool {
+
+    /**
+     * Number of seconds that excess (non-core) virtual threads are kept alive while idle.
+     * Must be greater than zero to allow {@link ThreadLocal} state to be reused across tasks.
+     */
+    static final long KEEP_ALIVE_SECONDS = 60L;
+
     @Override
     public Executor getExecutor(URL url) {
         String name =
                 url.getParameter(THREAD_NAME_KEY, (String) url.getAttribute(THREAD_NAME_KEY, DEFAULT_THREAD_NAME));
         int threads = url.getParameter(THREADS_VIRTUAL_CORE, 0);
         if (threads > 0) {
+            /*
+             * Pooled virtual-thread executor.
+             *
+             * corePoolSize   = threads   (warm pool of reusable virtual threads)
+             * maximumPoolSize = MAX_VALUE  (auto-expand under burst load, just like unpooled mode)
+             * keepAliveTime  = 60s        (non-core threads stay alive long enough to reuse
+             *                              ThreadLocal-cached buffers before being reclaimed)
+             * workQueue      = SynchronousQueue (no internal buffering; tasks are handed off
+             *                              directly to a thread, matching the behaviour of
+             *                              newThreadPerTaskExecutor under light load)
+             */
             return new ThreadPoolExecutor(
                     threads,
                     Integer.MAX_VALUE,
-                    0L,
-                    java.util.concurrent.TimeUnit.MILLISECONDS,
+                    KEEP_ALIVE_SECONDS,
+                    TimeUnit.SECONDS,
                     new SynchronousQueue<>(),
                     Thread.ofVirtual().name(name, 1).factory());
         } else {

dubbo-plugin/dubbo-plugin-loom/src/test/java/org/apache/dubbo/common/threadpool/support/loom/VirtualThreadPoolBenchmarkTest.java

@@ -0,0 +1,188 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.dubbo.common.threadpool.support.loom;
+
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.threadpool.ThreadPool;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executor;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.EnabledForJreRange;
+import org.junit.jupiter.api.condition.JRE;
+
+import static org.apache.dubbo.common.constants.CommonConstants.THREADS_VIRTUAL_CORE;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Benchmark-methodology correctness tests for {@link VirtualThreadPool}.
+ *
+ * <h2>Background (issue #16174)</h2>
+ *
+ * <p>The benchmark in issue #16042 / PR #16055 contained a synchronization bug: the timing code
+ * awaited {@code countDownLatch1} (the <em>start gate</em>) rather than {@code countDownLatch2}
+ * (the <em>completion latch</em>). 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.
+ *
+ * <h2>Correct two-latch pattern</h2>
+ *
+ * <pre>{@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;
+ * }</pre>
+ *
+ * <p>These tests verify that:
+ * <ol>
+ *   <li>Both pooled and non-pooled executors complete <em>all</em> tasks before the timing window
+ *       closes (i.e. they never return early due to the wrong latch being awaited).
+ *   <li>The task completion count exactly equals the number of submitted tasks in both modes.
+ * </ol>
+ */
+public class VirtualThreadPoolBenchmarkTest {
+
+    private static final int TASK_COUNT = 200;
+
+    /**
+     * Verifies that the non-pooled (default) executor runs all tasks to completion when measured
+     * with the corrected two-latch pattern.
+     *
+     * <p>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 <em>completionLatch</em> 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.
+     *
+     * <p>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");
+        }
+    }
+}

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.
+     *
+     * <p>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.
+     *
+     * <p>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<String> tl = new ThreadLocal<>();
+
+        AtomicReference<Thread> firstThread = new AtomicReference<>();
+
+        // Task 1: record the executing thread and set a ThreadLocal value.
+        CountDownLatch task1Done = new CountDownLatch(1);
+        executor.execute(() -> {
+            firstThread.set(Thread.currentThread());
+            tl.set("cached-value");
+            task1Done.countDown();
+        });
+        task1Done.await();
+
+        // Task 2: check whether the same thread is reused and the ThreadLocal survived.
+        AtomicBoolean sameThread = new AtomicBoolean(false);
+        AtomicBoolean tlValuePresent = new AtomicBoolean(false);
+
+        CountDownLatch task2Done = new CountDownLatch(1);
+        executor.execute(() -> {
+            sameThread.set(Thread.currentThread() == firstThread.get());
+            tlValuePresent.set("cached-value".equals(tl.get()));
+            task2Done.countDown();
+        });
+        task2Done.await();
+
+        // With corePoolSize=1 and a 60-second keepAlive, the same thread should be reused
+        // for two back-to-back tasks, making the ThreadLocal value visible in task 2.
+        assertTrue(
+                sameThread.get(),
+                "Pooled executor with corePoolSize=1 should reuse the same thread for "
+                        + "back-to-back tasks, enabling ThreadLocal cache reuse.");
+        assertTrue(
+                tlValuePresent.get(),
+                "ThreadLocal value set by task 1 should be visible in task 2 when the "
+                        + "same thread is reused (the core rationale for pooled mode).");
+    }
 }