Copilot commented on code in PR #16370:
URL: https://github.com/apache/dubbo/pull/16370#discussion_r3540902980
##########
dubbo-plugin/dubbo-plugin-loom/src/test/java/org/apache/dubbo/common/threadpool/support/loom/VirtualThreadPoolTest.java:
##########
@@ -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);
+
Review Comment:
`ThreadPoolExecutor` with a `SynchronousQueue` can still create a new
(non-core) thread for task #2 if the existing core worker hasn't reached the
blocking `take()` yet when `execute()` is called. That makes this test's
implicit assumption (same thread reused) timing-dependent and potentially flaky.
To make the test deterministic, cap `maximumPoolSize` to `1` for this test
instance so the pool cannot spawn an extra thread between the two submissions.
##########
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");
+ }
Review Comment:
The executor returned by `VirtualThreadPool#getExecutor` is an
`ExecutorService` (either `ThreadPoolExecutor` or JDK's
`ThreadPerTaskExecutor`). Not closing it can leave pooled virtual threads and
ThreadLocal state alive beyond the test, which may interfere with other tests
and increases resource usage.
Use try-with-resources to ensure the executor is closed after the benchmark
run.
##########
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();
Review Comment:
`completionLatch.await()` has no timeout, so if something regresses (e.g.,
tasks never start, rejection, deadlock), this test can hang the whole suite
instead of failing quickly. Adding a bounded await keeps CI reliable while
still validating the two-latch pattern.
##########
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");
+ }
Review Comment:
The pooled-mode test also creates an `ExecutorService` that should be closed
to avoid leaving a live `ThreadPoolExecutor` (and its core virtual threads)
running after the test completes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
