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 943f348117 fix(common): wake producers after memory limit changes and
reset (#7495)
943f348117 is described below
commit 943f348117a164ed77b9517f234eae96cb93f58a
Author: lymerin <[email protected]>
AuthorDate: Sat Oct 10 16:47:34 2026 +0800
fix(common): wake producers after memory limit changes and reset (#7495)
---
.../shenyu/common/concurrent/MemoryLimiter.java | 15 ++--
.../MemoryLimitedLinkedBlockingQueueTest.java | 100 +++++++++++++++++++++
2 files changed, 110 insertions(+), 5 deletions(-)
diff --git
a/shenyu-common/src/main/java/org/apache/shenyu/common/concurrent/MemoryLimiter.java
b/shenyu-common/src/main/java/org/apache/shenyu/common/concurrent/MemoryLimiter.java
index d771ef24ff..a3e8a9db8e 100644
---
a/shenyu-common/src/main/java/org/apache/shenyu/common/concurrent/MemoryLimiter.java
+++
b/shenyu-common/src/main/java/org/apache/shenyu/common/concurrent/MemoryLimiter.java
@@ -33,7 +33,7 @@ public class MemoryLimiter {
private final Instrumentation inst;
- private long memoryLimit;
+ private volatile long memoryLimit;
private final LongAdder memory = new LongAdder();
@@ -66,7 +66,13 @@ public class MemoryLimiter {
if (memoryLimit <= 0) {
throw new IllegalArgumentException();
}
- this.memoryLimit = memoryLimit;
+ acquireLock.lock();
+ try {
+ this.memoryLimit = memoryLimit;
+ notLimited.signalAll();
+ } finally {
+ acquireLock.unlock();
+ }
}
/**
@@ -332,9 +338,8 @@ public class MemoryLimiter {
public void reset() {
fullyLock();
try {
- if (memory.sumThenReset() < memoryLimit) {
- notLimited.signal();
- }
+ memory.reset();
+ notLimited.signalAll();
} finally {
fullyUnlock();
}
diff --git
a/shenyu-common/src/test/java/org/apache/shenyu/common/concurrent/MemoryLimitedLinkedBlockingQueueTest.java
b/shenyu-common/src/test/java/org/apache/shenyu/common/concurrent/MemoryLimitedLinkedBlockingQueueTest.java
index fc604efdee..994b29f761 100644
---
a/shenyu-common/src/test/java/org/apache/shenyu/common/concurrent/MemoryLimitedLinkedBlockingQueueTest.java
+++
b/shenyu-common/src/test/java/org/apache/shenyu/common/concurrent/MemoryLimitedLinkedBlockingQueueTest.java
@@ -18,6 +18,7 @@
package org.apache.shenyu.common.concurrent;
import net.bytebuddy.agent.ByteBuddyAgent;
+import org.apache.commons.lang3.reflect.FieldUtils;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
@@ -27,13 +28,19 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.ReentrantLock;
+import static org.awaitility.Awaitility.await;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.is;
+import static org.junit.jupiter.api.Assertions.assertAll;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
/**
* Test cases for MemoryLimitedLinkedBlcokingQueue.
@@ -92,6 +99,85 @@ public class MemoryLimitedLinkedBlockingQueueTest {
assertEquals(0, queue.getCurrentMemory());
}
+ @Test
+ void testIncreaseMemoryLimitWakesEligibleProducer() throws Exception {
+ Instrumentation sizedInstrumentation = mock(Instrumentation.class);
+ Object small = new Object();
+ Object large = new Object();
+ when(sizedInstrumentation.getObjectSize(small)).thenReturn(20L);
+ when(sizedInstrumentation.getObjectSize(large)).thenReturn(50L);
+ MemoryLimitedLinkedBlockingQueue<Object> queue = new
MemoryLimitedLinkedBlockingQueue<>(41, sizedInstrumentation);
+ queue.put(small);
+ queue.put(small);
+ ExecutorService executor = Executors.newFixedThreadPool(2);
+ assertAll(
+ () -> {
+ try {
+ final Future<Boolean> largePut = executor.submit(() ->
{
+ queue.put(large);
+ return true;
+ });
+ awaitWaitingProducers(queue, 1);
+ final Future<Boolean> smallPut = executor.submit(() ->
{
+ queue.put(small);
+ return true;
+ });
+ awaitWaitingProducers(queue, 2);
+
+ queue.setMemoryLimit(61);
+
+ assertTrue(smallPut.get(5, TimeUnit.SECONDS));
+ awaitWaitingProducers(queue, 1);
+ assertFalse(largePut.isDone());
+ assertEquals(60, queue.getCurrentMemory());
+ assertEquals(3, queue.size());
+ } finally {
+ executor.shutdownNow();
+ }
+ },
+ () -> assertTrue(executor.awaitTermination(5,
TimeUnit.SECONDS)));
+ }
+
+ @Test
+ void testClearAfterShrinkingMemoryLimitWakesEligibleProducer() throws
Exception {
+ Instrumentation sizedInstrumentation = mock(Instrumentation.class);
+ Object small = new Object();
+ Object large = new Object();
+ when(sizedInstrumentation.getObjectSize(small)).thenReturn(20L);
+ when(sizedInstrumentation.getObjectSize(large)).thenReturn(50L);
+ MemoryLimitedLinkedBlockingQueue<Object> queue = new
MemoryLimitedLinkedBlockingQueue<>(61, sizedInstrumentation);
+ queue.put(small);
+ queue.put(small);
+ queue.setMemoryLimit(21);
+ ExecutorService executor = Executors.newFixedThreadPool(2);
+ assertAll(
+ () -> {
+ try {
+ final Future<Boolean> largePut = executor.submit(() ->
{
+ queue.put(large);
+ return true;
+ });
+ awaitWaitingProducers(queue, 1);
+ final Future<Boolean> smallPut = executor.submit(() ->
{
+ queue.put(small);
+ return true;
+ });
+ awaitWaitingProducers(queue, 2);
+
+ queue.clear();
+
+ assertTrue(smallPut.get(5, TimeUnit.SECONDS));
+ awaitWaitingProducers(queue, 1);
+ assertFalse(largePut.isDone());
+ assertEquals(20, queue.getCurrentMemory());
+ assertEquals(1, queue.size());
+ } finally {
+ executor.shutdownNow();
+ }
+ },
+ () -> assertTrue(executor.awaitTermination(5,
TimeUnit.SECONDS)));
+ }
+
@Test
public void testPut() throws InterruptedException, ExecutionException {
Integer testObject = 0;
@@ -141,4 +227,18 @@ public class MemoryLimitedLinkedBlockingQueueTest {
assertTrue(queue.remove(testObject));
assertEquals(0, queue.getCurrentMemory());
}
+
+ private void awaitWaitingProducers(final
MemoryLimitedLinkedBlockingQueue<?> queue, final int count) throws
IllegalAccessException {
+ MemoryLimiter limiter = (MemoryLimiter) FieldUtils.readField(queue,
"memoryLimiter", true);
+ ReentrantLock lock = (ReentrantLock) FieldUtils.readField(limiter,
"acquireLock", true);
+ Condition condition = (Condition) FieldUtils.readField(limiter,
"notLimited", true);
+ await().atMost(5, TimeUnit.SECONDS).until(() -> {
+ lock.lock();
+ try {
+ return lock.getWaitQueueLength(condition) == count;
+ } finally {
+ lock.unlock();
+ }
+ });
+ }
}
