gh-153946: Add API to defer automatic garbage collection

Include/internal/pycore_gc.h

@@ -314,6 +314,7 @@ static inline void _PyObject_GC_UNTRACK(
 */
 
 extern void _PyGC_InitState(struct _gc_runtime_state *);
+extern void _PyGC_AfterFork(PyThreadState *tstate);
 
 extern Py_ssize_t _PyGC_Collect(PyThreadState *tstate, int generation, _PyGC_Reason reason);
 extern void _PyGC_CollectNoFail(PyThreadState *tstate);
@@ -330,6 +331,12 @@ extern PyObject *_PyGC_GetReferrers(PyInterpreterState *interp, PyObject *objs);
 
 // Functions to clear types free lists
 extern void _PyGC_ClearAllFreeLists(PyInterpreterState *interp);
+
+// Nesting is interpreter-wide. Explicit collections and allocation counting
+// continue, and resuming does not schedule a collection immediately.
+// An in-flight free-threaded collection may still emit start and stop callbacks.
+PyAPI_FUNC(void) _PyGC_DeferAutomaticCollection(PyThreadState *tstate);
+PyAPI_FUNC(void) _PyGC_ResumeAutomaticCollection(PyThreadState *tstate);
 extern void _Py_RunGC(PyThreadState *tstate);
 
 union _PyStackRef;

Include/internal/pycore_interp_structs.h

@@ -224,6 +224,7 @@ struct gc_stats {
 struct _gc_runtime_state {
     /* Is automatic collection enabled? */
     int enabled;
+    int automatic_collection_pause_count;
     int debug;
     /* linked lists of container objects */
 #ifndef Py_GIL_DISABLED

Include/internal/pycore_tstate.h

@@ -16,12 +16,14 @@ extern "C" {
 #include "pycore_uop.h"             // struct _PyUOpInstruction
 #include "pycore_structs.h"
 
-#ifdef Py_GIL_DISABLED
 struct _gc_thread_state {
+    /* Number of active automatic collection deferrals owned by this thread. */
+    int automatic_collection_pause_count;
+#ifdef Py_GIL_DISABLED
     /* Thread-local allocation count. */
     Py_ssize_t alloc_count;
-};
 #endif
+};
 
 
 // Every PyThreadState is actually allocated as a _PyThreadStateImpl. The
@@ -64,11 +66,11 @@ typedef struct _PyThreadStateImpl {
     struct llist_node asyncio_tasks_head;
     struct _qsbr_thread_state *qsbr;  // only used by free-threaded build
     struct llist_node mem_free_queue; // delayed free queue
+    struct _gc_thread_state gc;
 
 #ifdef Py_GIL_DISABLED
     // Stack references for the current thread that exist on the C stack
     struct _PyCStackRef *c_stack_refs;
-    struct _gc_thread_state gc;
     struct _mimalloc_thread_state mimalloc;
     struct _Py_freelists freelists;
     struct _brc_thread_state brc;

Lib/test/test_gc.py

@@ -11,11 +11,13 @@
 from test.support import threading_helper, gc_threshold
 
 import gc
+import os
 import sys
 import sysconfig
 import textwrap
 import threading
 import time
+import warnings
 import weakref
 
 try:
@@ -91,6 +93,125 @@ def __tp_del__(self):
 ###############################################################################
 
 class GCTests(unittest.TestCase):
+    @staticmethod
+    def total_collections():
+        return sum(stats["collections"] for stats in gc.get_stats())
+
+    @unittest.skipIf(_testinternalcapi is None, "requires _testinternalcapi")
+    def test_defer_automatic_collection(self):
+        was_enabled = gc.isenabled()
+        gc.enable()
+        try:
+            with gc_threshold(1, 0, 0):
+                _testinternalcapi.defer_automatic_gc()
+                try:
+                    before = self.total_collections()
+                    objects = [[] for _ in range(10_000)]
+                    self.assertEqual(self.total_collections(), before)
+                    self.assertTrue(gc.isenabled())
+
+                    gc.collect()
+                    self.assertEqual(self.total_collections(), before + 1)
+                finally:
+                    _testinternalcapi.resume_automatic_gc()
+        finally:
+            if not was_enabled:
+                gc.disable()
+
+    @unittest.skipIf(_testinternalcapi is None, "requires _testinternalcapi")
+    def test_defer_automatic_collection_nested(self):
+        was_enabled = gc.isenabled()
+        gc.enable()
+        try:
+            with gc_threshold(1, 0, 0):
+                _testinternalcapi.defer_automatic_gc()
+                try:
+                    _testinternalcapi.defer_automatic_gc()
+                    try:
+                        before = self.total_collections()
+                        objects = [[] for _ in range(10_000)]
+                        self.assertEqual(self.total_collections(), before)
+                    finally:
+                        _testinternalcapi.resume_automatic_gc()
+
+                    objects.extend([] for _ in range(10_000))
+                    self.assertEqual(self.total_collections(), before)
+                finally:
+                    _testinternalcapi.resume_automatic_gc()
+
+                objects.extend([] for _ in range(10_000))
+                self.assertGreater(self.total_collections(), before)
+        finally:
+            if not was_enabled:
+                gc.disable()
+
+    @unittest.skipIf(_testinternalcapi is None, "requires _testinternalcapi")
+    @threading_helper.requires_working_threading()
+    def test_defer_automatic_collection_across_threads(self):
+        was_enabled = gc.isenabled()
+        gc.enable()
+        try:
+            with gc_threshold(1, 0, 0):
+                _testinternalcapi.defer_automatic_gc()
+                try:
+                    before = self.total_collections()
+                    objects = []
+                    thread = threading.Thread(
+                        target=lambda: objects.extend(
+                            [] for _ in range(10_000)))
+                    thread.start()
+                    thread.join()
+                    self.assertEqual(len(objects), 10_000)
+                    self.assertEqual(self.total_collections(), before)
+                finally:
+                    _testinternalcapi.resume_automatic_gc()
+        finally:
+            if not was_enabled:
+                gc.disable()
+
+    @unittest.skipIf(_testinternalcapi is None, "requires _testinternalcapi")
+    @support.requires_fork()
+    @threading_helper.requires_working_threading()
+    def test_defer_automatic_collection_after_fork(self):
+        was_enabled = gc.isenabled()
+        gc.enable()
+        try:
+            with gc_threshold(1, 0, 0):
+                ready = threading.Event()
+                release = threading.Event()
+
+                def defer_in_thread():
+                    _testinternalcapi.defer_automatic_gc()
+                    try:
+                        ready.set()
+                        release.wait()
+                    finally:
+                        _testinternalcapi.resume_automatic_gc()
+
+                thread = threading.Thread(target=defer_in_thread)
+                with threading_helper.start_threads([thread], release.set):
+                    self.assertTrue(ready.wait(support.SHORT_TIMEOUT))
+                    _testinternalcapi.defer_automatic_gc()
+                    try:
+                        with warnings.catch_warnings():
+                            warnings.filterwarnings(
+                                "ignore",
+                                message="This process .* use of fork.*",
+                                category=DeprecationWarning,
+                            )
+                            pid = os.fork()
+                        if pid == 0:
+                            _testinternalcapi.resume_automatic_gc()
+                            before = self.total_collections()
+                            objects = [[] for _ in range(10_000)]
+                            os._exit(self.total_collections() <= before)
+                        support.wait_process(pid, exitcode=0)
+                    finally:
+                        _testinternalcapi.resume_automatic_gc()
+        finally:
+            if not was_enabled:
+                gc.disable()
+
     def test_list(self):
         l = []
         l.append(l)

Misc/NEWS.d/next/Core_and_Builtins/2026-07-19-12-00-00.gh-issue-153946.r6GkP2.rst

@@ -0,0 +1,4 @@
+Add a private, nestable API for temporarily deferring automatic garbage
+collection without changing the interpreter-visible garbage collector state.
+Use it to avoid unproductive collections while converting ASTs to Python
+objects and unmarshalling objects from memory.

Modules/_testinternalcapi.c

@@ -2927,6 +2927,20 @@ get_tracked_heap_size(PyObject *self, PyObject *Py_UNUSED(ignored))
     return PyLong_FromInt64(PyInterpreterState_Get()->gc.heap_size);
 }
 
+static PyObject *
+defer_automatic_gc(PyObject *self, PyObject *Py_UNUSED(ignored))
+{
+    _PyGC_DeferAutomaticCollection(_PyThreadState_GET());
+    Py_RETURN_NONE;
+}
+
+static PyObject *
+resume_automatic_gc(PyObject *self, PyObject *Py_UNUSED(ignored))
+{
+    _PyGC_ResumeAutomaticCollection(_PyThreadState_GET());
+    Py_RETURN_NONE;
+}
+
 static PyObject *
 is_static_immortal(PyObject *self, PyObject *op)
 {
@@ -3378,6 +3392,8 @@ static PyMethodDef module_functions[] = {
     {"identify_type_slot_wrappers", identify_type_slot_wrappers, METH_NOARGS},
     {"has_deferred_refcount", has_deferred_refcount, METH_O},
     {"get_tracked_heap_size", get_tracked_heap_size, METH_NOARGS},
+    {"defer_automatic_gc", defer_automatic_gc, METH_NOARGS},
+    {"resume_automatic_gc", resume_automatic_gc, METH_NOARGS},
     {"is_static_immortal", is_static_immortal, METH_O},
     {"incref_decref_delayed", incref_decref_delayed, METH_O},
     GET_NEXT_DICT_KEYS_VERSION_METHODDEF

Modules/posixmodule.c

@@ -18,6 +18,7 @@
 #include "pycore_call.h"          // _PyObject_CallNoArgs()
 #include "pycore_ceval.h"         // _PyEval_ReInitThreads()
 #include "pycore_fileutils.h"     // _Py_closerange()
+#include "pycore_gc.h"            // _PyGC_AfterFork()
 #include "pycore_import.h"        // _PyImport_AcquireLock()
 #include "pycore_initconfig.h"    // _PyStatus_EXCEPTION()
 #include "pycore_jit_unwind.h"    // _Py_jit_debug_mutex
@@ -776,6 +777,8 @@ PyOS_AfterFork_Child(void)
         goto fatal_error;
     }
 
+    _PyGC_AfterFork(tstate);
+
 #if defined(PY_HAVE_JIT_GDB_UNWIND)
     // The child can inherit this mutex locked if another thread held it at
     // fork(), but the child itself cannot be inside gdb_jit_register_code().

Parser/asdl_c.py

@@ -2117,7 +2117,10 @@ class PartingShots(StaticVisitor):
     if (state == NULL) {
         return NULL;
     }
+    PyThreadState *tstate = _PyThreadState_GET();
+    _PyGC_DeferAutomaticCollection(tstate);
     PyObject *result = ast2obj_mod(state, t);
+    _PyGC_ResumeAutomaticCollection(tstate);
 
     return result;
 }
@@ -2254,6 +2257,7 @@ def generate_module_def(mod, metadata, f, internal_h):
         #include "pycore_ast.h"
         #include "pycore_ast_state.h"     // struct ast_state
         #include "pycore_ceval.h"         // _Py_EnterRecursiveCall()
+        #include "pycore_gc.h"            // _PyGC_DeferAutomaticCollection()
         #include "pycore_lock.h"          // _PyOnceFlag
         #include "pycore_modsupport.h"    // _PyArg_NoPositional()
         #include "pycore_pystate.h"       // _PyInterpreterState_GET()

Python/gc.c

@@ -1783,6 +1783,36 @@ PyGC_IsEnabled(void)
     return gcstate->enabled;
 }
 
+void
+_PyGC_DeferAutomaticCollection(PyThreadState *tstate)
+{
+    GCState *gcstate = &tstate->interp->gc;
+    _PyThreadStateImpl *tstate_impl = (_PyThreadStateImpl *)tstate;
+    assert(gcstate->automatic_collection_pause_count >= 0);
+    assert(tstate_impl->gc.automatic_collection_pause_count >= 0);
+    gcstate->automatic_collection_pause_count++;
+    tstate_impl->gc.automatic_collection_pause_count++;
+}
+
+void
+_PyGC_ResumeAutomaticCollection(PyThreadState *tstate)
+{
+    GCState *gcstate = &tstate->interp->gc;
+    _PyThreadStateImpl *tstate_impl = (_PyThreadStateImpl *)tstate;
+    assert(gcstate->automatic_collection_pause_count > 0);
+    assert(tstate_impl->gc.automatic_collection_pause_count > 0);
+    gcstate->automatic_collection_pause_count--;
+    tstate_impl->gc.automatic_collection_pause_count--;
+}
+
+void
+_PyGC_AfterFork(PyThreadState *tstate)
+{
+    _PyThreadStateImpl *tstate_impl = (_PyThreadStateImpl *)tstate;
+    tstate->interp->gc.automatic_collection_pause_count =
+        tstate_impl->gc.automatic_collection_pause_count;
+}
+
 /* Public API to invoke gc.collect() from C */
 Py_ssize_t
 PyGC_Collect(void)
@@ -1983,8 +2013,9 @@ _PyObject_GC_Link(PyObject *op)
     gc->_gc_prev = 0;
     gcstate->generations[0].count++; /* number of allocated GC objects */
     if (gcstate->generations[0].count > gcstate->generations[0].threshold &&
-        gcstate->enabled &&
         gcstate->generations[0].threshold &&
+        gcstate->enabled &&
+        !gcstate->automatic_collection_pause_count &&
         !_Py_atomic_load_int_relaxed(&gcstate->collecting) &&
         !_PyErr_Occurred(tstate))
     {
@@ -1996,7 +2027,8 @@ void
 _Py_RunGC(PyThreadState *tstate)
 {
     GCState *gcstate = get_gc_state();
-    if (!gcstate->enabled) {
+    if (!gcstate->enabled ||
+        gcstate->automatic_collection_pause_count) {
         return;
     }
     gc_collect_main(tstate, GENERATION_AUTO, _Py_GC_REASON_HEAP);