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 6bd8d208c3 fix(common): keep memory-limited queue accounting with
removals (#7496)
6bd8d208c3 is described below
commit 6bd8d208c3d9f689e20d5998fd55f11f15b701e1
Author: Burak Kalaycı <[email protected]>
AuthorDate: Sat Oct 10 11:21:34 2026 +0300
fix(common): keep memory-limited queue accounting with removals (#7496)
Seeded elements were not charged, so take() waited in release on an empty
budget. drainTo and iterator, removeIf, removeAll and retainAll dropped
elements without releasing their reservation.
Fixes #7461
---
.../MemoryLimitedLinkedBlockingQueue.java | 89 +++++++++++
...ryLimitedLinkedBlockingQueueAccountingTest.java | 174 +++++++++++++++++++++
2 files changed, 263 insertions(+)
diff --git
a/shenyu-common/src/main/java/org/apache/shenyu/common/concurrent/MemoryLimitedLinkedBlockingQueue.java
b/shenyu-common/src/main/java/org/apache/shenyu/common/concurrent/MemoryLimitedLinkedBlockingQueue.java
index ea6ec11b93..55252b94b6 100644
---
a/shenyu-common/src/main/java/org/apache/shenyu/common/concurrent/MemoryLimitedLinkedBlockingQueue.java
+++
b/shenyu-common/src/main/java/org/apache/shenyu/common/concurrent/MemoryLimitedLinkedBlockingQueue.java
@@ -18,9 +18,14 @@
package org.apache.shenyu.common.concurrent;
import java.lang.instrument.Instrumentation;
+import java.util.ArrayList;
import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Objects;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
+import java.util.function.Predicate;
/**
* Can completely solve the OOM problem caused by {@link
java.util.concurrent.LinkedBlockingQueue}.
@@ -49,6 +54,11 @@ public class MemoryLimitedLinkedBlockingQueue<E> extends
LinkedBlockingQueue<E>
final Instrumentation inst) {
super(c);
this.memoryLimiter = new MemoryLimiter(memoryLimit, inst);
+ for (final E element : this) {
+ if (!this.memoryLimiter.acquire(element)) {
+ throw new IllegalStateException("memory limit exceeded");
+ }
+ }
}
/**
@@ -133,6 +143,85 @@ public class MemoryLimitedLinkedBlockingQueue<E> extends
LinkedBlockingQueue<E>
return success;
}
+ @Override
+ public int drainTo(final Collection<? super E> c) {
+ return drainTo(c, Integer.MAX_VALUE);
+ }
+
+ @Override
+ public int drainTo(final Collection<? super E> c, final int maxElements) {
+ if (Objects.isNull(c)) {
+ throw new NullPointerException();
+ }
+ if (c == this) {
+ throw new IllegalArgumentException();
+ }
+ final List<E> drained = new ArrayList<>();
+ final int count = super.drainTo(drained, maxElements);
+ try {
+ for (final E element : drained) {
+ c.add(element);
+ }
+ } finally {
+ for (final E element : drained) {
+ memoryLimiter.release(element);
+ }
+ }
+ return count;
+ }
+
+ @Override
+ public Iterator<E> iterator() {
+ final Iterator<E> delegate = super.iterator();
+ return new Iterator<E>() {
+
+ private E current;
+
+ @Override
+ public boolean hasNext() {
+ return delegate.hasNext();
+ }
+
+ @Override
+ public E next() {
+ current = delegate.next();
+ return current;
+ }
+
+ @Override
+ public void remove() {
+ delegate.remove();
+ memoryLimiter.release(current);
+ }
+ };
+ }
+
+ @Override
+ public boolean removeIf(final Predicate<? super E> filter) {
+ Objects.requireNonNull(filter);
+ boolean modified = false;
+ final Iterator<E> elements = iterator();
+ while (elements.hasNext()) {
+ if (filter.test(elements.next())) {
+ elements.remove();
+ modified = true;
+ }
+ }
+ return modified;
+ }
+
+ @Override
+ public boolean removeAll(final Collection<?> c) {
+ Objects.requireNonNull(c);
+ return removeIf(c::contains);
+ }
+
+ @Override
+ public boolean retainAll(final Collection<?> c) {
+ Objects.requireNonNull(c);
+ return removeIf(element -> !c.contains(element));
+ }
+
@Override
public void clear() {
super.clear();
diff --git
a/shenyu-common/src/test/java/org/apache/shenyu/common/concurrent/MemoryLimitedLinkedBlockingQueueAccountingTest.java
b/shenyu-common/src/test/java/org/apache/shenyu/common/concurrent/MemoryLimitedLinkedBlockingQueueAccountingTest.java
new file mode 100644
index 0000000000..c770679e9d
--- /dev/null
+++
b/shenyu-common/src/test/java/org/apache/shenyu/common/concurrent/MemoryLimitedLinkedBlockingQueueAccountingTest.java
@@ -0,0 +1,174 @@
+/*
+ * 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.shenyu.common.concurrent;
+
+import net.bytebuddy.agent.ByteBuddyAgent;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import java.lang.instrument.Instrumentation;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Accounting coverage for seeded and bulk removals on {@link
MemoryLimitedLinkedBlockingQueue}.
+ */
+public class MemoryLimitedLinkedBlockingQueueAccountingTest {
+
+ private static Instrumentation instrumentation;
+
+ @BeforeAll
+ public static void initialInstrumentation() {
+ ByteBuddyAgent.install();
+ instrumentation = ByteBuddyAgent.getInstrumentation();
+ }
+
+ @Test
+ void testCollectionConstructorChargesSeedMemory() {
+ Integer testObject = 7;
+ long testObjectSize = instrumentation.getObjectSize(testObject);
+ MemoryLimitedLinkedBlockingQueue<Integer> queue = new
MemoryLimitedLinkedBlockingQueue<>(
+ List.of(testObject), testObjectSize * 4, instrumentation);
+ assertEquals(1, queue.size());
+ assertEquals(testObjectSize, queue.getCurrentMemory());
+ }
+
+ @Test
+ void testDrainToReleasesReservedMemory() {
+ Integer first = 1;
+ Integer second = 2;
+ long itemSize = instrumentation.getObjectSize(first);
+ MemoryLimitedLinkedBlockingQueue<Integer> queue = new
MemoryLimitedLinkedBlockingQueue<>(itemSize * 4, instrumentation);
+ assertTrue(queue.offer(first));
+ assertTrue(queue.offer(second));
+ List<Integer> drained = new ArrayList<>();
+ assertEquals(2, queue.drainTo(drained));
+ assertEquals(0, queue.size());
+ assertEquals(0, queue.getCurrentMemory());
+ assertTrue(queue.offer(first));
+ }
+
+ @Test
+ void testIteratorRemoveReleasesReservedMemory() {
+ Integer first = 1;
+ Integer second = 2;
+ long itemSize = instrumentation.getObjectSize(first);
+ MemoryLimitedLinkedBlockingQueue<Integer> queue = new
MemoryLimitedLinkedBlockingQueue<>(itemSize * 4, instrumentation);
+ assertTrue(queue.offer(first));
+ assertTrue(queue.offer(second));
+ Iterator<Integer> iterator = queue.iterator();
+ assertEquals(first, iterator.next());
+ iterator.remove();
+ assertEquals(1, queue.size());
+ assertEquals(itemSize, queue.getCurrentMemory());
+ }
+
+ @Test
+ void testRemoveIfReleasesReservedMemory() {
+ Integer first = 1;
+ Integer second = 2;
+ long itemSize = instrumentation.getObjectSize(first);
+ MemoryLimitedLinkedBlockingQueue<Integer> queue = new
MemoryLimitedLinkedBlockingQueue<>(itemSize * 4, instrumentation);
+ assertTrue(queue.offer(first));
+ assertTrue(queue.offer(second));
+ assertTrue(queue.removeIf(value -> value.equals(first)));
+ assertEquals(1, queue.size());
+ assertEquals(itemSize, queue.getCurrentMemory());
+ }
+
+ @Test
+ void testCollectionConstructorTakeAndPollReleaseMemory() throws
InterruptedException {
+ Integer testObject = 7;
+ long testObjectSize = instrumentation.getObjectSize(testObject);
+ MemoryLimitedLinkedBlockingQueue<Integer> taken = new
MemoryLimitedLinkedBlockingQueue<>(
+ List.of(testObject), testObjectSize * 4, instrumentation);
+ assertEquals(testObject, taken.take());
+ assertEquals(0, taken.getCurrentMemory());
+ assertTrue(taken.offer(testObject));
+
+ MemoryLimitedLinkedBlockingQueue<Integer> polled = new
MemoryLimitedLinkedBlockingQueue<>(
+ List.of(testObject), testObjectSize * 4, instrumentation);
+ assertEquals(testObject, polled.poll());
+ assertEquals(0, polled.getCurrentMemory());
+ assertEquals(0, polled.size());
+ }
+
+ @Test
+ void testCollectionConstructorRejectsOverBudgetSeed() {
+ Integer first = 1;
+ Integer second = 2;
+ long itemSize = instrumentation.getObjectSize(first);
+ assertThrows(IllegalStateException.class, () -> new
MemoryLimitedLinkedBlockingQueue<>(
+ List.of(first, second), itemSize + 1, instrumentation));
+ }
+
+ @Test
+ void testDrainToMaxElementsReleasesOnlyDrainedMemory() {
+ Integer first = 1;
+ Integer second = 2;
+ long itemSize = instrumentation.getObjectSize(first);
+ MemoryLimitedLinkedBlockingQueue<Integer> queue = new
MemoryLimitedLinkedBlockingQueue<>(itemSize * 4, instrumentation);
+ assertTrue(queue.offer(first));
+ assertTrue(queue.offer(second));
+ List<Integer> drained = new ArrayList<>();
+ assertEquals(1, queue.drainTo(drained, 1));
+ assertEquals(List.of(first), drained);
+ assertEquals(1, queue.size());
+ assertEquals(itemSize, queue.getCurrentMemory());
+ }
+
+ @Test
+ void testDrainToRejectsSelfAndLeavesAccountingUntouched() {
+ Integer testObject = 1;
+ long itemSize = instrumentation.getObjectSize(testObject);
+ MemoryLimitedLinkedBlockingQueue<Integer> queue = new
MemoryLimitedLinkedBlockingQueue<>(itemSize * 4, instrumentation);
+ assertTrue(queue.offer(testObject));
+ assertThrows(IllegalArgumentException.class, () ->
queue.drainTo(queue));
+ assertEquals(1, queue.size());
+ assertEquals(itemSize, queue.getCurrentMemory());
+ }
+
+ @Test
+ void testRemoveAllAndRetainAllReleaseMemory() {
+ Integer first = 1;
+ Integer second = 2;
+ long itemSize = instrumentation.getObjectSize(first);
+ MemoryLimitedLinkedBlockingQueue<Integer> removed = new
MemoryLimitedLinkedBlockingQueue<>(itemSize * 6, instrumentation);
+ assertTrue(removed.offer(first));
+ assertTrue(removed.offer(second));
+ assertTrue(removed.removeAll(List.of(first)));
+ assertEquals(1, removed.size());
+ assertEquals(itemSize, removed.getCurrentMemory());
+
+ MemoryLimitedLinkedBlockingQueue<Integer> retained = new
MemoryLimitedLinkedBlockingQueue<>(itemSize * 6, instrumentation);
+ assertTrue(retained.offer(first));
+ assertTrue(retained.offer(second));
+ Integer third = 3;
+ assertTrue(retained.offer(third));
+ assertTrue(retained.retainAll(List.of(second)));
+ assertEquals(1, retained.size());
+ assertEquals(itemSize, retained.getCurrentMemory());
+ assertEquals(second, retained.poll());
+ assertEquals(0, retained.getCurrentMemory());
+ }
+}
