Enhance real memory circuit breaker with G1 GC

server/src/main/java/org/elasticsearch/common/util/concurrent/ReleasableLock.java

@@ -21,6 +21,7 @@
 
 import org.elasticsearch.Assertions;
 import org.elasticsearch.common.lease.Releasable;
+import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.index.engine.EngineException;
 
 import java.util.concurrent.locks.Lock;
@@ -57,6 +58,19 @@ public ReleasableLock acquire() throws EngineException {
         return this;
     }
 
+    /**
+     * Try acquiring lock, returning null if unable to acquire lock within timeout.
+     */
+    public ReleasableLock tryAcquire(TimeValue timeout) throws InterruptedException {
+        boolean locked = lock.tryLock(timeout.duration(), timeout.timeUnit());
+        if (locked) {
+            assert addCurrentThread();
+            return this;
+        } else {
+            return null;
+        }
+    }
+
     private boolean addCurrentThread() {
         final Integer current = holdingThreads.get();
         holdingThreads.set(current == null ? 1 : current + 1);

server/src/main/java/org/elasticsearch/indices/breaker/HierarchyCircuitBreakerService.java

@@ -22,6 +22,7 @@
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.apache.logging.log4j.message.ParameterizedMessage;
+import org.elasticsearch.common.Booleans;
 import org.elasticsearch.common.breaker.ChildMemoryCircuitBreaker;
 import org.elasticsearch.common.breaker.CircuitBreaker;
 import org.elasticsearch.common.breaker.CircuitBreakingException;
@@ -30,8 +31,14 @@
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Setting.Property;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.unit.ByteSizeUnit;
 import org.elasticsearch.common.unit.ByteSizeValue;
+import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.common.util.concurrent.ReleasableLock;
+import org.elasticsearch.monitor.jvm.GcNames;
+import org.elasticsearch.monitor.jvm.JvmInfo;
 
+import java.lang.management.GarbageCollectorMXBean;
 import java.lang.management.ManagementFactory;
 import java.lang.management.MemoryMXBean;
 import java.util.ArrayList;
@@ -40,6 +47,9 @@
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.Function;
+import java.util.function.LongSupplier;
 import java.util.stream.Collectors;
 
 import static org.elasticsearch.indices.breaker.BreakerSettings.CIRCUIT_BREAKER_LIMIT_SETTING;
@@ -104,7 +114,14 @@ public class HierarchyCircuitBreakerService extends CircuitBreakerService {
     // Tripped count for when redistribution was attempted but wasn't successful
     private final AtomicLong parentTripCount = new AtomicLong(0);
 
+    private final OverLimitStrategy overLimitStrategy;
+
     public HierarchyCircuitBreakerService(Settings settings, List<BreakerSettings> customBreakers, ClusterSettings clusterSettings) {
+        this(settings, customBreakers, clusterSettings, HierarchyCircuitBreakerService::createOverLimitStrategy);
+    }
+
+    HierarchyCircuitBreakerService(Settings settings, List<BreakerSettings> customBreakers, ClusterSettings clusterSettings,
+                                   Function<Boolean, OverLimitStrategy> overLimitStrategyFactory) {
         super();
         HashMap<String, CircuitBreaker> childCircuitBreakers = new HashMap<>();
         childCircuitBreakers.put(CircuitBreaker.FIELDDATA, validateAndCreateBreaker(
@@ -168,6 +185,8 @@ public HierarchyCircuitBreakerService(Settings settings, List<BreakerSettings> c
             CIRCUIT_BREAKER_OVERHEAD_SETTING,
             (name, updatedValues) -> updateCircuitBreakerSettings(name, updatedValues.v1(), updatedValues.v2()),
             (s, t) -> {});
+
+        this.overLimitStrategy = overLimitStrategyFactory.apply(this.trackRealMemoryUsage);
     }
 
     private void updateCircuitBreakerSettings(String name, ByteSizeValue newLimit, Double newOverhead) {
@@ -231,7 +250,7 @@ public CircuitBreakerStats stats(String name) {
             breaker.getTrippedCount());
     }
 
-    private static class MemoryUsage {
+    static class MemoryUsage {
         final long baseUsage;
         final long totalUsage;
         final long transientChildUsage;
@@ -268,6 +287,10 @@ private MemoryUsage memoryUsed(long newBytesReserved) {
 
     //package private to allow overriding it in tests
     long currentMemoryUsage() {
+        return realMemoryUsage();
+    }
+
+    static long realMemoryUsage() {
         try {
             return MEMORY_MX_BEAN.getHeapMemoryUsage().getUsed();
         } catch (IllegalArgumentException ex) {
@@ -290,7 +313,7 @@ public long getParentLimit() {
     public void checkParentLimit(long newBytesReserved, String label) throws CircuitBreakingException {
         final MemoryUsage memoryUsed = memoryUsed(newBytesReserved);
         long parentLimit = this.parentSettings.getLimit();
-        if (memoryUsed.totalUsage > parentLimit) {
+        if (memoryUsed.totalUsage > parentLimit && overLimitStrategy.overLimit(memoryUsed).totalUsage > parentLimit) {
             this.parentTripCount.incrementAndGet();
             final StringBuilder message = new StringBuilder("[parent] Data too large, data for [" + label + "]" +
                     " would be [" + memoryUsed.totalUsage + "/" + new ByteSizeValue(memoryUsed.totalUsage) + "]" +
@@ -334,4 +357,163 @@ private CircuitBreaker validateAndCreateBreaker(BreakerSettings breakerSettings)
                 this,
                 breakerSettings.getName());
     }
+
+    static OverLimitStrategy createOverLimitStrategy(boolean trackRealMemoryUsage) {
+        JvmInfo jvmInfo = JvmInfo.jvmInfo();
+        if (trackRealMemoryUsage && jvmInfo.useG1GC().equals("true")
+            // messing with GC is "dangerous" so we apply an escape hatch. Not intended to be used.
+            && Booleans.parseBoolean(System.getProperty("es.real_memory_circuit_breaker.g1_over_limit_strategy.enabled"), true)) {
+            TimeValue lockTimeout = TimeValue.timeValueMillis(
+                Integer.parseInt(System.getProperty("es.real_memory_circuit_breaker.g1_over_limit_strategy.lock_timeout_ms", "500"))
+            );
+            // hardcode interval, do not want any tuning of it outside code changes.
+            return new G1OverLimitStrategy(jvmInfo, HierarchyCircuitBreakerService::realMemoryUsage, createYoungGcCountSupplier(),
+                System::currentTimeMillis, 5000, lockTimeout);
+        } else {
+            return memoryUsed -> memoryUsed;
+        }
+    }
+
+    static LongSupplier createYoungGcCountSupplier() {
+        List<GarbageCollectorMXBean> youngBeans =
+            ManagementFactory.getGarbageCollectorMXBeans().stream()
+                .filter(mxBean -> GcNames.getByGcName(mxBean.getName(), mxBean.getName()).equals(GcNames.YOUNG))
+                .collect(Collectors.toList());
+        assert youngBeans.size() == 1;
+        assert youngBeans.get(0).getCollectionCount() != -1 : "G1 must support getting collection count";
+
+        if (youngBeans.size() == 1) {
+            return youngBeans.get(0)::getCollectionCount;
+        } else {
+            logger.warn("Unable to find young generation collector, G1 over limit strategy might be impacted [{}]", youngBeans);
+            return () -> -1;
+        }
+    }
+
+    interface OverLimitStrategy {
+        MemoryUsage overLimit(MemoryUsage memoryUsed);
+    }
+
+    static class G1OverLimitStrategy implements OverLimitStrategy {
+        private final long g1RegionSize;
+        private final LongSupplier currentMemoryUsageSupplier;
+        private final LongSupplier gcCountSupplier;
+        private final LongSupplier timeSupplier;
+        private final TimeValue lockTimeout;
+        private final long maxHeap;
+
+        private long lastCheckTime = Long.MIN_VALUE;
+        private final long minimumInterval;
+
+        private long blackHole;
+        private final ReleasableLock lock = new ReleasableLock(new ReentrantLock());
+
+        G1OverLimitStrategy(JvmInfo jvmInfo, LongSupplier currentMemoryUsageSupplier,
+                            LongSupplier gcCountSupplier,
+                            LongSupplier timeSupplier, long minimumInterval, TimeValue lockTimeout) {
+            this.lockTimeout = lockTimeout;
+            assert minimumInterval > 0;
+            this.currentMemoryUsageSupplier = currentMemoryUsageSupplier;
+            this.gcCountSupplier = gcCountSupplier;
+            this.timeSupplier = timeSupplier;
+            this.minimumInterval = minimumInterval;
+            this.maxHeap = jvmInfo.getMem().getHeapMax().getBytes();
+            long g1RegionSize = jvmInfo.getG1RegionSize();
+            if (g1RegionSize <= 0) {
+                this.g1RegionSize = fallbackRegionSize(jvmInfo);
+            } else {
+                this.g1RegionSize = g1RegionSize;
+            }
+        }
+
+        static long fallbackRegionSize(JvmInfo jvmInfo) {
+            // mimick JDK calculation based on JDK 14 source:
+            // https://hg.openjdk.java.net/jdk/jdk14/file/6c954123ee8d/src/hotspot/share/gc/g1/heapRegion.cpp#l65
+            // notice that newer JDKs will have a slight variant only considering max-heap:
+            // https://hg.openjdk.java.net/jdk/jdk/file/e7d0ec2d06e8/src/hotspot/share/gc/g1/heapRegion.cpp#l67
+            // based on this JDK "bug":
+            // https://bugs.openjdk.java.net/browse/JDK-8241670
+            long averageHeapSize =
+                (jvmInfo.getMem().getHeapMax().getBytes() + JvmInfo.jvmInfo().getMem().getHeapMax().getBytes()) / 2;
+            long regionSize = Long.highestOneBit(averageHeapSize / 2048);
+            if (regionSize < ByteSizeUnit.MB.toBytes(1)) {
+                regionSize = ByteSizeUnit.MB.toBytes(1);
+            } else if (regionSize > ByteSizeUnit.MB.toBytes(32)) {
+                regionSize = ByteSizeUnit.MB.toBytes(32);
+            }
+            return regionSize;
+        }
+
+        @Override
+        public MemoryUsage overLimit(MemoryUsage memoryUsed) {
+            boolean leader = false;
+            int allocationIndex = 0;
+            long allocationDuration = 0;
+            try (ReleasableLock locked = lock.tryAcquire(lockTimeout)) {
+                if (locked != null) {
+                    long begin = timeSupplier.getAsLong();
+                    leader = begin >= lastCheckTime + minimumInterval;
+                    overLimitTriggered(leader);
+                    if (leader) {
+                        long initialCollectionCount = gcCountSupplier.getAsLong();
+                        logger.info("attempting to trigger G1GC due to high heap usage [{}]", memoryUsed.baseUsage);
+                        long localBlackHole = 0;
+                        // number of allocations, corresponding to (approximately) number of free regions + 1
+                        int allocationCount = Math.toIntExact((maxHeap - memoryUsed.baseUsage) / g1RegionSize + 1);
+                        // allocations of half-region size becomes single humongous alloc, thus taking up a full region.
+                        int allocationSize = (int) (g1RegionSize >> 1);
+                        long maxUsageObserved = memoryUsed.baseUsage;
+                        for (; allocationIndex < allocationCount; ++allocationIndex) {
+                            long current = currentMemoryUsageSupplier.getAsLong();
+                            if (current >= maxUsageObserved) {
+                                maxUsageObserved = current;
+                            } else {
+                                // we observed a memory drop, so some GC must have occurred
+                                break;
+                            }
+                            if (initialCollectionCount != gcCountSupplier.getAsLong()) {
+                                break;
+                            }
+                            localBlackHole += new byte[allocationSize].hashCode();
+                        }
+
+                        blackHole += localBlackHole;
+                        logger.trace("black hole [{}]", blackHole);
+
+                        long now = timeSupplier.getAsLong();
+                        this.lastCheckTime = now;
+                        allocationDuration = now - begin;
+                    }
+                }
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                // fallthrough
+            }
+
+            final long current = currentMemoryUsageSupplier.getAsLong();
+            if (current < memoryUsed.baseUsage) {
+                if (leader) {
+                    logger.info("GC did bring memory usage down, before [{}], after [{}], allocations [{}], duration [{}]",
+                        memoryUsed.baseUsage, current, allocationIndex, allocationDuration);
+                }
+                return new MemoryUsage(current, memoryUsed.totalUsage - memoryUsed.baseUsage + current,
+                    memoryUsed.transientChildUsage, memoryUsed.permanentChildUsage);
+            } else {
+                if (leader) {
+                    logger.info("GC did not bring memory usage down, before [{}], after [{}], allocations [{}], duration [{}]",
+                        memoryUsed.baseUsage, current, allocationIndex, allocationDuration);
+                }
+                // prefer original measurement when reporting if heap usage was not brought down.
+                return memoryUsed;
+            }
+        }
+
+        void overLimitTriggered(boolean leader) {
+            // for tests to override.
+        }
+
+        TimeValue getLockTimeout() {
+            return lockTimeout;
+        }
+    }
 }

server/src/main/java/org/elasticsearch/monitor/jvm/JvmInfo.java

@@ -98,6 +98,7 @@ public class JvmInfo implements ReportingService.Info {
         String onOutOfMemoryError = null;
         String useCompressedOops = "unknown";
         String useG1GC = "unknown";
+        long g1RegisionSize = -1;
         String useSerialGC = "unknown";
         long configuredInitialHeapSize = -1;
         long configuredMaxHeapSize = -1;
@@ -130,6 +131,8 @@ public class JvmInfo implements ReportingService.Info {
             try {
                 Object useG1GCVmOptionObject = vmOptionMethod.invoke(hotSpotDiagnosticMXBean, "UseG1GC");
                 useG1GC = (String) valueMethod.invoke(useG1GCVmOptionObject);
+                Object regionSizeVmOptionObject = vmOptionMethod.invoke(hotSpotDiagnosticMXBean, "G1HeapRegionSize");
+                g1RegisionSize = Long.parseLong((String) valueMethod.invoke(regionSizeVmOptionObject));
             } catch (Exception ignored) {
             }
 
@@ -180,9 +183,11 @@ public class JvmInfo implements ReportingService.Info {
                 onOutOfMemoryError,
                 useCompressedOops,
                 useG1GC,
-                useSerialGC);
+                useSerialGC,
+                g1RegisionSize);
     }
 
+
     @SuppressForbidden(reason = "PathUtils#get")
     private static boolean usingBundledJdk() {
         /*
@@ -229,12 +234,13 @@ public static JvmInfo jvmInfo() {
     private final String useCompressedOops;
     private final String useG1GC;
     private final String useSerialGC;
+    private final long g1RegionSize;
 
     private JvmInfo(long pid, String version, String vmName, String vmVersion, String vmVendor, boolean bundledJdk, Boolean usingBundledJdk,
                     long startTime, long configuredInitialHeapSize, long configuredMaxHeapSize, Mem mem, String[] inputArguments,
                     String bootClassPath, String classPath, Map<String, String> systemProperties, String[] gcCollectors,
                     String[] memoryPools, String onError, String onOutOfMemoryError, String useCompressedOops, String useG1GC,
-                    String useSerialGC) {
+                    String useSerialGC, long g1RegionSize) {
         this.pid = pid;
         this.version = version;
         this.vmName = vmName;
@@ -257,6 +263,7 @@ private JvmInfo(long pid, String version, String vmName, String vmVersion, Strin
         this.useCompressedOops = useCompressedOops;
         this.useG1GC = useG1GC;
         this.useSerialGC = useSerialGC;
+        this.g1RegionSize = g1RegionSize;
     }
 
     public JvmInfo(StreamInput in) throws IOException {
@@ -291,6 +298,7 @@ public JvmInfo(StreamInput in) throws IOException {
         this.onOutOfMemoryError = null;
         this.useG1GC = "unknown";
         this.useSerialGC = "unknown";
+        this.g1RegionSize = -1;
     }
 
     @Override
@@ -486,6 +494,10 @@ public String useSerialGC() {
         return this.useSerialGC;
     }
 
+    public long getG1RegionSize() {
+        return g1RegionSize;
+    }
+
     public String[] getGcCollectors() {
         return gcCollectors;
     }