This is an automated email from the ASF dual-hosted git repository.
Aias00 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shenyu.git
The following commit(s) were added to refs/heads/master by this push:
new c5931a105c fix(common): restore eager task queue offer policy (#7482)
c5931a105c is described below
commit c5931a105cb1cb354d5eb24546850ad2bcabe73f
Author: lymerin <[email protected]>
AuthorDate: Sun Oct 11 16:37:30 2026 +0800
fix(common): restore eager task queue offer policy (#7482)
---
.../common/concurrent/MemoryLimitedTaskQueue.java | 5 ++
.../common/concurrent/MemorySafeTaskQueue.java | 5 ++
.../concurrent/ShenyuThreadPoolExecutorTest.java | 61 ++++++++++++++++++++--
3 files changed, 66 insertions(+), 5 deletions(-)
diff --git
a/shenyu-common/src/main/java/org/apache/shenyu/common/concurrent/MemoryLimitedTaskQueue.java
b/shenyu-common/src/main/java/org/apache/shenyu/common/concurrent/MemoryLimitedTaskQueue.java
index 5df59eb09e..efd70edd0a 100644
---
a/shenyu-common/src/main/java/org/apache/shenyu/common/concurrent/MemoryLimitedTaskQueue.java
+++
b/shenyu-common/src/main/java/org/apache/shenyu/common/concurrent/MemoryLimitedTaskQueue.java
@@ -50,6 +50,11 @@ public class MemoryLimitedTaskQueue<R extends Runnable>
extends MemoryLimitedLin
this.executor = executor;
}
+ @Override
+ public boolean offer(final Runnable runnable) {
+ return TaskQueue.super.offer(runnable);
+ }
+
@Override
public boolean doOffer(final Runnable runnable) {
return super.offer(runnable);
diff --git
a/shenyu-common/src/main/java/org/apache/shenyu/common/concurrent/MemorySafeTaskQueue.java
b/shenyu-common/src/main/java/org/apache/shenyu/common/concurrent/MemorySafeTaskQueue.java
index dca0ccfe46..ea52f6e7cd 100644
---
a/shenyu-common/src/main/java/org/apache/shenyu/common/concurrent/MemorySafeTaskQueue.java
+++
b/shenyu-common/src/main/java/org/apache/shenyu/common/concurrent/MemorySafeTaskQueue.java
@@ -50,6 +50,11 @@ public class MemorySafeTaskQueue<R extends Runnable> extends
MemorySafeLinkedBlo
this.executor = executor;
}
+ @Override
+ public boolean offer(final Runnable runnable) {
+ return TaskQueue.super.offer(runnable);
+ }
+
@Override
public boolean doOffer(final Runnable runnable) {
return super.offer(runnable);
diff --git
a/shenyu-common/src/test/java/org/apache/shenyu/common/concurrent/ShenyuThreadPoolExecutorTest.java
b/shenyu-common/src/test/java/org/apache/shenyu/common/concurrent/ShenyuThreadPoolExecutorTest.java
index a8b3e71ab7..b098120bed 100644
---
a/shenyu-common/src/test/java/org/apache/shenyu/common/concurrent/ShenyuThreadPoolExecutorTest.java
+++
b/shenyu-common/src/test/java/org/apache/shenyu/common/concurrent/ShenyuThreadPoolExecutorTest.java
@@ -20,11 +20,21 @@ package org.apache.shenyu.common.concurrent;
import net.bytebuddy.agent.ByteBuddyAgent;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
import java.lang.instrument.Instrumentation;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Future;
+import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
+import java.util.stream.Stream;
+import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
/**
* Test cases for ShenyuThreadPoolExecutor.
@@ -39,15 +49,56 @@ public class ShenyuThreadPoolExecutorTest {
instrumentation = ByteBuddyAgent.getInstrumentation();
}
- private ShenyuThreadPoolExecutor getTestExecutor(final TaskQueue<Runnable>
taskQueue) {
- return new ShenyuThreadPoolExecutor(5, 10, 100, TimeUnit.SECONDS,
taskQueue, ShenyuThreadFactory.create("Test", false), (r, e) -> {
- });
- }
-
@Test
public void testNullCommand() {
ShenyuThreadPoolExecutor executor = getTestExecutor(new
MemoryLimitedTaskQueue<>(instrumentation));
assertThrows(NullPointerException.class, () -> executor.execute(null));
}
+ @ParameterizedTest
+ @MethodSource("taskQueues")
+ public void testGrowsWhenCoreWorkerIsBusy(final TaskQueue<Runnable> queue)
throws Exception {
+ ShenyuThreadPoolExecutor executor = getTestExecutor(queue);
+ CountDownLatch started = new CountDownLatch(1);
+ CountDownLatch release = new CountDownLatch(1);
+ try {
+ final Future<?> first = executor.submit(() -> {
+ started.countDown();
+ release.await();
+ return null;
+ });
+ assertTrue(started.await(5, TimeUnit.SECONDS));
+ executor.submit(() -> { }).get(5, TimeUnit.SECONDS);
+ release.countDown();
+ first.get(5, TimeUnit.SECONDS);
+ } finally {
+ release.countDown();
+ executor.shutdownNow();
+ assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS));
+ }
+ }
+
+ @ParameterizedTest
+ @MethodSource("taskQueues")
+ public void testOffersToIdleWorker(final TaskQueue<Runnable> queue) {
+ EagerExecutorService executor = mock(EagerExecutorService.class);
+ when(executor.getPoolSize()).thenReturn(2);
+ when(executor.getActiveCount()).thenReturn(1);
+ when(executor.getMaximumPoolSize()).thenReturn(3);
+ queue.setExecutor(executor);
+ Runnable task = () -> { };
+
+ assertTrue(queue.offer(task));
+ assertSame(task, queue.poll());
+ }
+
+ private static Stream<TaskQueue<Runnable>> taskQueues() {
+ return Stream.of(new MemorySafeTaskQueue<>(1), new
MemoryLimitedTaskQueue<>(instrumentation));
+ }
+
+ private ShenyuThreadPoolExecutor getTestExecutor(final TaskQueue<Runnable>
queue) {
+ return new ShenyuThreadPoolExecutor(1, 2, 60, TimeUnit.SECONDS, queue,
+ ShenyuThreadFactory.create("Test", true), new
ThreadPoolExecutor.AbortPolicy());
+ }
+
}
