Background GC with RSS guidance

qunaibit · qunaibit · commit 0f2639e2ec0e · 2024-10-15T13:35:39.000+03:00
skip rss test if native is not enabled
diff --git a/graalpython/com.oracle.graal.python.cext/src/capi.c b/graalpython/com.oracle.graal.python.cext/src/capi.c
@@ -566,6 +566,44 @@ PyAPI_FUNC(void) PyTruffle_ObjectArrayRelease(PyObject** array, int32_t size) {
     }
 }
 
+#if defined(__APPLE__) && defined(__MACH__)
+#include <mach/mach.h>
+#elif defined(_WIN32)
+#include <windows.h>
+#include "psapi.h"
+#endif
+
+PyAPI_FUNC(size_t) PyTruffle_GetCurrentRSS() {
+    size_t rss = 0;
+#if defined(__APPLE__) && defined(__MACH__)
+    // MacOS
+    struct mach_task_basic_info info;
+    mach_msg_type_number_t infoCount = MACH_TASK_BASIC_INFO_COUNT;
+    if (task_info(mach_task_self(), MACH_TASK_BASIC_INFO, (task_info_t) &info, &infoCount) == KERN_SUCCESS) {
+        rss = (size_t)info.resident_size;
+    }
+
+#elif defined(__linux__) || defined(__gnu_linux__)
+    // Linux
+    FILE* fp = NULL;
+    if ((fp = fopen( "/proc/self/statm", "r" )) != NULL) {
+        if (fscanf(fp, "%*s%ld", (long) &rss)) {
+            rss *= (uint64_t) sysconf( _SC_PAGESIZE);
+        }
+        fclose(fp);
+    }
+
+#elif defined(_WIN32)
+    // Windows
+    PROCESS_MEMORY_COUNTERS pmc;
+    if (GetProcessMemoryInfo(GetCurrentProcess(), &pmc, sizeof(pmc))) {
+        rss = pmc.WorkingSetSize;
+    }
+#endif
+    return rss / (1024 * 1024 /* bytes -> megabytes*/);
+}
+
+
 #define ReadMember(object, offset, T) ((T*)(((char*)object) + offset))[0]
 
 PyAPI_FUNC(int) ReadShortMember(void* object, Py_ssize_t offset) {
diff --git a/graalpython/com.oracle.graal.python.test/src/tests/cpyext/test_gc.py b/graalpython/com.oracle.graal.python.test/src/tests/cpyext/test_gc.py
@@ -896,3 +896,7 @@ def test_module_globals(self):
         ####################################### GC #######################################
         self._trigger_gc()
         ##################################################################################
+
+@skipIf(not (GRAALPY) or RUNS_ON_LLVM, "Internal GraalPy RSS function")
+def test_current_rss_monitor():
+    assert __graalpython__.get_current_rss() > 0
diff --git a/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/builtins/modules/GraalPythonModuleBuiltins.java b/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/builtins/modules/GraalPythonModuleBuiltins.java
@@ -78,8 +78,6 @@
 import java.util.List;
 import java.util.logging.Level;
 
-import com.oracle.graal.python.nodes.util.ToNativePrimitiveStorageNode;
-import com.oracle.graal.python.runtime.sequence.storage.NativePrimitiveSequenceStorage;
 import org.graalvm.home.Version;
 import org.graalvm.nativeimage.ImageInfo;
 
@@ -148,6 +146,7 @@
 import com.oracle.graal.python.nodes.statement.AbstractImportNode;
 import com.oracle.graal.python.nodes.truffle.PythonArithmeticTypes;
 import com.oracle.graal.python.nodes.util.CastToTruffleStringNode;
+import com.oracle.graal.python.nodes.util.ToNativePrimitiveStorageNode;
 import com.oracle.graal.python.runtime.PosixSupportLibrary;
 import com.oracle.graal.python.runtime.PythonContext;
 import com.oracle.graal.python.runtime.PythonImageBuildOptions;
@@ -156,6 +155,7 @@
 import com.oracle.graal.python.runtime.exception.PythonExitException;
 import com.oracle.graal.python.runtime.object.PythonObjectFactory;
 import com.oracle.graal.python.runtime.sequence.PSequence;
+import com.oracle.graal.python.runtime.sequence.storage.NativePrimitiveSequenceStorage;
 import com.oracle.graal.python.runtime.sequence.storage.NativeSequenceStorage;
 import com.oracle.graal.python.runtime.sequence.storage.SequenceStorage;
 import com.oracle.graal.python.util.PythonUtils;
@@ -1131,4 +1131,15 @@ Object doClear() {
             return PNone.NONE;
         }
     }
+
+    @Builtin(name = "get_current_rss", maxNumOfPositionalArgs = 0)
+    @GenerateNodeFactory
+    public abstract static class GetCurrentRSS extends PythonBuiltinNode {
+
+        @Specialization
+        @TruffleBoundary
+        Object currentRSS() {
+            return getContext().getCApiContext().getCurrentRSS();
+        }
+    }
 }
diff --git a/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/builtins/modules/cext/PythonCextWeakrefBuiltins.java b/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/builtins/modules/cext/PythonCextWeakrefBuiltins.java
@@ -79,8 +79,14 @@ static Object refType(Object object, Object callback,
     @CApiBuiltin(ret = PyObjectBorrowed, args = {PyObject}, call = Direct)
     abstract static class PyWeakref_GetObject extends CApiUnaryBuiltinNode {
         @Specialization
-        static Object call(PReferenceType self) {
-            return self.getPyObject();
+        static Object call(Object reference) {
+            if (reference instanceof PReferenceType ref) {
+                return ref.getPyObject();
+            }
+            /*
+             * This weak reference has died in the managed side due to its referent being collected.
+             */
+            return PNone.NONE;
         }
     }
 }
diff --git a/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/builtins/objects/cext/capi/CApiContext.java b/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/builtins/objects/cext/capi/CApiContext.java
@@ -52,6 +52,7 @@
 import java.io.IOException;
 import java.io.PrintStream;
 import java.lang.invoke.VarHandle;
+import java.lang.ref.WeakReference;
 import java.nio.file.LinkOption;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -62,6 +63,7 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.logging.Level;
 
 import org.graalvm.collections.Pair;
 import org.graalvm.nativeimage.ImageInfo;
@@ -114,6 +116,7 @@
 import com.oracle.graal.python.util.Function;
 import com.oracle.graal.python.util.PythonUtils;
 import com.oracle.graal.python.util.Supplier;
+import com.oracle.graal.python.util.SuppressFBWarnings;
 import com.oracle.graal.python.util.WeakIdentityHashMap;
 import com.oracle.truffle.api.CallTarget;
 import com.oracle.truffle.api.CompilerAsserts;
@@ -254,6 +257,9 @@ private record ClosureInfo(Object closure, Object delegate, Object executable, l
      */
     private final List<Object> loadedExtensions = new LinkedList<>();
 
+    public final BackgroundGCTask gcTask;
+    private Thread backgroundGCTaskThread;
+
     public static TruffleLogger getLogger(Class<?> clazz) {
         return PythonLanguage.getLogger(LOGGER_CAPI_NAME + "." + clazz.getSimpleName());
     }
@@ -311,6 +317,8 @@ public CApiContext(PythonContext context, Object llvmLibrary, boolean useNativeB
             assert CApiGuards.isSmallInteger(value);
             primitiveNativeWrapperCache[i] = PrimitiveNativeWrapper.createInt(value);
         }
+
+        this.gcTask = new BackgroundGCTask(context);
     }
 
     @TruffleBoundary
@@ -595,6 +603,150 @@ public void untrackObject(Object ptr, PFrame.Reference curFrame, TruffleString c
         // TODO(fa): implement untracking of container objects
     }
 
+    private static final class BackgroundGCTask implements Runnable {
+
+        private BackgroundGCTask(PythonContext context) {
+            this.ctx = new WeakReference<>(context);
+        }
+
+        Object nativeSymbol = null;
+        InteropLibrary callNative = null;
+
+        long currentRSS = -1;
+        long previousRSS = -1;
+        int previousWeakrefCount = -1;
+
+        final WeakReference<PythonContext> ctx;
+
+        // RSS monitor interval in ms
+        static final int RSS_INTERVAL = Integer.getInteger("python.RSSInterval", 1000);
+        /**
+         * RSS percentage increase between System.gc() calls. Low percentage will trigger
+         * System.gc() more often which can cause unnecessary overhead.
+         * 
+         * <ul>
+         * why 30%? it's purely based on the {@code huggingface} example.
+         * <li>less than 30%: max RSS ~22GB (>200 second per iteration)</li>
+         * <li>30%: max RSS ~24GB (~150 second per iteration)</li>
+         * <li>larger than 30%: max RSS ~38GB (~140 second per iteration)</li>
+         * </ul>
+         * 
+         * <pre>
+         */
+        static final double GC_RSS_THRESHOLD = Integer.getInteger("python.RSSThreshold", 30) / 100.0;
+
+        Long getCurrentRSS() {
+            if (nativeSymbol == null) {
+                nativeSymbol = CApiContext.getNativeSymbol(null, NativeCAPISymbol.FUN_GET_CURRENT_RSS);
+                callNative = InteropLibrary.getUncached(nativeSymbol);
+            }
+            Long rss = 0L;
+            try {
+                rss = (Long) callNative.execute(nativeSymbol);
+            } catch (Exception ignored) {
+            }
+            return rss;
+        }
+
+        @Override
+        public void run() {
+            try {
+                while (true) {
+                    Thread.sleep(RSS_INTERVAL);
+                    perform();
+                }
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+            }
+        }
+
+        private void perform() {
+            PythonContext context = ctx.get();
+            if (context == null) {
+                return;
+            }
+
+            long rss = currentRSS = getCurrentRSS();
+            if (rss == 0) {
+                LOGGER.finer("We are unable to get resident set size (RSS) from the system. " +
+                                "We will skip the java collection routine.");
+                Thread.currentThread().interrupt();
+                return;
+            }
+
+            // reset RSS baseline
+            if (rss < this.previousRSS || this.previousRSS == -1) {
+                this.previousRSS = rss;
+                return;
+            }
+
+            // skip GC if no new native weakrefs have been created.
+            int currentWeakrefCount = context.nativeContext.nativeLookup.size();
+            if (currentWeakrefCount < this.previousWeakrefCount || this.previousWeakrefCount == -1) {
+                this.previousWeakrefCount = currentWeakrefCount;
+                return;
+            }
+
+            double ratio = ((rss - this.previousRSS) / (double) this.previousRSS);
+            if (ratio >= GC_RSS_THRESHOLD) {
+                this.previousWeakrefCount = currentWeakrefCount;
+
+                long start = System.nanoTime();
+                PythonUtils.forceFullGC();
+                long gcTime = (System.nanoTime() - start) / 1000000;
+
+                if (LOGGER.isLoggable(Level.FINER)) {
+                    LOGGER.finer(PythonUtils.formatJString("Background GC Task -- GC [%d ms] RSS [%d MB]->[%d MB](%.1f%%)",
+                                    gcTime, previousRSS, rss, ratio * 100));
+                }
+                /*
+                 * cap the previous RSS increase to GC_RSS_THRESHOLD. If the ratio is much larger
+                 * than GC_RSS_THRESHOLD, then we should do GC more frequently. Though, if we get a
+                 * lower RSS in subsequent runs, the lower RSS will be set as previous RSS (see
+                 * above).
+                 *
+                 * Note: Resident Set Size (RSS) in the system isn't always an accurate indication
+                 * of used memory but rather a combination of anonymous memory (RssAnon), file
+                 * mappings (RssFile) and shmem memory (RssShmem). GC can only reduce RssAnon while
+                 * RssFile is managed by the operating system which doesn't go down easily.
+                 */
+                this.previousRSS += (long) (this.previousRSS * GC_RSS_THRESHOLD);
+            }
+        }
+    }
+
+    @TruffleBoundary
+    public long getCurrentRSS() {
+        if (backgroundGCTaskThread != null && backgroundGCTaskThread.isAlive()) {
+            long rss = gcTask.currentRSS;
+            if (rss == -1) {
+                try {
+                    // in case it just started
+                    Thread.sleep(BackgroundGCTask.RSS_INTERVAL);
+                } catch (InterruptedException e) {
+                    Thread.currentThread().interrupt();
+                }
+                rss = gcTask.currentRSS;
+            }
+            return rss;
+        }
+        return 0L;
+    }
+
+    @SuppressFBWarnings(value = "NP_NULL_ON_SOME_PATH") // context.get() is never null here
+    void runBackgroundGCTask(PythonContext context) {
+        CompilerAsserts.neverPartOfCompilation();
+        if (ImageInfo.inImageBuildtimeCode() || context.getOption(PythonOptions.NoAsyncActions)) {
+            return;
+        }
+        if (PythonOptions.AUTOMATIC_ASYNC_ACTIONS) {
+            backgroundGCTaskThread = context.getEnv().newTruffleThreadBuilder(gcTask).context(context.getEnv().getContext()).build();
+            backgroundGCTaskThread.setDaemon(true);
+            backgroundGCTaskThread.setName("python-gc-task");
+            backgroundGCTaskThread.start();
+        }
+    }
+
     /**
      * This represents whether the current process has already loaded an instance of the native CAPI
      * extensions - this can only be loaded once per process.
@@ -691,6 +843,7 @@ public static CApiContext ensureCapiWasLoaded(Node node, PythonContext context,
                     Object finalizingPointer = SignatureLibrary.getUncached().call(finalizeSignature, finalizeFunction);
                     try {
                         cApiContext.addNativeFinalizer(env, finalizingPointer);
+                        cApiContext.runBackgroundGCTask(context);
                     } catch (RuntimeException e) {
                         // This can happen when other languages restrict multithreading
                         LOGGER.warning(() -> "didn't register a native finalizer due to: " + e.getMessage());
@@ -771,6 +924,15 @@ public void exitCApiContext() {
     public void finalizeCApi() {
         CompilerAsserts.neverPartOfCompilation();
         HandleContext handleContext = getContext().nativeContext;
+        if (backgroundGCTaskThread != null && backgroundGCTaskThread.isAlive()) {
+            try {
+                backgroundGCTaskThread.interrupt();
+                backgroundGCTaskThread.join();
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+            }
+        }
+
         /*
          * Disable reference queue polling because during finalization, we will free any known
          * allocated resources (e.g. native object stubs). Calling
diff --git a/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/builtins/objects/cext/capi/NativeCAPISymbol.java b/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/builtins/objects/cext/capi/NativeCAPISymbol.java
@@ -139,6 +139,7 @@ public enum NativeCAPISymbol implements NativeCExtSymbol {
     FUN_PY_DEALLOC("_Py_Dealloc", Void, Pointer),
     FUN_BULK_DEALLOC("PyTruffle_bulk_DEALLOC", Py_ssize_t, Pointer, INT64_T),
     FUN_SHUTDOWN_BULK_DEALLOC("PyTruffle_shutdown_bulk_DEALLOC", Py_ssize_t, Pointer, INT64_T),
+    FUN_GET_CURRENT_RSS("PyTruffle_GetCurrentRSS", SIZE_T),
     FUN_TRUFFLE_ADD_SUBOFFSET("truffle_add_suboffset", Pointer, Pointer, Py_ssize_t, Py_ssize_t),
     FUN_PY_TRUFFLE_MEMORYVIEW_FROM_OBJECT("PyTruffle_MemoryViewFromObject", PyObjectTransfer, PyObject, Int),
     FUN_PY_TRUFFLE_RELEASE_BUFFER("PyTruffle_ReleaseBuffer", ArgDescriptor.Void, Pointer),
diff --git a/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/builtins/objects/cext/capi/transitions/CApiTransitions.java b/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/builtins/objects/cext/capi/transitions/CApiTransitions.java

Original file line number	Diff line number	Diff line change
`@@ -79,8 +79,14 @@ static Object refType(Object object, Object callback,`
`79`	`79`	`@CApiBuiltin(ret = PyObjectBorrowed, args = {PyObject}, call = Direct)`
`80`	`80`	`abstract static class PyWeakref_GetObject extends CApiUnaryBuiltinNode {`
`81`	`81`	`@Specialization`
`82`		`- static Object call(PReferenceType self) {`
`83`		`- return self.getPyObject();`
	`82`	`+ static Object call(Object reference) {`
	`83`	`+ if (reference instanceof PReferenceType ref) {`
	`84`	`+ return ref.getPyObject();`
	`85`	`+ }`
	`86`	`+ /*`
	`87`	`+ * This weak reference has died in the managed side due to its referent being collected.`
	`88`	`+ */`
	`89`	`+ return PNone.NONE;`
`84`	`90`	`}`
`85`	`91`	`}`
`86`	`92`	`}`