Enhance real memory circuit breaker with G1 GC (#58674) (#59394)

Using G1 GC, Elasticsearch can rarely trigger that heap usage goes above
the real memory circuit breaker limit and stays there for an extended
period. This situation will persist until the next young GC. The circuit
breaking itself hinders that from occurring in a timely manner since it
breaks all request before real work is done.

This commit gently nudges G1 to do a young GC and then double checks
that heap usage is still above the real memory circuit breaker limit
before throwing the circuit breaker exception.

Related to #57202

Author: henningandersen

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) {
             }
 
@@ -161,9 +164,11 @@ public class JvmInfo implements ReportingService.Info {
         INSTANCE = new JvmInfo(JvmPid.getPid(), System.getProperty("java.version"), runtimeMXBean.getVmName(), runtimeMXBean.getVmVersion(),
                 runtimeMXBean.getVmVendor(), bundledJdk, usingBundledJdk, runtimeMXBean.getStartTime(), configuredInitialHeapSize,
                 configuredMaxHeapSize, mem, inputArguments, bootClassPath, classPath, systemProperties, gcCollectors, memoryPools, onError,
-                onOutOfMemoryError, useCompressedOops, useG1GC, useSerialGC);
+                onOutOfMemoryError, useCompressedOops, useG1GC, useSerialGC,
+                g1RegisionSize);
     }
 
+
     @SuppressForbidden(reason = "PathUtils#get")
     private static boolean usingBundledJdk() {
         /*
@@ -210,12 +215,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;
@@ -238,6 +244,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 {
@@ -272,6 +279,7 @@ public JvmInfo(StreamInput in) throws IOException {
         this.onOutOfMemoryError = null;
         this.useG1GC = "unknown";
         this.useSerialGC = "unknown";
+        this.g1RegionSize = -1;
     }
 
     @Override
@@ -467,6 +475,10 @@ public String useSerialGC() {
         return this.useSerialGC;
     }
 
+    public long getG1RegionSize() {
+        return g1RegionSize;
+    }
+
     public String[] getGcCollectors() {
         return gcCollectors;
     }