gh-153946: Add API to defer automatic garbage collection

Include/internal/pycore_gc.h

@@ -330,6 +330,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

Lib/test/test_gc.py

@@ -91,6 +91,82 @@ 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()
+
     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

@@ -2928,6 +2928,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)
 {
@@ -3379,6 +3393,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

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,22 @@ PyGC_IsEnabled(void)
     return gcstate->enabled;
 }
 
+void
+_PyGC_DeferAutomaticCollection(PyThreadState *tstate)
+{
+    GCState *gcstate = &tstate->interp->gc;
+    assert(gcstate->automatic_collection_pause_count >= 0);
+    gcstate->automatic_collection_pause_count++;
+}
+
+void
+_PyGC_ResumeAutomaticCollection(PyThreadState *tstate)
+{
+    GCState *gcstate = &tstate->interp->gc;
+    assert(gcstate->automatic_collection_pause_count > 0);
+    gcstate->automatic_collection_pause_count--;
+}
+
 /* Public API to invoke gc.collect() from C */
 Py_ssize_t
 PyGC_Collect(void)
@@ -1983,8 +1999,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 +2013,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);

Python/gc_free_threading.c

@@ -1997,8 +1997,12 @@ gc_should_collect(GCState *gcstate)
 {
     int count = _Py_atomic_load_int_relaxed(&gcstate->young.count);
     int threshold = gcstate->young.threshold;
-    int gc_enabled = _Py_atomic_load_int_relaxed(&gcstate->enabled);
-    if (count <= threshold || threshold == 0 || !gc_enabled) {
+    if (count <= threshold || threshold == 0) {
+        return false;
+    }
+    if (!_Py_atomic_load_int_relaxed(&gcstate->enabled) ||
+        _Py_atomic_load_int_relaxed(
+            &gcstate->automatic_collection_pause_count)) {
         return false;
     }
     if (gcstate->old[0].threshold == 0) {
@@ -2061,11 +2065,20 @@ record_deallocation(PyThreadState *tstate)
     }
 }
 
-static void
-gc_collect_internal(PyInterpreterState *interp, struct collection_state *state, int generation)
+static bool
+gc_collect_internal(PyInterpreterState *interp,
+                    struct collection_state *state, int generation)
 {
     _PyEval_StopTheWorld(interp);
 
+    // Close the race with a deferral that started before the world stopped.
+    if (state->reason == _Py_GC_REASON_HEAP &&
+        _Py_atomic_load_int(
+            &state->gcstate->automatic_collection_pause_count)) {
+        _PyEval_StartTheWorld(interp);
+        return false;
+    }
+
     // update collection and allocation counters
     if (generation+1 < NUM_GENERATIONS) {
         state->gcstate->old[generation].count += 1;
@@ -2110,7 +2123,7 @@ gc_collect_internal(PyInterpreterState *interp, struct collection_state *state,
         if (err < 0) {
             _PyEval_StartTheWorld(interp);
             PyErr_NoMemory();
-            return;
+            return true;
         }
     }
     #endif
@@ -2120,7 +2133,7 @@ gc_collect_internal(PyInterpreterState *interp, struct collection_state *state,
     if (err < 0) {
         _PyEval_StartTheWorld(interp);
         PyErr_NoMemory();
-        return;
+        return true;
     }
 
 #ifdef GC_DEBUG
@@ -2167,7 +2180,7 @@ gc_collect_internal(PyInterpreterState *interp, struct collection_state *state,
         cleanup_worklist(&state->wrcb_to_call);
         cleanup_worklist(&state->objs_to_decref);
         PyErr_NoMemory();
-        return;
+        return true;
     }
 
     // Call tp_clear on objects in the unreachable set. This will cause
@@ -2177,6 +2190,7 @@ gc_collect_internal(PyInterpreterState *interp, struct collection_state *state,
 
     // Append objects with legacy finalizers to the "gc.garbage" list.
     handle_legacy_finalizers(state);
+    return true;
 }
 
 static struct gc_generation_stats *
@@ -2229,8 +2243,6 @@ gc_collect_main(PyThreadState *tstate, int generation, _PyGC_Reason reason)
         s->object_stats.object_visits = 0;
     }
 #endif
-    GC_STAT_ADD(generation, collections, 1);
-
     if (reason != _Py_GC_REASON_SHUTDOWN) {
         invoke_gc_callback(tstate, "start", generation, 0, 0, 0, 0.0);
     }
@@ -2254,7 +2266,18 @@ gc_collect_main(PyThreadState *tstate, int generation, _PyGC_Reason reason)
         .reason = reason,
     };
 
-    gc_collect_internal(interp, &state, generation);
+    if (!gc_collect_internal(interp, &state, generation)) {
+        if (PyDTrace_GC_DONE_ENABLED()) {
+            PyDTrace_GC_DONE(0);
+        }
+        if (reason != _Py_GC_REASON_SHUTDOWN) {
+            invoke_gc_callback(tstate, "stop", generation, 0, 0, 0, 0.0);
+        }
+        gcstate->frame = NULL;
+        _Py_atomic_store_int(&gcstate->collecting, 0);
+        return 0;
+    }
+    GC_STAT_ADD(generation, collections, 1);
 
     m = state.collected;
     n = state.uncollectable;
@@ -2545,6 +2568,26 @@ PyGC_IsEnabled(void)
     return _Py_atomic_load_int_relaxed(&gcstate->enabled);
 }
 
+void
+_PyGC_DeferAutomaticCollection(PyThreadState *tstate)
+{
+    GCState *gcstate = &tstate->interp->gc;
+    int previous = _Py_atomic_add_int(
+        &gcstate->automatic_collection_pause_count, 1);
+    (void)previous;
+    assert(previous >= 0);
+}
+
+void
+_PyGC_ResumeAutomaticCollection(PyThreadState *tstate)
+{
+    GCState *gcstate = &tstate->interp->gc;
+    int previous = _Py_atomic_add_int(
+        &gcstate->automatic_collection_pause_count, -1);
+    (void)previous;
+    assert(previous > 0);
+}
+
 /* Public API to invoke gc.collect() from C */
 Py_ssize_t
 PyGC_Collect(void)

Python/marshal.c

@@ -9,6 +9,7 @@
 #include "Python.h"
 #include "pycore_call.h"             // _PyObject_CallNoArgs()
 #include "pycore_code.h"             // _PyCode_New()
+#include "pycore_gc.h"               // _PyGC_DeferAutomaticCollection()
 #include "pycore_hashtable.h"        // _Py_hashtable_t
 #include "pycore_long.h"             // _PyLong_IsZero()
 #include "pycore_object.h"           // _PyObject_IsUniquelyReferenced
@@ -1760,11 +1761,12 @@ static PyObject *
 read_object(RFILE *p)
 {
     PyObject *v;
+    int from_memory = p->ptr && p->end;
     if (PyErr_Occurred()) {
         fprintf(stderr, "XXX readobject called with exception set\n");
         return NULL;
     }
-    if (p->ptr && p->end) {
+    if (from_memory) {
         if (PySys_Audit("marshal.loads", "y#", p->ptr, (Py_ssize_t)(p->end - p->ptr)) < 0) {
             return NULL;
         }
@@ -1773,7 +1775,15 @@ read_object(RFILE *p)
             return NULL;
         }
     }
+    PyThreadState *tstate;
+    if (from_memory) {
+        tstate = _PyThreadState_GET();
+        _PyGC_DeferAutomaticCollection(tstate);
+    }
     v = r_object(p);
+    if (from_memory) {
+        _PyGC_ResumeAutomaticCollection(tstate);
+    }
     if (v == NULL && !PyErr_Occurred())
         PyErr_SetString(PyExc_TypeError, "NULL object in marshal data for object");
     return v;