[ISSUE #10020] add MemorySafeLinkedBlockingQueue

dubbo-common/src/main/java/org/apache/dubbo/common/threadpool/MemoryLimitCalculator.java

@@ -0,0 +1,81 @@
+/*
+ * 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;
+
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * {@link java.lang.Runtime#freeMemory()} technology is used to calculate the
+ * memory limit by using the percentage of the current maximum available memory,
+ * which can be used with {@link MemoryLimiter}.
+ *
+ * @see MemoryLimiter
+ * @see <a href="https://github.com/apache/incubator-shenyu/blob/master/shenyu-common/src/main/java/org/apache/shenyu/common/concurrent/MemoryLimitCalculator.java">MemoryLimitCalculator</a>
+ */
+public class MemoryLimitCalculator {
+
+    private static volatile long maxAvailable;
+
+    private static final ScheduledExecutorService SCHEDULER = Executors.newSingleThreadScheduledExecutor();
+
+    static {
+        // immediately refresh when this class is loaded to prevent maxAvailable from being 0
+        refresh();
+        // check every 50 ms to improve performance
+        SCHEDULER.scheduleWithFixedDelay(MemoryLimitCalculator::refresh, 50, 50, TimeUnit.MILLISECONDS);
+        Runtime.getRuntime().addShutdownHook(new Thread(SCHEDULER::shutdown));
+    }
+
+    private static void refresh() {
+        maxAvailable = Runtime.getRuntime().freeMemory();
+    }
+
+    /**
+     * Get the maximum available memory of the current JVM.
+     *
+     * @return maximum available memory
+     */
+    public static long maxAvailable() {
+        return maxAvailable;
+    }
+
+    /**
+     * Take the current JVM's maximum available memory
+     * as a percentage of the result as the limit.
+     *
+     * @param percentage percentage
+     * @return available memory
+     */
+    public static long calculate(final float percentage) {
+        if (percentage <= 0 || percentage > 1) {
+            throw new IllegalArgumentException();
+        }
+        return (long) (maxAvailable() * percentage);
+    }
+
+    /**
+     * By default, it takes 80% of the maximum available memory of the current JVM.
+     *
+     * @return available memory
+     */
+    public static long defaultLimit() {
+        return (long) (maxAvailable() * 0.8);
+    }
+}

dubbo-common/src/main/java/org/apache/dubbo/common/threadpool/MemoryLimiter.java

@@ -122,7 +122,8 @@ public boolean acquire(Object e) {
                 return false;
             }
             memory.add(objectSize);
-            if (sum < memoryLimit) {
+            // see https://github.com/apache/incubator-shenyu/pull/3356
+            if (memory.sum() < memoryLimit) {
                 notLimited.signal();
             }
         } finally {
@@ -140,13 +141,13 @@ public void acquireInterruptibly(Object e) throws InterruptedException {
         }
         acquireLock.lockInterruptibly();
         try {
-            final long sum = memory.sum();
             final long objectSize = inst.getObjectSize(e);
-            while (sum + objectSize >= memoryLimit) {
+            // see https://github.com/apache/incubator-shenyu/pull/3335
+            while (memory.sum() + objectSize >= memoryLimit) {
                 notLimited.await();
             }
             memory.add(objectSize);
-            if (sum < memoryLimit) {
+            if (memory.sum() < memoryLimit) {
                 notLimited.signal();
             }
         } finally {

dubbo-common/src/main/java/org/apache/dubbo/common/threadpool/MemorySafeLinkedBlockingQueue.java

@@ -0,0 +1,97 @@
+/*
+ * 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;
+
+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},
+ * does not depend on {@link java.lang.instrument.Instrumentation} and is easier to use than
+ * {@link MemoryLimitedLinkedBlockingQueue}.
+ *
+ * @see <a href="https://github.com/apache/incubator-shenyu/blob/master/shenyu-common/src/main/java/org/apache/shenyu/common/concurrent/MemorySafeLinkedBlockingQueue.java">MemorySafeLinkedBlockingQueue</a>
+ */
+public class MemorySafeLinkedBlockingQueue<E> extends LinkedBlockingQueue<E> {
+
+    private static final long serialVersionUID = 8032578371739960142L;
+
+    public static int THE_256_MB = 256 * 1024 * 1024;
+
+    private int maxFreeMemory;
+
+    public MemorySafeLinkedBlockingQueue() {
+        this(THE_256_MB);
+    }
+
+    public MemorySafeLinkedBlockingQueue(final int maxFreeMemory) {
+        super(Integer.MAX_VALUE);
+        this.maxFreeMemory = maxFreeMemory;
+    }
+
+    public MemorySafeLinkedBlockingQueue(final Collection<? extends E> c,
+                                         final int maxFreeMemory) {
+        super(c);
+        this.maxFreeMemory = maxFreeMemory;
+    }
+
+    /**
+     * set the max free memory.
+     *
+     * @param maxFreeMemory the max free memory
+     */
+    public void setMaxFreeMemory(final int maxFreeMemory) {
+        this.maxFreeMemory = maxFreeMemory;
+    }
+
+    /**
+     * get the max free memory.
+     *
+     * @return the max free memory limit
+     */
+    public int getMaxFreeMemory() {
+        return maxFreeMemory;
+    }
+
+    /**
+     * determine if there is any remaining free memory.
+     *
+     * @return true if has free memory
+     */
+    public boolean hasRemainedMemory() {
+        return MemoryLimitCalculator.maxAvailable() > maxFreeMemory;
+    }
+
+    @Override
+    public void put(final E e) throws InterruptedException {
+        if (hasRemainedMemory()) {
+            super.put(e);
+        }
+    }
+
+    @Override
+    public boolean offer(final E e, final long timeout, final TimeUnit unit) throws InterruptedException {
+        return hasRemainedMemory() && super.offer(e, timeout, unit);
+    }
+
+    @Override
+    public boolean offer(final E e) {
+        return hasRemainedMemory() && super.offer(e);
+    }
+}

dubbo-common/src/main/java/org/apache/dubbo/common/threadpool/support/cached/CachedThreadPool.java

@@ -18,6 +18,7 @@
 
 import org.apache.dubbo.common.URL;
 import org.apache.dubbo.common.threadlocal.NamedInternalThreadFactory;
+import org.apache.dubbo.common.threadpool.MemorySafeLinkedBlockingQueue;
 import org.apache.dubbo.common.threadpool.ThreadPool;
 import org.apache.dubbo.common.threadpool.support.AbortPolicyWithReport;
 
@@ -54,7 +55,7 @@ public Executor getExecutor(URL url) {
         int alive = url.getParameter(ALIVE_KEY, DEFAULT_ALIVE);
         return new ThreadPoolExecutor(cores, threads, alive, TimeUnit.MILLISECONDS,
                 queues == 0 ? new SynchronousQueue<Runnable>() :
-                        (queues < 0 ? new LinkedBlockingQueue<Runnable>()
+                        (queues < 0 ? new MemorySafeLinkedBlockingQueue<Runnable>()
                                 : new LinkedBlockingQueue<Runnable>(queues)),
                 new NamedInternalThreadFactory(name, true), new AbortPolicyWithReport(name, url));
     }

dubbo-common/src/main/java/org/apache/dubbo/common/threadpool/support/fixed/FixedThreadPool.java

@@ -18,6 +18,7 @@
 
 import org.apache.dubbo.common.URL;
 import org.apache.dubbo.common.threadlocal.NamedInternalThreadFactory;
+import org.apache.dubbo.common.threadpool.MemorySafeLinkedBlockingQueue;
 import org.apache.dubbo.common.threadpool.ThreadPool;
 import org.apache.dubbo.common.threadpool.support.AbortPolicyWithReport;
 
@@ -48,7 +49,7 @@ public Executor getExecutor(URL url) {
         int queues = url.getParameter(QUEUES_KEY, DEFAULT_QUEUES);
         return new ThreadPoolExecutor(threads, threads, 0, TimeUnit.MILLISECONDS,
                 queues == 0 ? new SynchronousQueue<Runnable>() :
-                        (queues < 0 ? new LinkedBlockingQueue<Runnable>()
+                        (queues < 0 ? new MemorySafeLinkedBlockingQueue<Runnable>()
                                 : new LinkedBlockingQueue<Runnable>(queues)),
                 new NamedInternalThreadFactory(name, true), new AbortPolicyWithReport(name, url));
     }

dubbo-common/src/main/java/org/apache/dubbo/common/threadpool/support/limited/LimitedThreadPool.java

@@ -19,6 +19,7 @@
 
 import org.apache.dubbo.common.URL;
 import org.apache.dubbo.common.threadlocal.NamedInternalThreadFactory;
+import org.apache.dubbo.common.threadpool.MemorySafeLinkedBlockingQueue;
 import org.apache.dubbo.common.threadpool.ThreadPool;
 import org.apache.dubbo.common.threadpool.support.AbortPolicyWithReport;
 
@@ -51,7 +52,7 @@ public Executor getExecutor(URL url) {
         int queues = url.getParameter(QUEUES_KEY, DEFAULT_QUEUES);
         return new ThreadPoolExecutor(cores, threads, Long.MAX_VALUE, TimeUnit.MILLISECONDS,
                 queues == 0 ? new SynchronousQueue<Runnable>() :
-                        (queues < 0 ? new LinkedBlockingQueue<Runnable>()
+                        (queues < 0 ? new MemorySafeLinkedBlockingQueue<Runnable>()
                                 : new LinkedBlockingQueue<Runnable>(queues)),
                 new NamedInternalThreadFactory(name, true), new AbortPolicyWithReport(name, url));
     }

dubbo-common/src/test/java/org/apache/dubbo/common/threadpool/MemorySafeLinkedBlockingQueueTest.java

@@ -0,0 +1,46 @@
+/*
+ * 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;
+
+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 MemorySafeLinkedBlockingQueueTest {
+    @Test
+    public void test() throws Exception {
+        ByteBuddyAgent.install();
+        final Instrumentation instrumentation = ByteBuddyAgent.getInstrumentation();
+        final long objectSize = instrumentation.getObjectSize((Runnable) () -> {
+        });
+        int maxFreeMemory = (int) MemoryLimitCalculator.maxAvailable();
+        MemorySafeLinkedBlockingQueue<Runnable> queue = new MemorySafeLinkedBlockingQueue<>(maxFreeMemory);
+        // all memory is reserved for JVM, so it will fail here
+        assertThat(queue.offer(() -> {
+        }), is(false));
+
+        // maxFreeMemory-objectSize Byte memory is reserved for the JVM, so this will succeed
+        queue.setMaxFreeMemory((int) (MemoryLimitCalculator.maxAvailable() - objectSize));
+        assertThat(queue.offer(() -> {
+        }), is(true));
+    }
+}