This is an automated email from the ASF dual-hosted git repository.
xiaoyu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-shenyu.git
The following commit(s) were added to refs/heads/master by this push:
new a3909ea [ISSUE #3026] part1: support MemoryLimitedLinkedBlockingQueue
(#3043)
a3909ea is described below
commit a3909eaa0a0274f80d335367d57c6f180e0406a0
Author: dragon-zhang <[email protected]>
AuthorDate: Wed Mar 16 10:02:25 2022 +0800
[ISSUE #3026] part1: support MemoryLimitedLinkedBlockingQueue (#3043)
* [ISSUE #3026] part1: support MemoryLimitedLinkedBlockingQueue
* fix code style
* fix code style
* fix code style
* fix code style
---
.../MemoryLimitedLinkedBlockingQueue.java | 138 +++++++++
.../shenyu/common/concurrent/MemoryLimiter.java | 340 +++++++++++++++++++++
.../MemoryLimitedLinkedBlockingQueueTest.java | 43 +++
3 files changed, 521 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
new file mode 100644
index 0000000..dc64d99
--- /dev/null
+++
b/shenyu-common/src/main/java/org/apache/shenyu/common/concurrent/MemoryLimitedLinkedBlockingQueue.java
@@ -0,0 +1,138 @@
+/*
+ * 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 java.lang.instrument.Instrumentation;
+import java.util.Collection;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Can completely solve the OOM problem caused by {@link
java.util.concurrent.LinkedBlockingQueue}.
+ */
+public class MemoryLimitedLinkedBlockingQueue<E> extends
LinkedBlockingQueue<E> {
+
+ private static final long serialVersionUID = -6106022470621447542L;
+
+ private final MemoryLimiter memoryLimiter;
+
+ public MemoryLimitedLinkedBlockingQueue(final Instrumentation inst) {
+ this(Integer.MAX_VALUE, inst);
+ }
+
+ public MemoryLimitedLinkedBlockingQueue(final long memoryLimit,
+ final Instrumentation inst) {
+ super(Integer.MAX_VALUE);
+ this.memoryLimiter = new MemoryLimiter(memoryLimit, inst);
+ }
+
+ public MemoryLimitedLinkedBlockingQueue(final Collection<? extends E> c,
+ final long memoryLimit,
+ final Instrumentation inst) {
+ super(c);
+ this.memoryLimiter = new MemoryLimiter(memoryLimit, inst);
+ }
+
+ /**
+ * set the memory limit.
+ *
+ * @param memoryLimit the memory limit
+ */
+ public void setMemoryLimit(final long memoryLimit) {
+ memoryLimiter.setMemoryLimit(memoryLimit);
+ }
+
+ /**
+ * get the memory limit.
+ *
+ * @return the memory limit
+ */
+ public long getMemoryLimit() {
+ return memoryLimiter.getMemoryLimit();
+ }
+
+ /**
+ * get the current memory.
+ *
+ * @return the current memory
+ */
+ public long getCurrentMemory() {
+ return memoryLimiter.getCurrentMemory();
+ }
+
+ /**
+ * get the current remain memory.
+ *
+ * @return the current remain memory
+ */
+ public long getCurrentRemainMemory() {
+ return memoryLimiter.getCurrentRemainMemory();
+ }
+
+ @Override
+ public void put(final E e) throws InterruptedException {
+ memoryLimiter.acquireInterruptibly(e);
+ super.put(e);
+ }
+
+ @Override
+ public boolean offer(final E e, final long timeout, final TimeUnit unit)
throws InterruptedException {
+ return memoryLimiter.acquire(e, timeout, unit) && super.offer(e,
timeout, unit);
+ }
+
+ @Override
+ public boolean offer(final E e) {
+ return memoryLimiter.acquire(e) && super.offer(e);
+ }
+
+ @Override
+ public E take() throws InterruptedException {
+ final E e = super.take();
+ memoryLimiter.releaseInterruptibly(e);
+ return e;
+ }
+
+ @Override
+ public E poll(final long timeout, final TimeUnit unit) throws
InterruptedException {
+ final E e = super.poll(timeout, unit);
+ memoryLimiter.releaseInterruptibly(e, timeout, unit);
+ return e;
+ }
+
+ @Override
+ public E poll() {
+ final E e = super.poll();
+ memoryLimiter.release(e);
+ return e;
+ }
+
+ @Override
+ public boolean remove(final Object o) {
+ final boolean success = super.remove(o);
+ if (success) {
+ memoryLimiter.release(o);
+ }
+ return success;
+ }
+
+ @Override
+ public void clear() {
+ super.clear();
+ memoryLimiter.reset();
+ }
+}
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
new file mode 100644
index 0000000..2d3e62f
--- /dev/null
+++
b/shenyu-common/src/main/java/org/apache/shenyu/common/concurrent/MemoryLimiter.java
@@ -0,0 +1,340 @@
+/*
+ * 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 java.lang.instrument.Instrumentation;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.LongAdder;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.ReentrantLock;
+
+/**
+ * memory limiter.
+ */
+public class MemoryLimiter {
+
+ private final Instrumentation inst;
+
+ private long memoryLimit;
+
+ private final LongAdder memory = new LongAdder();
+
+ private final ReentrantLock acquireLock = new ReentrantLock();
+
+ private final Condition notLimited = acquireLock.newCondition();
+
+ private final ReentrantLock releaseLock = new ReentrantLock();
+
+ private final Condition notEmpty = releaseLock.newCondition();
+
+ public MemoryLimiter(final Instrumentation inst) {
+ this(Integer.MAX_VALUE, inst);
+ }
+
+ public MemoryLimiter(final long memoryLimit, final Instrumentation inst) {
+ if (memoryLimit <= 0) {
+ throw new IllegalArgumentException();
+ }
+ this.memoryLimit = memoryLimit;
+ this.inst = inst;
+ }
+
+ /**
+ * set the memory limit.
+ *
+ * @param memoryLimit the memory limit
+ */
+ public void setMemoryLimit(final long memoryLimit) {
+ if (memoryLimit <= 0) {
+ throw new IllegalArgumentException();
+ }
+ this.memoryLimit = memoryLimit;
+ }
+
+ /**
+ * get the memory limit.
+ *
+ * @return the memory limit
+ */
+ public long getMemoryLimit() {
+ return memoryLimit;
+ }
+
+ /**
+ * get the current memory.
+ *
+ * @return the current memory
+ */
+ public long getCurrentMemory() {
+ return memory.sum();
+ }
+
+ /**
+ * get the current remain memory.
+ *
+ * @return the current remain memory
+ */
+ public long getCurrentRemainMemory() {
+ return getMemoryLimit() - getCurrentMemory();
+ }
+
+ private void signalNotEmpty() {
+ releaseLock.lock();
+ try {
+ notEmpty.signal();
+ } finally {
+ releaseLock.unlock();
+ }
+ }
+
+ private void signalNotLimited() {
+ acquireLock.lock();
+ try {
+ notLimited.signal();
+ } finally {
+ acquireLock.unlock();
+ }
+ }
+
+ /**
+ * Locks to prevent both acquires and releases.
+ */
+ private void fullyLock() {
+ acquireLock.lock();
+ releaseLock.lock();
+ }
+
+ /**
+ * Unlocks to allow both acquires and releases.
+ */
+ private void fullyUnlock() {
+ releaseLock.unlock();
+ acquireLock.unlock();
+ }
+
+ /**
+ * acquire memory by {@link Object}.
+ * this method does not respond to interrupts.
+ *
+ * @param o memory size to be applied by calculating
+ * @return true if acquire success
+ */
+ public boolean acquire(final Object o) {
+ if (o == null) {
+ throw new NullPointerException();
+ }
+ if (memory.sum() >= memoryLimit) {
+ return false;
+ }
+ acquireLock.lock();
+ try {
+ final long sum = memory.sum();
+ final long objectSize = inst.getObjectSize(o);
+ if (sum + objectSize >= memoryLimit) {
+ return false;
+ }
+ memory.add(objectSize);
+ if (sum < memoryLimit) {
+ notLimited.signal();
+ }
+ } finally {
+ acquireLock.unlock();
+ }
+ if (memory.sum() > 0) {
+ signalNotEmpty();
+ }
+ return true;
+ }
+
+ /**
+ * acquire memory by {@link Object}.
+ * this method response to interrupts.
+ *
+ * @param o memory size to be applied by calculating
+ * @param timeout max time to wait
+ * @param unit time unit
+ * @return true if acquire success
+ * @throws InterruptedException the InterruptedException
+ */
+ public boolean acquire(final Object o, final long timeout,
+ final TimeUnit unit) throws InterruptedException {
+ if (o == null) {
+ throw new NullPointerException();
+ }
+ long nanos = unit.toNanos(timeout);
+ acquireLock.lockInterruptibly();
+ try {
+ final long objectSize = inst.getObjectSize(o);
+ while (memory.sum() + objectSize >= memoryLimit) {
+ if (nanos <= 0) {
+ return false;
+ }
+ nanos = notLimited.awaitNanos(nanos);
+ }
+ memory.add(objectSize);
+ if (memory.sum() < memoryLimit) {
+ notLimited.signal();
+ }
+ } finally {
+ acquireLock.unlock();
+ }
+ if (memory.sum() > 0) {
+ signalNotEmpty();
+ }
+ return true;
+ }
+
+ /**
+ * acquire memory by {@link Object}.
+ * this method response to interrupts.
+ *
+ * @param o memory size to be applied by calculating
+ * @throws InterruptedException the InterruptedException
+ */
+ public void acquireInterruptibly(final Object o) throws
InterruptedException {
+ if (o == null) {
+ throw new NullPointerException();
+ }
+ acquireLock.lockInterruptibly();
+ try {
+ final long sum = memory.sum();
+ final long objectSize = inst.getObjectSize(o);
+ while (sum + objectSize >= memoryLimit) {
+ notLimited.await();
+ }
+ memory.add(objectSize);
+ if (sum < memoryLimit) {
+ notLimited.signal();
+ }
+ } finally {
+ acquireLock.unlock();
+ }
+ if (memory.sum() > 0) {
+ signalNotEmpty();
+ }
+ }
+
+ /**
+ * release memory by {@link Object}.
+ * this method does not respond to interrupts.
+ *
+ * @param o memory size to be applied by calculating
+ */
+ public void release(final Object o) {
+ if (null == o) {
+ return;
+ }
+ if (memory.sum() == 0) {
+ return;
+ }
+ releaseLock.lock();
+ try {
+ final long objectSize = inst.getObjectSize(o);
+ if (memory.sum() > 0) {
+ memory.add(-objectSize);
+ if (memory.sum() > 0) {
+ notEmpty.signal();
+ }
+ }
+ } finally {
+ releaseLock.unlock();
+ }
+ if (memory.sum() < memoryLimit) {
+ signalNotLimited();
+ }
+ }
+
+ /**
+ * release memory by {@link Object}.
+ * this method response to interrupts.
+ *
+ * @param o memory size to be applied by calculating
+ * @throws InterruptedException the InterruptedException
+ */
+ public void releaseInterruptibly(final Object o) throws
InterruptedException {
+ if (null == o) {
+ return;
+ }
+ releaseLock.lockInterruptibly();
+ try {
+ final long objectSize = inst.getObjectSize(o);
+ while (memory.sum() == 0) {
+ notEmpty.await();
+ }
+ memory.add(-objectSize);
+ if (memory.sum() > 0) {
+ notEmpty.signal();
+ }
+ } finally {
+ releaseLock.unlock();
+ }
+ if (memory.sum() < memoryLimit) {
+ signalNotLimited();
+ }
+ }
+
+ /**
+ * release memory by {@link Object}.
+ * this method response to interrupts.
+ *
+ * @param o memory size to be applied by calculating
+ * @param timeout max time to wait
+ * @param unit time unit
+ * @throws InterruptedException the InterruptedException
+ */
+ public void releaseInterruptibly(final Object o, final long timeout,
+ final TimeUnit unit) throws
InterruptedException {
+ if (null == o) {
+ return;
+ }
+ long nanos = unit.toNanos(timeout);
+ releaseLock.lockInterruptibly();
+ try {
+ final long objectSize = inst.getObjectSize(o);
+ while (memory.sum() == 0) {
+ if (nanos <= 0) {
+ return;
+ }
+ nanos = notEmpty.awaitNanos(nanos);
+ }
+ memory.add(-objectSize);
+ if (memory.sum() > 0) {
+ notEmpty.signal();
+ }
+ } finally {
+ releaseLock.unlock();
+ }
+ if (memory.sum() < memoryLimit) {
+ signalNotLimited();
+ }
+ }
+
+ /**
+ * reset this MemoryLimiter.
+ */
+ public void reset() {
+ fullyLock();
+ try {
+ if (memory.sumThenReset() < memoryLimit) {
+ notLimited.signal();
+ }
+ } 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
new file mode 100644
index 0000000..047b983
--- /dev/null
+++
b/shenyu-common/src/test/java/org/apache/shenyu/common/concurrent/MemoryLimitedLinkedBlockingQueueTest.java
@@ -0,0 +1,43 @@
+/*
+ * 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.Test;
+
+import java.lang.instrument.Instrumentation;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+
+public class MemoryLimitedLinkedBlockingQueueTest {
+ @Test
+ public void test() throws Exception {
+ ByteBuddyAgent.install();
+ final Instrumentation instrumentation =
ByteBuddyAgent.getInstrumentation();
+ MemoryLimitedLinkedBlockingQueue<Runnable> queue = new
MemoryLimitedLinkedBlockingQueue<>(1, instrumentation);
+ //an Runnable needs more than 1 byte of space, so it will fail here
+ assertThat(queue.offer(() -> {
+ }), is(false));
+
+ //will success
+ queue.setMemoryLimit(Integer.MAX_VALUE);
+ assertThat(queue.offer(() -> {
+ }), is(true));
+ }
+}
