6857566: (bf) DirectByteBuffer garbage creation can outpace reclamation

Help ReferenceHandler thread process References while attempting to allocate direct memory

Reviewed-by: alanb

jdk/src/share/classes/java/lang/ref/Reference.java

@@ -26,6 +26,8 @@
 package java.lang.ref;
 
 import sun.misc.Cleaner;
+import sun.misc.JavaLangRefAccess;
+import sun.misc.SharedSecrets;
 
 /**
  * Abstract base class for reference objects.  This class defines the
@@ -147,51 +149,75 @@ public abstract class Reference<T> {
         }
 
         public void run() {
-            for (;;) {
+            while (true) {
-                Reference<Object> r;
+                tryHandlePending(true);
-                Cleaner c;
-                try {
-                    synchronized (lock) {
-                        if (pending != null) {
-                            r = pending;
-                            // 'instanceof' might throw OutOfMemoryError sometimes
-                            // so do this before un-linking 'r' from the 'pending' chain...
-                            c = r instanceof Cleaner ? (Cleaner) r : null;
-                            // unlink 'r' from 'pending' chain
-                            pending = r.discovered;
-                            r.discovered = null;
-                        } else {
-                            // The waiting on the lock may cause an OutOfMemoryError
-                            // because it may try to allocate exception objects.
-                            lock.wait();
-                            continue;
-                        }
-                    }
-                } catch (OutOfMemoryError x) {
-                    // Give other threads CPU time so they hopefully drop some live references
-                    // and GC reclaims some space.
-                    // Also prevent CPU intensive spinning in case 'r instanceof Cleaner' above
-                    // persistently throws OOME for some time...
-                    Thread.yield();
-                    // retry
-                    continue;
-                } catch (InterruptedException x) {
-                    // retry
-                    continue;
-                }
-
-                // Fast path for cleaners
-                if (c != null) {
-                    c.clean();
-                    continue;
-                }
-
-                ReferenceQueue<Object> q = r.queue;
-                if (q != ReferenceQueue.NULL) q.enqueue(r);
             }
         }
     }
 
+    /**
+     * Try handle pending {@link Reference} if there is one.<p>
+     * Return {@code true} as a hint that there might be another
+     * {@link Reference} pending or {@code false} when there are no more pending
+     * {@link Reference}s at the moment and the program can do some other
+     * useful work instead of looping.
+     *
+     * @param waitForNotify if {@code true} and there was no pending
+     *                      {@link Reference}, wait until notified from VM
+     *                      or interrupted; if {@code false}, return immediately
+     *                      when there is no pending {@link Reference}.
+     * @return {@code true} if there was a {@link Reference} pending and it
+     *         was processed, or we waited for notification and either got it
+     *         or thread was interrupted before being notified;
+     *         {@code false} otherwise.
+     */
+    static boolean tryHandlePending(boolean waitForNotify) {
+        Reference<Object> r;
+        Cleaner c;
+        try {
+            synchronized (lock) {
+                if (pending != null) {
+                    r = pending;
+                    // 'instanceof' might throw OutOfMemoryError sometimes
+                    // so do this before un-linking 'r' from the 'pending' chain...
+                    c = r instanceof Cleaner ? (Cleaner) r : null;
+                    // unlink 'r' from 'pending' chain
+                    pending = r.discovered;
+                    r.discovered = null;
+                } else {
+                    // The waiting on the lock may cause an OutOfMemoryError
+                    // because it may try to allocate exception objects.
+                    if (waitForNotify) {
+                        lock.wait();
+                    }
+                    // retry if waited
+                    return waitForNotify;
+                }
+            }
+        } catch (OutOfMemoryError x) {
+            // Give other threads CPU time so they hopefully drop some live references
+            // and GC reclaims some space.
+            // Also prevent CPU intensive spinning in case 'r instanceof Cleaner' above
+            // persistently throws OOME for some time...
+            Thread.yield();
+            // retry
+            return true;
+        } catch (InterruptedException x) {
+            // retry
+            return true;
+        }
+
+        // Fast path for cleaners
+        if (c != null) {
+            c.clean();
+            return true;
+        }
+
+        ReferenceQueue<? super Object> q = r.queue;
+        if (q != ReferenceQueue.NULL) q.enqueue(r);
+        return true;
+    }
+
     static {
         ThreadGroup tg = Thread.currentThread().getThreadGroup();
         for (ThreadGroup tgn = tg;
@@ -204,8 +230,15 @@ public abstract class Reference<T> {
         handler.setPriority(Thread.MAX_PRIORITY);
         handler.setDaemon(true);
         handler.start();
-    }
 
+        // provide access in SharedSecrets
+        SharedSecrets.setJavaLangRefAccess(new JavaLangRefAccess() {
+            @Override
+            public boolean tryHandlePendingReference() {
+                return tryHandlePending(false);
+            }
+        });
+    }
 
     /* -- Referent accessor and setters -- */
 

jdk/src/share/classes/java/nio/Bits.java

@@ -26,6 +26,11 @@
 package java.nio;
 
 import java.security.AccessController;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.LongAdder;
+
+import sun.misc.JavaLangRefAccess;
+import sun.misc.SharedSecrets;
 import sun.misc.Unsafe;
 import sun.misc.VM;
 
@@ -621,55 +626,103 @@ class Bits {                            // package-private
     // direct buffer memory.  This value may be changed during VM
     // initialization if it is launched with "-XX:MaxDirectMemorySize=<size>".
     private static volatile long maxMemory = VM.maxDirectMemory();
-    private static volatile long reservedMemory;
+    private static final AtomicLong reservedMemory = new AtomicLong();
-    private static volatile long totalCapacity;
+    private static final AtomicLong totalCapacity = new AtomicLong();
-    private static volatile long count;
+    private static final AtomicLong count = new AtomicLong();
-    private static boolean memoryLimitSet = false;
+    private static volatile boolean memoryLimitSet = false;
+    // max. number of sleeps during try-reserving with exponentially
+    // increasing delay before throwing OutOfMemoryError:
+    // 1, 2, 4, 8, 16, 32, 64, 128, 256 (total 511 ms ~ 0.5 s)
+    // which means that OOME will be thrown after 0.5 s of trying
+    private static final int MAX_SLEEPS = 9;
 
     // These methods should be called whenever direct memory is allocated or
     // freed.  They allow the user to control the amount of direct memory
     // which a process may access.  All sizes are specified in bytes.
     static void reserveMemory(long size, int cap) {
-        synchronized (Bits.class) {
+
-            if (!memoryLimitSet && VM.isBooted()) {
+        if (!memoryLimitSet && VM.isBooted()) {
-                maxMemory = VM.maxDirectMemory();
+            maxMemory = VM.maxDirectMemory();
-                memoryLimitSet = true;
+            memoryLimitSet = true;
-            }
+        }
-            // -XX:MaxDirectMemorySize limits the total capacity rather than the
+
-            // actual memory usage, which will differ when buffers are page
+        // optimist!
-            // aligned.
+        if (tryReserveMemory(size, cap)) {
-            if (cap <= maxMemory - totalCapacity) {
+            return;
-                reservedMemory += size;
+        }
-                totalCapacity += cap;
+
-                count++;
+        final JavaLangRefAccess jlra = SharedSecrets.getJavaLangRefAccess();
+
+        // retry while helping enqueue pending Reference objects
+        // which includes executing pending Cleaner(s) which includes
+        // Cleaner(s) that free direct buffer memory
+        while (jlra.tryHandlePendingReference()) {
+            if (tryReserveMemory(size, cap)) {
                 return;
             }
         }
 
+        // trigger VM's Reference processing
         System.gc();
-        try {
-            Thread.sleep(100);
-        } catch (InterruptedException x) {
-            // Restore interrupt status
-            Thread.currentThread().interrupt();
-        }
-        synchronized (Bits.class) {
-            if (totalCapacity + cap > maxMemory)
-                throw new OutOfMemoryError("Direct buffer memory");
-            reservedMemory += size;
-            totalCapacity += cap;
-            count++;
-        }
 
+        // a retry loop with exponential back-off delays
+        // (this gives VM some time to do it's job)
+        boolean interrupted = false;
+        try {
+            long sleepTime = 1;
+            int sleeps = 0;
+            while (true) {
+                if (tryReserveMemory(size, cap)) {
+                    return;
+                }
+                if (sleeps >= MAX_SLEEPS) {
+                    break;
+                }
+                if (!jlra.tryHandlePendingReference()) {
+                    try {
+                        Thread.sleep(sleepTime);
+                        sleepTime <<= 1;
+                        sleeps++;
+                    } catch (InterruptedException e) {
+                        interrupted = true;
+                    }
+                }
+            }
+
+            // no luck
+            throw new OutOfMemoryError("Direct buffer memory");
+
+        } finally {
+            if (interrupted) {
+                // don't swallow interrupts
+                Thread.currentThread().interrupt();
+            }
+        }
     }
 
-    static synchronized void unreserveMemory(long size, int cap) {
+    private static boolean tryReserveMemory(long size, int cap) {
-        if (reservedMemory > 0) {
+
-            reservedMemory -= size;
+        // -XX:MaxDirectMemorySize limits the total capacity rather than the
-            totalCapacity -= cap;
+        // actual memory usage, which will differ when buffers are page
-            count--;
+        // aligned.
-            assert (reservedMemory > -1);
+        long totalCap;
+        while (cap <= maxMemory - (totalCap = totalCapacity.get())) {
+            if (totalCapacity.compareAndSet(totalCap, totalCap + cap)) {
+                reservedMemory.addAndGet(size);
+                count.incrementAndGet();
+                return true;
+            }
         }
+
+        return false;
+    }
+
+
+    static void unreserveMemory(long size, int cap) {
+        long cnt = count.decrementAndGet();
+        long reservedMem = reservedMemory.addAndGet(-size);
+        long totalCap = totalCapacity.addAndGet(-cap);
+        assert cnt >= 0 && reservedMem >= 0 && totalCap >= 0;
     }
 
     // -- Monitoring of direct buffer usage --
@@ -687,15 +740,15 @@ class Bits {                            // package-private
                         }
                         @Override
                         public long getCount() {
-                            return Bits.count;
+                            return Bits.count.get();
                         }
                         @Override
                         public long getTotalCapacity() {
-                            return Bits.totalCapacity;
+                            return Bits.totalCapacity.get();
                         }
                         @Override
                         public long getMemoryUsed() {
-                            return Bits.reservedMemory;
+                            return Bits.reservedMemory.get();
                         }
                     };
                 }

jdk/src/share/classes/sun/misc/JavaLangRefAccess.java

@@ -0,0 +1,39 @@
+/*
+ * Copyright (c) 2014, Oracle and/or its affiliates. All rights reserved.
+ * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
+ *
+ * This code is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU General Public License version 2 only, as
+ * published by the Free Software Foundation.  Oracle designates this
+ * particular file as subject to the "Classpath" exception as provided
+ * by Oracle in the LICENSE file that accompanied this code.
+ *
+ * This code is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
+ * version 2 for more details (a copy is included in the LICENSE file that
+ * accompanied this code).
+ *
+ * You should have received a copy of the GNU General Public License version
+ * 2 along with this work; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
+ *
+ * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
+ * or visit www.oracle.com if you need additional information or have any
+ * questions.
+ */
+
+package sun.misc;
+
+public interface JavaLangRefAccess {
+
+    /**
+     * Help ReferenceHandler thread process next pending
+     * {@link java.lang.ref.Reference}
+     *
+     * @return {@code true} if there was a pending reference and it
+     *         was enqueue-ed or {@code false} if there was no
+     *         pending reference
+     */
+    boolean tryHandlePendingReference();
+}

jdk/src/share/classes/sun/misc/SharedSecrets.java

@@ -45,6 +45,7 @@ public class SharedSecrets {
     private static final Unsafe unsafe = Unsafe.getUnsafe();
     private static JavaUtilJarAccess javaUtilJarAccess;
     private static JavaLangAccess javaLangAccess;
+    private static JavaLangRefAccess javaLangRefAccess;
     private static JavaIOAccess javaIOAccess;
     private static JavaNetAccess javaNetAccess;
     private static JavaNetHttpCookieAccess javaNetHttpCookieAccess;
@@ -76,6 +77,14 @@ public class SharedSecrets {
         return javaLangAccess;
     }
 
+    public static void setJavaLangRefAccess(JavaLangRefAccess jlra) {
+        javaLangRefAccess = jlra;
+    }
+
+    public static JavaLangRefAccess getJavaLangRefAccess() {
+        return javaLangRefAccess;
+    }
+
     public static void setJavaNetAccess(JavaNetAccess jna) {
         javaNetAccess = jna;
     }

jdk/test/java/nio/Buffer/DirectBufferAllocTest.java

@@ -0,0 +1,174 @@
+/*
+ * Copyright (c) 2014, Oracle and/or its affiliates. All rights reserved.
+ * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
+ *
+ * This code is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU General Public License version 2 only, as
+ * published by the Free Software Foundation.
+ *
+ * This code is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
+ * version 2 for more details (a copy is included in the LICENSE file that
+ * accompanied this code).
+ *
+ * You should have received a copy of the GNU General Public License version
+ * 2 along with this work; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
+ *
+ * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
+ * or visit www.oracle.com if you need additional information or have any
+ * questions.
+ */
+
+/**
+ * @test
+ * @bug 6857566
+ * @summary DirectByteBuffer garbage creation can outpace reclamation
+ *
+ * @run main/othervm -XX:MaxDirectMemorySize=128m DirectBufferAllocTest
+ */
+
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.concurrent.*;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+public class DirectBufferAllocTest {
+    // defaults
+    static final int RUN_TIME_SECONDS = 5;
+    static final int MIN_THREADS = 4;
+    static final int MAX_THREADS = 64;
+    static final int CAPACITY = 1024 * 1024; // bytes
+
+    /**
+     * This test spawns multiple threads that constantly allocate direct
+     * {@link ByteBuffer}s in a loop, trying to provoke {@link OutOfMemoryError}.<p>
+     * When run without command-line arguments, it runs as a regression test
+     * for at most 5 seconds.<p>
+     * Command line arguments:
+     * <pre>
+     * -r run-time-seconds <i>(duration of successful test - default 5 s)</i>
+     * -t threads <i>(default is 2 * # of CPUs, at least 4 but no more than 64)</i>
+     * -c capacity <i>(of direct buffers in bytes - default is 1MB)</i>
+     * -p print-alloc-time-batch-size <i>(every "batch size" iterations,
+     *                                 average time per allocation is printed)</i>
+     * </pre>
+     * Use something like the following to run a 10 minute stress test and
+     * print allocation times as it goes:
+     * <pre>
+     * java -XX:MaxDirectMemorySize=128m DirectBufferAllocTest -r 600 -t 32 -p 5000
+     * </pre>
+     */
+    public static void main(String[] args) throws Exception {
+        int runTimeSeconds = RUN_TIME_SECONDS;
+        int threads = Math.max(
+            Math.min(
+                Runtime.getRuntime().availableProcessors() * 2,
+                MAX_THREADS
+            ),
+            MIN_THREADS
+        );
+        int capacity = CAPACITY;
+        int printBatchSize = 0;
+
+        // override with command line arguments
+        for (int i = 0; i < args.length; i++) {
+            switch (args[i]) {
+                case "-r":
+                    runTimeSeconds = Integer.parseInt(args[++i]);
+                    break;
+                case "-t":
+                    threads = Integer.parseInt(args[++i]);
+                    break;
+                case "-c":
+                    capacity = Integer.parseInt(args[++i]);
+                    break;
+                case "-p":
+                    printBatchSize = Integer.parseInt(args[++i]);
+                    break;
+                default:
+                    System.err.println(
+                        "Usage: java" +
+                        " [-XX:MaxDirectMemorySize=XXXm]" +
+                        " DirectBufferAllocTest" +
+                        " [-r run-time-seconds]" +
+                        " [-t threads]" +
+                        " [-c capacity-of-direct-buffers]" +
+                        " [-p print-alloc-time-batch-size]"
+                    );
+                    System.exit(-1);
+            }
+        }
+
+        System.out.printf(
+            "Allocating direct ByteBuffers with capacity %d bytes, using %d threads for %d seconds...\n",
+            capacity, threads, runTimeSeconds
+        );
+
+        ExecutorService executor = Executors.newFixedThreadPool(threads);
+
+        int pbs = printBatchSize;
+        int cap = capacity;
+
+        List<Future<Void>> futures =
+            IntStream.range(0, threads)
+                     .mapToObj(
+                         i -> (Callable<Void>) () -> {
+                             long t0 = System.nanoTime();
+                             loop:
+                             while (true) {
+                                 for (int n = 0; pbs == 0 || n < pbs; n++) {
+                                     if (Thread.interrupted()) {
+                                         break loop;
+                                     }
+                                     ByteBuffer.allocateDirect(cap);
+                                 }
+                                 long t1 = System.nanoTime();
+                                 if (pbs > 0) {
+                                     System.out.printf(
+                                         "Thread %2d: %5.2f ms/allocation\n",
+                                         i, ((double) (t1 - t0) / (1_000_000d * pbs))
+                                     );
+                                 }
+                                 t0 = t1;
+                             }
+                             return null;
+                         }
+                     )
+                     .map(executor::submit)
+                     .collect(Collectors.toList());
+
+        for (int i = 0; i < runTimeSeconds; i++) {
+            if (futures.stream().anyMatch(Future::isDone)) {
+                break;
+            }
+            Thread.sleep(1000L);
+        }
+
+        Exception exception = null;
+        for (Future<Void> future : futures) {
+            if (future.isDone()) {
+                try {
+                    future.get();
+                } catch (ExecutionException e) {
+                    if (exception == null) {
+                        exception = new RuntimeException("Errors encountered!");
+                    }
+                    exception.addSuppressed(e.getCause());
+                }
+            } else {
+                future.cancel(true);
+            }
+        }
+
+        executor.shutdown();
+
+        if (exception != null) {
+            throw exception;
+        } else {
+            System.out.printf("No errors after %d seconds.\n", runTimeSeconds);
+        }
+    }
+}