GH-146527: Add get_gc_stats function to _remote_debugging (#148071)

2026-05-06 04:37:33 -04:00 · 2026-05-02 20:04:18 +05:00
parent 2ca6333065
commit 39f123c587
22 changed files with 1194 additions and 6 deletions
@@ -700,6 +700,16 @@ arguments (:pep:`791`).
 Improved modules
 ================

+_remote_debugging
+-----------------
+
+* Added :class:`!GCMonitor` and :func:`!get_gc_stats` to the
+  :mod:`!_remote_debugging` module to read garbage collection statistics
+  from a running Python process without constructing a :class:`!RemoteUnwinder`.
+  Results are returned as :class:`!GCStatsInfo` objects.
+  (Contributed by Sergey Miryanov and Pablo Galindo in :gh:`146527`.)
+
+
 argparse
 --------

@@ -1582,6 +1582,7 @@ _PyStaticObjects_CheckRefcnt(PyInterpreterState *interp) {
    _PyStaticObject_CheckRefcnt((PyObject *)&_Py_ID(alias));
    _PyStaticObject_CheckRefcnt((PyObject *)&_Py_ID(align));
    _PyStaticObject_CheckRefcnt((PyObject *)&_Py_ID(all));
+    _PyStaticObject_CheckRefcnt((PyObject *)&_Py_ID(all_interpreters));
    _PyStaticObject_CheckRefcnt((PyObject *)&_Py_ID(all_threads));
    _PyStaticObject_CheckRefcnt((PyObject *)&_Py_ID(allow_code));
    _PyStaticObject_CheckRefcnt((PyObject *)&_Py_ID(alphabet));
@@ -305,6 +305,7 @@ struct _Py_global_strings {
        STRUCT_FOR_ID(alias)
        STRUCT_FOR_ID(align)
        STRUCT_FOR_ID(all)
+        STRUCT_FOR_ID(all_interpreters)
        STRUCT_FOR_ID(all_threads)
        STRUCT_FOR_ID(allow_code)
        STRUCT_FOR_ID(alphabet)
@@ -1580,6 +1580,7 @@ extern "C" {
    INIT_ID(alias), \
    INIT_ID(align), \
    INIT_ID(all), \
+    INIT_ID(all_interpreters), \
    INIT_ID(all_threads), \
    INIT_ID(allow_code), \
    INIT_ID(alphabet), \
@@ -1000,6 +1000,10 @@ _PyUnicode_InitStaticStrings(PyInterpreterState *interp) {
    _PyUnicode_InternStatic(interp, &string);
    assert(_PyUnicode_CheckConsistency(string, 1));
    assert(PyUnicode_GET_LENGTH(string) != 1);
+    string = &_Py_ID(all_interpreters);
+    _PyUnicode_InternStatic(interp, &string);
+    assert(_PyUnicode_CheckConsistency(string, 1));
+    assert(PyUnicode_GET_LENGTH(string) != 1);
    string = &_Py_ID(all_threads);
    _PyUnicode_InternStatic(interp, &string);
    assert(_PyUnicode_CheckConsistency(string, 1));
@@ -0,0 +1,350 @@
+import gc
+import os
+import textwrap
+import time
+import unittest
+
+from test.support import (
+    Py_GIL_DISABLED,
+    import_helper,
+    requires_gil_enabled,
+    requires_remote_subprocess_debugging,
+)
+from test.test_profiling.test_sampling_profiler.helpers import test_subprocess
+
+try:
+    import _remote_debugging  # noqa: F401
+except ImportError:
+    raise unittest.SkipTest(
+        "Test only runs when _remote_debugging is available"
+    )
+
+
+GC_STATS_FIELDS = (
+    "gen", "iid", "ts_start", "ts_stop", "collections", "collected",
+    "uncollectable", "candidates", "duration")
+
+
+def get_interpreter_identifiers(gc_stats) -> tuple[int,...]:
+    return tuple(sorted({s.iid for s in gc_stats}))
+
+
+def get_generations(gc_stats) -> tuple[int,int,int]:
+    generations = set()
+    for s in gc_stats:
+        generations.add(s.gen)
+
+    return tuple(sorted(generations))
+
+
+def get_last_item(gc_stats, generation: int, iid: int):
+    item = None
+    for s in gc_stats:
+        if s.gen == generation and s.iid == iid:
+            if item is None or item.ts_start < s.ts_start:
+                item = s
+
+    return item
+
+
+def has_local_process_debugging():
+    try:
+        return _remote_debugging.is_python_process(os.getpid())
+    except Exception:
+        return False
+
+
+def check_gc_stats_fields(testcase, stats):
+    testcase.assertIsInstance(stats, list)
+    testcase.assertGreater(len(stats), 0)
+    for item in stats:
+        testcase.assertIsInstance(item, _remote_debugging.GCStatsInfo)
+        testcase.assertEqual(type(item).__match_args__, GC_STATS_FIELDS)
+        testcase.assertEqual(len(item), len(GC_STATS_FIELDS))
+
+
+def gc_stats_counters_advanced(before_stats, after_stats, generations, iid):
+    for generation in generations:
+        before = get_last_item(before_stats, generation, iid)
+        after = get_last_item(after_stats, generation, iid)
+        if after is None or before is None:
+            return False
+        if after.duration <= before.duration:
+            return False
+        if after.candidates <= before.candidates:
+            return False
+    return True
+
+
+@unittest.skipUnless(
+    has_local_process_debugging(), "requires local process debugging")
+class TestLocalGCStats(unittest.TestCase):
+
+    _main_iid = 0  # main interpreter ID
+
+    def test_gc_stats_fields(self):
+        monitor = _remote_debugging.GCMonitor(os.getpid(), debug=True)
+        stats = monitor.get_gc_stats(all_interpreters=False)
+        check_gc_stats_fields(self, stats)
+
+    def test_module_get_gc_stats_fields(self):
+        stats = _remote_debugging.get_gc_stats(
+            os.getpid(), all_interpreters=False)
+        check_gc_stats_fields(self, stats)
+
+    def test_all_interpreters_filter_for_local_process(self):
+        interpreters = import_helper.import_module("concurrent.interpreters")
+        source = """
+            import gc
+            objects = []
+            obj = []
+            obj.append(obj)
+            objects.append(obj)
+            gc.collect(0)
+            gc.collect(1)
+            gc.collect(2)
+        """
+        interp = interpreters.create()
+        try:
+            interp.exec(textwrap.dedent(source))
+            for generation in range(3):
+                gc.collect(generation)
+
+            main_stats = _remote_debugging.get_gc_stats(
+                os.getpid(), all_interpreters=False)
+            all_stats = _remote_debugging.get_gc_stats(
+                os.getpid(), all_interpreters=True)
+        finally:
+            interp.close()
+
+        self.assertEqual(get_interpreter_identifiers(main_stats), (0,))
+        self.assertIn(0, get_interpreter_identifiers(all_stats))
+        self.assertGreater(len(get_interpreter_identifiers(all_stats)), 1)
+        self.assertEqual(get_generations(main_stats), (0, 1, 2))
+        self.assertEqual(get_generations(all_stats), (0, 1, 2))
+        for iid in get_interpreter_identifiers(all_stats):
+            for generation in range(3):
+                self.assertIsNotNone(get_last_item(all_stats, generation, iid))
+
+    @unittest.skipUnless(Py_GIL_DISABLED, "requires free-threaded GC")
+    def test_gc_stats_counters_for_main_interpreter_free_threaded(self):
+        generations = (0, 1, 2)
+        before_stats = _remote_debugging.get_gc_stats(
+            os.getpid(), all_interpreters=False)
+        for generation in generations:
+            self.assertIsNotNone(
+                get_last_item(before_stats, generation, self._main_iid))
+
+        objects = []
+        for _ in range(1000):
+            obj = []
+            obj.append(obj)
+            objects.append(obj)
+        for generation in generations:
+            gc.collect(generation)
+
+        after_stats = _remote_debugging.get_gc_stats(
+            os.getpid(), all_interpreters=False)
+        self.assertTrue(
+            gc_stats_counters_advanced(
+                before_stats, after_stats, generations, self._main_iid),
+            (before_stats, after_stats)
+        )
+
+
+@requires_remote_subprocess_debugging()
+class TestGCStats(unittest.TestCase):
+
+    @classmethod
+    def setUpClass(cls):
+        cls._main_iid = 0 # main interpreter ID
+        cls._main_interpreter_script = '''
+            import gc
+            import time
+
+            gc.collect(0)
+            gc.collect(1)
+            gc.collect(2)
+
+            _test_sock.sendall(b"working")
+            objects = []
+            while True:
+                if len(objects) > 100:
+                    objects = []
+
+                obj = []
+                obj.append(obj)
+                objects.append(obj)
+
+                time.sleep(0.1)
+                gc.collect(0)
+                gc.collect(1)
+                gc.collect(2)
+            '''
+        cls._script = '''
+            import concurrent.interpreters as interpreters
+            import gc
+            import time
+
+            source = """if True:
+                import gc
+
+                if "objects" not in globals():
+                    objects = []
+                if len(objects) > 100:
+                    objects = []
+
+                obj = []
+                obj.append(obj)
+                objects.append(obj)
+
+                gc.collect(0)
+                gc.collect(1)
+                gc.collect(2)
+            """
+
+            if {0}:
+                interp = interpreters.create()
+                interp.exec(source)
+
+            gc.collect(0)
+            gc.collect(1)
+            gc.collect(2)
+
+            _test_sock.sendall(b"working")
+            objects = []
+            while True:
+                if len(objects) > 100:
+                    objects = []
+
+                obj = []
+                obj.append(obj)
+                objects.append(obj)
+
+                time.sleep(0.1)
+                if {0}:
+                    interp.exec(source)
+                gc.collect(0)
+                gc.collect(1)
+                gc.collect(2)
+            '''
+
+    def _gc_stats_advanced(self, before_stats, after_stats, generations):
+        for generation in generations:
+            before = get_last_item(before_stats, generation, self._main_iid)
+            after = get_last_item(after_stats, generation, self._main_iid)
+            if after is None or before is None:
+                return False
+            if after.ts_stop <= before.ts_stop:
+                return False
+        return True
+
+    def _collect_gc_stats(self, script: str, all_interpreters: bool,
+                          generations=(2,)):
+        with (test_subprocess(script, wait_for_working=True) as subproc):
+            monitor = _remote_debugging.GCMonitor(subproc.process.pid, debug=True)
+            before_stats = monitor.get_gc_stats(all_interpreters=all_interpreters)
+            for generation in generations:
+                before = get_last_item(before_stats, generation, self._main_iid)
+                self.assertIsNotNone(before)
+
+            after_stats = before_stats
+            for _ in range(10):
+                time.sleep(0.5)
+                after_stats = monitor.get_gc_stats(all_interpreters=all_interpreters)
+                if self._gc_stats_advanced(before_stats, after_stats, generations):
+                    break
+            else:
+                self.fail(
+                    f"GC stats for generations {generations!r} did not "
+                    f"advance: {before_stats!r} -> {after_stats!r}"
+                )
+
+        return before_stats, after_stats
+
+    def _check_gc_stats(self, before, after):
+        self.assertIsNotNone(before)
+        self.assertIsNotNone(after)
+
+        self.assertGreater(after.collections, before.collections, (before, after))
+        self.assertGreater(after.ts_start, before.ts_start, (before, after))
+        self.assertGreater(after.ts_stop, before.ts_stop, (before, after))
+        self.assertGreater(after.duration, before.duration, (before, after))
+
+        self.assertGreater(after.candidates, before.candidates, (before, after))
+
+        # may not grow
+        self.assertGreaterEqual(after.collected, before.collected, (before, after))
+        self.assertGreaterEqual(after.uncollectable, before.uncollectable, (before, after))
+
+    def _check_interpreter_gc_stats(self, before_stats, after_stats):
+        before_iids = get_interpreter_identifiers(before_stats)
+        after_iids = get_interpreter_identifiers(after_stats)
+
+        self.assertEqual(before_iids, after_iids)
+
+        self.assertEqual(get_generations(before_stats), (0, 1, 2))
+        self.assertEqual(get_generations(after_stats), (0, 1, 2))
+
+        for iid in after_iids:
+            with self.subTest(f"interpreter id={iid}"):
+                before_last_items = (get_last_item(before_stats, 0, iid),
+                                     get_last_item(before_stats, 1, iid),
+                                     get_last_item(before_stats, 2, iid))
+
+                after_last_items = (get_last_item(after_stats, 0, iid),
+                                    get_last_item(after_stats, 1, iid),
+                                    get_last_item(after_stats, 2, iid))
+
+                for before, after in zip(before_last_items, after_last_items):
+                    self._check_gc_stats(before, after)
+
+    def test_gc_stats_timestamps_for_main_interpreter(self):
+        script = textwrap.dedent(self._main_interpreter_script)
+        before_stats, after_stats = self._collect_gc_stats(
+            script, False, generations=(0, 1, 2))
+
+        for generation in range(3):
+            with self.subTest(generation=generation):
+                before = get_last_item(before_stats, generation, self._main_iid)
+                after = get_last_item(after_stats, generation, self._main_iid)
+
+                self.assertIsNotNone(before)
+                self.assertIsNotNone(after)
+                self.assertGreater(
+                    after.collections, before.collections,
+                    (before, after))
+                self.assertGreater(
+                    after.ts_start, before.ts_start,
+                    (before, after))
+                self.assertGreater(
+                    after.ts_stop, before.ts_stop,
+                    (before, after))
+
+    @requires_gil_enabled()
+    def test_gc_stats_for_main_interpreter(self):
+        script = textwrap.dedent(self._script.format(False))
+        before_stats, after_stats = self._collect_gc_stats(script, False)
+
+        self._check_interpreter_gc_stats(before_stats, after_stats)
+
+    @requires_gil_enabled()
+    def test_gc_stats_for_main_interpreter_if_subinterpreter_exists(self):
+        script = textwrap.dedent(self._script.format(True))
+        before_stats, after_stats = self._collect_gc_stats(script, False)
+
+        self._check_interpreter_gc_stats(before_stats, after_stats)
+
+    @requires_gil_enabled()
+    def test_gc_stats_for_all_interpreters(self):
+        script = textwrap.dedent(self._script.format(True))
+        before_stats, after_stats = self._collect_gc_stats(script, True)
+
+        before_iids = get_interpreter_identifiers(before_stats)
+        after_iids = get_interpreter_identifiers(after_stats)
+
+        self.assertGreater(len(before_iids), 1)
+        self.assertGreater(len(after_iids), 1)
+        self.assertEqual(before_iids, after_iids)
+
+        self._check_interpreter_gc_stats(before_stats, after_stats)
@@ -3496,7 +3496,7 @@ MODULE__DECIMAL_DEPS=@LIBMPDEC_INTERNAL@
 MODULE__ELEMENTTREE_DEPS=$(srcdir)/Modules/pyexpat.c @LIBEXPAT_INTERNAL@
 MODULE__HASHLIB_DEPS=$(srcdir)/Modules/hashlib.h
 MODULE__IO_DEPS=$(srcdir)/Modules/_io/_iomodule.h
-MODULE__REMOTE_DEBUGGING_DEPS=$(srcdir)/Modules/_remote_debugging/_remote_debugging.h
+MODULE__REMOTE_DEBUGGING_DEPS=$(srcdir)/Modules/_remote_debugging/_remote_debugging.h $(srcdir)/Modules/_remote_debugging/gc_stats.h

 # HACL*-based cryptographic primitives
 MODULE__MD5_DEPS=$(srcdir)/Modules/hashlib.h $(LIBHACL_MD5_HEADERS) $(LIBHACL_MD5_LIB_@LIBHACL_LDEPS_LIBTYPE@)
@@ -0,0 +1,5 @@
+Add a ``GCMonitor`` class with a ``get_gc_stats`` method to the
+:mod:`!_remote_debugging` module to allow reading GC statistics from an
+external Python process without requiring the full ``RemoteUnwinder``
+functionality.
+Patch by Sergey Miryanov and Pablo Galindo.
@@ -285,7 +285,7 @@ PYTHONPATH=$(COREPYTHONPATH)

 #*shared*
 #_ctypes_test _ctypes/_ctypes_test.c
-#_remote_debugging _remote_debugging/module.c _remote_debugging/object_reading.c _remote_debugging/code_objects.c _remote_debugging/frames.c _remote_debugging/threads.c _remote_debugging/asyncio.c
+#_remote_debugging _remote_debugging/module.c _remote_debugging/object_reading.c _remote_debugging/code_objects.c _remote_debugging/frames.c _remote_debugging/threads.c _remote_debugging/asyncio.c _remote_debugging/interpreters.c
 #_testcapi _testcapimodule.c
 #_testimportmultiple _testimportmultiple.c
 #_testmultiphase _testmultiphase.c
@@ -41,7 +41,7 @@
@MODULE__PICKLE_TRUE@_pickle _pickle.c
@MODULE__QUEUE_TRUE@_queue _queuemodule.c
@MODULE__RANDOM_TRUE@_random _randommodule.c
-@MODULE__REMOTE_DEBUGGING_TRUE@_remote_debugging _remote_debugging/module.c _remote_debugging/object_reading.c _remote_debugging/code_objects.c _remote_debugging/frames.c _remote_debugging/frame_cache.c _remote_debugging/threads.c _remote_debugging/asyncio.c _remote_debugging/binary_io_writer.c _remote_debugging/binary_io_reader.c _remote_debugging/subprocess.c
+@MODULE__REMOTE_DEBUGGING_TRUE@_remote_debugging _remote_debugging/module.c _remote_debugging/gc_stats.c _remote_debugging/object_reading.c _remote_debugging/code_objects.c _remote_debugging/frames.c _remote_debugging/frame_cache.c _remote_debugging/threads.c _remote_debugging/asyncio.c _remote_debugging/binary_io_writer.c _remote_debugging/binary_io_reader.c _remote_debugging/subprocess.c _remote_debugging/interpreters.c
@MODULE__STRUCT_TRUE@_struct _struct.c

 # build supports subinterpreters
@@ -260,8 +260,10 @@ typedef struct {
    PyTypeObject *ThreadInfo_Type;
    PyTypeObject *InterpreterInfo_Type;
    PyTypeObject *AwaitedInfo_Type;
+    PyTypeObject *GCStatsInfo_Type;
    PyTypeObject *BinaryWriter_Type;
    PyTypeObject *BinaryReader_Type;
+    PyTypeObject *GCMonitor_Type;
 } RemoteDebuggingState;

 enum _ThreadState {
@@ -346,6 +348,13 @@ typedef struct {
    size_t count;
 } StackChunkList;

+typedef struct {
+    proc_handle_t handle;
+    uintptr_t runtime_start_address;
+    struct _Py_DebugOffsets debug_offsets;
+    int debug;
+} RuntimeOffsets;
+
 /*
 * Context for frame chain traversal operations.
 */
@@ -376,6 +385,13 @@ typedef struct {
    int32_t tlbc_index;             // Thread-local bytecode index (free-threading)
 } CodeObjectContext;

+typedef struct {
+    PyObject_HEAD
+    RuntimeOffsets offsets;
+} GCMonitorObject;
+
+#define GCMonitor_CAST(op) ((GCMonitorObject *)(op))
+
 /* Function pointer types for iteration callbacks */
 typedef int (*thread_processor_func)(
    RemoteUnwinderObject *unwinder,
@@ -390,6 +406,14 @@ typedef int (*set_entry_processor_func)(
    void *context
 );

+typedef int (*interpreter_processor_func)(
+    RuntimeOffsets *offsets,
+    uintptr_t interpreter_state_addr,
+    int64_t iid,
+    void *context
+);
+
+
 /* ============================================================================
 * STRUCTSEQ DESCRIPTORS (extern declarations)
 * ============================================================================ */
@@ -401,6 +425,7 @@ extern PyStructSequence_Desc CoroInfo_desc;
 extern PyStructSequence_Desc ThreadInfo_desc;
 extern PyStructSequence_Desc InterpreterInfo_desc;
 extern PyStructSequence_Desc AwaitedInfo_desc;
+extern PyStructSequence_Desc GCStatsInfo_desc;

 /* ============================================================================
 * UTILITY FUNCTION DECLARATIONS
@@ -588,6 +613,17 @@ extern void _Py_RemoteDebug_InitThreadsState(RemoteUnwinderObject *unwinder, _Py
 extern int _Py_RemoteDebug_StopAllThreads(RemoteUnwinderObject *unwinder, _Py_RemoteDebug_ThreadsState *st);
 extern void _Py_RemoteDebug_ResumeAllThreads(RemoteUnwinderObject *unwinder, _Py_RemoteDebug_ThreadsState *st);

+/* ============================================================================
+ * INTERPRETER FUNCTION DECLARATIONS
+ * ============================================================================ */
+
+extern int
+iterate_interpreters(
+    RuntimeOffsets *offsets,
+    interpreter_processor_func processor,
+    void *context
+);
+
 /* ============================================================================
 * ASYNCIO FUNCTION DECLARATIONS
 * ============================================================================ */
@@ -495,6 +495,181 @@ _remote_debugging_RemoteUnwinder_resume_threads(PyObject *self, PyObject *Py_UNU
    return return_value;
 }

+PyDoc_STRVAR(_remote_debugging_GCMonitor___init____doc__,
+"GCMonitor(pid, *, debug=False)\n"
+"--\n"
+"\n"
+"Initialize a new GCMonitor object for monitoring GC events from remote process.\n"
+"\n"
+"Args:\n"
+"    pid: Process ID of the target Python process to monitor\n"
+"    debug: If True, chain exceptions to explain the sequence of events that\n"
+"           lead to the exception.\n"
+"\n"
+"The GCMonitor provides functionality to read GC statistics from a running\n"
+"Python process.\n"
+"\n"
+"Raises:\n"
+"    PermissionError: If access to the target process is denied\n"
+"    OSError: If unable to attach to the target process or access its memory\n"
+"    RuntimeError: If unable to read debug information from the target process");
+
+static int
+_remote_debugging_GCMonitor___init___impl(GCMonitorObject *self, int pid,
+                                          int debug);
+
+static int
+_remote_debugging_GCMonitor___init__(PyObject *self, PyObject *args, PyObject *kwargs)
+{
+    int return_value = -1;
+    #if defined(Py_BUILD_CORE) && !defined(Py_BUILD_CORE_MODULE)
+
+    #define NUM_KEYWORDS 2
+    static struct {
+        PyGC_Head _this_is_not_used;
+        PyObject_VAR_HEAD
+        Py_hash_t ob_hash;
+        PyObject *ob_item[NUM_KEYWORDS];
+    } _kwtuple = {
+        .ob_base = PyVarObject_HEAD_INIT(&PyTuple_Type, NUM_KEYWORDS)
+        .ob_hash = -1,
+        .ob_item = { &_Py_ID(pid), &_Py_ID(debug), },
+    };
+    #undef NUM_KEYWORDS
+    #define KWTUPLE (&_kwtuple.ob_base.ob_base)
+
+    #else  // !Py_BUILD_CORE
+    #  define KWTUPLE NULL
+    #endif  // !Py_BUILD_CORE
+
+    static const char * const _keywords[] = {"pid", "debug", NULL};
+    static _PyArg_Parser _parser = {
+        .keywords = _keywords,
+        .fname = "GCMonitor",
+        .kwtuple = KWTUPLE,
+    };
+    #undef KWTUPLE
+    PyObject *argsbuf[2];
+    PyObject * const *fastargs;
+    Py_ssize_t nargs = PyTuple_GET_SIZE(args);
+    Py_ssize_t noptargs = nargs + (kwargs ? PyDict_GET_SIZE(kwargs) : 0) - 1;
+    int pid;
+    int debug = 0;
+
+    fastargs = _PyArg_UnpackKeywords(_PyTuple_CAST(args)->ob_item, nargs, kwargs, NULL, &_parser,
+            /*minpos*/ 1, /*maxpos*/ 1, /*minkw*/ 0, /*varpos*/ 0, argsbuf);
+    if (!fastargs) {
+        goto exit;
+    }
+    pid = PyLong_AsInt(fastargs[0]);
+    if (pid == -1 && PyErr_Occurred()) {
+        goto exit;
+    }
+    if (!noptargs) {
+        goto skip_optional_kwonly;
+    }
+    debug = PyObject_IsTrue(fastargs[1]);
+    if (debug < 0) {
+        goto exit;
+    }
+skip_optional_kwonly:
+    return_value = _remote_debugging_GCMonitor___init___impl((GCMonitorObject *)self, pid, debug);
+
+exit:
+    return return_value;
+}
+
+PyDoc_STRVAR(_remote_debugging_GCMonitor_get_gc_stats__doc__,
+"get_gc_stats($self, /, all_interpreters=False)\n"
+"--\n"
+"\n"
+"Get garbage collector statistics from external Python process.\n"
+"\n"
+"  all_interpreters\n"
+"    If True, return GC statistics from all interpreters.\n"
+"    If False, return only from main interpreter.\n"
+"\n"
+"Returns a list of GCStatsInfo objects with GC statistics data.\n"
+"\n"
+"Returns:\n"
+"    list of GCStatsInfo: A list of stats samples containing:\n"
+"        - gen: GC generation number.\n"
+"        - iid: Interpreter ID.\n"
+"        - ts_start: Raw timestamp at collection start.\n"
+"        - ts_stop: Raw timestamp at collection stop.\n"
+"        - collections: Total number of collections.\n"
+"        - collected: Total number of collected objects.\n"
+"        - uncollectable: Total number of uncollectable objects.\n"
+"        - candidates: Total objects considered and traversed.\n"
+"        - duration: Total collection time, in seconds.\n"
+"\n"
+"Raises:\n"
+"    RuntimeError: If the target process cannot be inspected or if its\n"
+"        debug offsets or GC stats layout are incompatible.");
+
+#define _REMOTE_DEBUGGING_GCMONITOR_GET_GC_STATS_METHODDEF    \
+    {"get_gc_stats", _PyCFunction_CAST(_remote_debugging_GCMonitor_get_gc_stats), METH_FASTCALL|METH_KEYWORDS, _remote_debugging_GCMonitor_get_gc_stats__doc__},
+
+static PyObject *
+_remote_debugging_GCMonitor_get_gc_stats_impl(GCMonitorObject *self,
+                                              int all_interpreters);
+
+static PyObject *
+_remote_debugging_GCMonitor_get_gc_stats(PyObject *self, PyObject *const *args, Py_ssize_t nargs, PyObject *kwnames)
+{
+    PyObject *return_value = NULL;
+    #if defined(Py_BUILD_CORE) && !defined(Py_BUILD_CORE_MODULE)
+
+    #define NUM_KEYWORDS 1
+    static struct {
+        PyGC_Head _this_is_not_used;
+        PyObject_VAR_HEAD
+        Py_hash_t ob_hash;
+        PyObject *ob_item[NUM_KEYWORDS];
+    } _kwtuple = {
+        .ob_base = PyVarObject_HEAD_INIT(&PyTuple_Type, NUM_KEYWORDS)
+        .ob_hash = -1,
+        .ob_item = { &_Py_ID(all_interpreters), },
+    };
+    #undef NUM_KEYWORDS
+    #define KWTUPLE (&_kwtuple.ob_base.ob_base)
+
+    #else  // !Py_BUILD_CORE
+    #  define KWTUPLE NULL
+    #endif  // !Py_BUILD_CORE
+
+    static const char * const _keywords[] = {"all_interpreters", NULL};
+    static _PyArg_Parser _parser = {
+        .keywords = _keywords,
+        .fname = "get_gc_stats",
+        .kwtuple = KWTUPLE,
+    };
+    #undef KWTUPLE
+    PyObject *argsbuf[1];
+    Py_ssize_t noptargs = nargs + (kwnames ? PyTuple_GET_SIZE(kwnames) : 0) - 0;
+    int all_interpreters = 0;
+
+    args = _PyArg_UnpackKeywords(args, nargs, NULL, kwnames, &_parser,
+            /*minpos*/ 0, /*maxpos*/ 1, /*minkw*/ 0, /*varpos*/ 0, argsbuf);
+    if (!args) {
+        goto exit;
+    }
+    if (!noptargs) {
+        goto skip_optional_pos;
+    }
+    all_interpreters = PyObject_IsTrue(args[0]);
+    if (all_interpreters < 0) {
+        goto exit;
+    }
+skip_optional_pos:
+    Py_BEGIN_CRITICAL_SECTION(self);
+    return_value = _remote_debugging_GCMonitor_get_gc_stats_impl((GCMonitorObject *)self, all_interpreters);
+    Py_END_CRITICAL_SECTION();
+
+exit:
+    return return_value;
+}
+
 PyDoc_STRVAR(_remote_debugging_BinaryWriter___init____doc__,
 "BinaryWriter(filename, sample_interval_us, start_time_us, *,\n"
 "             compression=0)\n"
@@ -1296,4 +1471,96 @@ _remote_debugging_is_python_process(PyObject *module, PyObject *const *args, Py_
 exit:
    return return_value;
 }
-/*[clinic end generated code: output=34f50b18f317b9b6 input=a9049054013a1b77]*/
+
+PyDoc_STRVAR(_remote_debugging_get_gc_stats__doc__,
+"get_gc_stats($module, /, pid, *, all_interpreters=False)\n"
+"--\n"
+"\n"
+"Get garbage collector statistics from external Python process.\n"
+"\n"
+"  all_interpreters\n"
+"    If True, return GC statistics from all interpreters.\n"
+"    If False, return only from main interpreter.\n"
+"\n"
+"Returns:\n"
+"    list of GCStatsInfo: A list of stats samples containing:\n"
+"        - gen: GC generation number.\n"
+"        - iid: Interpreter ID.\n"
+"        - ts_start: Raw timestamp at collection start.\n"
+"        - ts_stop: Raw timestamp at collection stop.\n"
+"        - collections: Total number of collections.\n"
+"        - collected: Total number of collected objects.\n"
+"        - uncollectable: Total number of uncollectable objects.\n"
+"        - candidates: Total objects considered and traversed.\n"
+"        - duration: Total collection time, in seconds.\n"
+"\n"
+"Raises:\n"
+"    RuntimeError: If the target process cannot be inspected or if its\n"
+"        debug offsets or GC stats layout are incompatible.");
+
+#define _REMOTE_DEBUGGING_GET_GC_STATS_METHODDEF    \
+    {"get_gc_stats", _PyCFunction_CAST(_remote_debugging_get_gc_stats), METH_FASTCALL|METH_KEYWORDS, _remote_debugging_get_gc_stats__doc__},
+
+static PyObject *
+_remote_debugging_get_gc_stats_impl(PyObject *module, int pid,
+                                    int all_interpreters);
+
+static PyObject *
+_remote_debugging_get_gc_stats(PyObject *module, PyObject *const *args, Py_ssize_t nargs, PyObject *kwnames)
+{
+    PyObject *return_value = NULL;
+    #if defined(Py_BUILD_CORE) && !defined(Py_BUILD_CORE_MODULE)
+
+    #define NUM_KEYWORDS 2
+    static struct {
+        PyGC_Head _this_is_not_used;
+        PyObject_VAR_HEAD
+        Py_hash_t ob_hash;
+        PyObject *ob_item[NUM_KEYWORDS];
+    } _kwtuple = {
+        .ob_base = PyVarObject_HEAD_INIT(&PyTuple_Type, NUM_KEYWORDS)
+        .ob_hash = -1,
+        .ob_item = { &_Py_ID(pid), &_Py_ID(all_interpreters), },
+    };
+    #undef NUM_KEYWORDS
+    #define KWTUPLE (&_kwtuple.ob_base.ob_base)
+
+    #else  // !Py_BUILD_CORE
+    #  define KWTUPLE NULL
+    #endif  // !Py_BUILD_CORE
+
+    static const char * const _keywords[] = {"pid", "all_interpreters", NULL};
+    static _PyArg_Parser _parser = {
+        .keywords = _keywords,
+        .fname = "get_gc_stats",
+        .kwtuple = KWTUPLE,
+    };
+    #undef KWTUPLE
+    PyObject *argsbuf[2];
+    Py_ssize_t noptargs = nargs + (kwnames ? PyTuple_GET_SIZE(kwnames) : 0) - 1;
+    int pid;
+    int all_interpreters = 0;
+
+    args = _PyArg_UnpackKeywords(args, nargs, NULL, kwnames, &_parser,
+            /*minpos*/ 1, /*maxpos*/ 1, /*minkw*/ 0, /*varpos*/ 0, argsbuf);
+    if (!args) {
+        goto exit;
+    }
+    pid = PyLong_AsInt(args[0]);
+    if (pid == -1 && PyErr_Occurred()) {
+        goto exit;
+    }
+    if (!noptargs) {
+        goto skip_optional_kwonly;
+    }
+    all_interpreters = PyObject_IsTrue(args[1]);
+    if (all_interpreters < 0) {
+        goto exit;
+    }
+skip_optional_kwonly:
+    return_value = _remote_debugging_get_gc_stats_impl(module, pid, all_interpreters);
+
+exit:
+    return return_value;
+}
+/*[clinic end generated code: output=1151e58683dab9f4 input=a9049054013a1b77]*/
@@ -372,6 +372,12 @@ _PyRemoteDebug_ValidateDebugOffsetsLayout(struct _Py_DebugOffsets *debug_offsets
        sizeof(uintptr_t),
        _Alignof(uintptr_t),
        SIZEOF_GC_RUNTIME_STATE);
+    PY_REMOTE_DEBUG_VALIDATE_FIELD(
+        gc,
+        generation_stats,
+        sizeof(uintptr_t),
+        _Alignof(uintptr_t),
+        SIZEOF_GC_RUNTIME_STATE);
    PY_REMOTE_DEBUG_VALIDATE_NESTED_FIELD(
        interpreter_state,
        gc,
@@ -380,6 +386,14 @@ _PyRemoteDebug_ValidateDebugOffsetsLayout(struct _Py_DebugOffsets *debug_offsets
        sizeof(uintptr_t),
        _Alignof(uintptr_t),
        INTERP_STATE_BUFFER_SIZE);
+    PY_REMOTE_DEBUG_VALIDATE_NESTED_FIELD(
+        interpreter_state,
+        gc,
+        gc,
+        generation_stats,
+        sizeof(uintptr_t),
+        _Alignof(uintptr_t),
+        INTERP_STATE_BUFFER_SIZE);

    PY_REMOTE_DEBUG_VALIDATE_SECTION(interpreter_frame);
    PY_REMOTE_DEBUG_INTERPRETER_FRAME_FIELDS(
@@ -0,0 +1,153 @@
+/******************************************************************************
+ * Remote Debugging Module - GC Stats Functions
+ *
+ * This file contains functions for reading GC stats from interpreter state.
+ ******************************************************************************/
+
+#include "gc_stats.h"
+
+typedef struct {
+    PyObject *result;
+    PyTypeObject *gc_stats_info_type;
+    bool all_interpreters;
+} GetGCStatsContext;
+
+static int
+read_gc_stats(struct gc_stats *stats, int64_t iid, PyObject *result,
+              PyTypeObject *gc_stats_info_type)
+{
+#define SET_FIELD(converter, expr) do { \
+    PyObject *value = converter(expr); \
+    if (value == NULL) { \
+        goto error; \
+    } \
+    PyStructSequence_SetItem(item, field++, value); \
+} while (0)
+
+    PyObject *item = NULL;
+
+    for (unsigned long gen = 0; gen < NUM_GENERATIONS; gen++) {
+        struct gc_generation_stats *items;
+        int size;
+        if (gen == 0) {
+            items = (struct gc_generation_stats *)stats->young.items;
+            size = GC_YOUNG_STATS_SIZE;
+        }
+        else {
+            items = (struct gc_generation_stats *)stats->old[gen-1].items;
+            size = GC_OLD_STATS_SIZE;
+        }
+        for (int i = 0; i < size; i++, items++) {
+            item = PyStructSequence_New(gc_stats_info_type);
+            if (item == NULL) {
+                goto error;
+            }
+            Py_ssize_t field = 0;
+
+            SET_FIELD(PyLong_FromUnsignedLong, gen);
+            SET_FIELD(PyLong_FromInt64, iid);
+
+            SET_FIELD(PyLong_FromInt64, items->ts_start);
+            SET_FIELD(PyLong_FromInt64, items->ts_stop);
+            SET_FIELD(PyLong_FromSsize_t, items->collections);
+            SET_FIELD(PyLong_FromSsize_t, items->collected);
+            SET_FIELD(PyLong_FromSsize_t, items->uncollectable);
+            SET_FIELD(PyLong_FromSsize_t, items->candidates);
+
+            SET_FIELD(PyFloat_FromDouble, items->duration);
+
+            int rc = PyList_Append(result, item);
+            Py_CLEAR(item);
+            if (rc < 0) {
+                goto error;
+            }
+        }
+    }
+
+#undef SET_FIELD
+
+    return 0;
+
+error:
+    Py_XDECREF(item);
+
+    return -1;
+}
+
+static int
+get_gc_stats_from_interpreter_state(RuntimeOffsets *offsets,
+                                    uintptr_t interpreter_state_addr,
+                                    int64_t iid,
+                                    void *context)
+{
+    GetGCStatsContext *ctx = (GetGCStatsContext *)context;
+    if (!ctx->all_interpreters && iid > 0) {
+        return 0;
+    }
+
+    uintptr_t gc_stats_addr = 0;
+    uintptr_t gc_stats_pointer_address = interpreter_state_addr
+        + offsets->debug_offsets.interpreter_state.gc
+        + offsets->debug_offsets.gc.generation_stats;
+    if (_Py_RemoteDebug_ReadRemoteMemory(&offsets->handle,
+                                         gc_stats_pointer_address,
+                                         sizeof(gc_stats_addr),
+                                         &gc_stats_addr) < 0) {
+        set_exception_cause(offsets, PyExc_RuntimeError, "Failed to read GC state address");
+        return -1;
+    }
+    if (gc_stats_addr == 0) {
+        PyErr_SetString(PyExc_RuntimeError, "GC state address is NULL");
+        return -1;
+    }
+
+    struct gc_stats stats;
+    if (_Py_RemoteDebug_ReadRemoteMemory(&offsets->handle,
+                                         gc_stats_addr,
+                                         sizeof(stats),
+                                         &stats) < 0) {
+        set_exception_cause(offsets, PyExc_RuntimeError, "Failed to read GC state");
+        return -1;
+    }
+
+    if (read_gc_stats(&stats, iid, ctx->result,
+                      ctx->gc_stats_info_type) < 0) {
+        set_exception_cause(offsets, PyExc_RuntimeError, "Failed to populate GC stats result");
+        return -1;
+    }
+
+    return 0;
+}
+
+PyObject *
+get_gc_stats(RuntimeOffsets *offsets, bool all_interpreters,
+             PyTypeObject *gc_stats_info_type)
+{
+    uint64_t gc_stats_size = offsets->debug_offsets.gc.generation_stats_size;
+    if (gc_stats_size != sizeof(struct gc_stats)) {
+        PyErr_Format(PyExc_RuntimeError,
+                     "Remote gc_stats size (%llu) does not match "
+                     "local size (%zu)",
+                     (unsigned long long)gc_stats_size,
+                     sizeof(struct gc_stats));
+        set_exception_cause(offsets, PyExc_RuntimeError, "Remote gc_stats size mismatch");
+        return NULL;
+    }
+
+    PyObject *result = PyList_New(0);
+    if (result == NULL) {
+        return NULL;
+    }
+    GetGCStatsContext ctx = {
+        .result = result,
+        .gc_stats_info_type = gc_stats_info_type,
+        .all_interpreters = all_interpreters,
+    };
+    if (iterate_interpreters(offsets, get_gc_stats_from_interpreter_state,
+                             &ctx) < 0) {
+        Py_CLEAR(result);
+        return NULL;
+    }
+
+    return result;
+}
@@ -0,0 +1,24 @@
+/******************************************************************************
+ * Remote Debugging Module - GC Stats Functions
+ *
+ * This file contains declarations for reading GC stats from interpreter state.
+ ******************************************************************************/
+
+#ifndef Py_REMOTE_DEBUGGING_GC_STATS_H
+#define Py_REMOTE_DEBUGGING_GC_STATS_H
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#include "_remote_debugging.h"
+
+PyObject *
+get_gc_stats(RuntimeOffsets *offsets, bool all_interpreters,
+             PyTypeObject *gc_stats_info_type);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif  /* Py_REMOTE_DEBUGGING_GC_STATS_H */
@@ -0,0 +1,66 @@
+/******************************************************************************
+ * Remote Debugging Module - Interpreters Functions
+ *
+ * This file contains function for iterating interpreters.
+ ******************************************************************************/
+
+#include "_remote_debugging.h"
+
+int
+iterate_interpreters(
+    RuntimeOffsets *offsets,
+    interpreter_processor_func processor,
+    void *context
+) {
+    uintptr_t interpreters_head_addr =
+        offsets->runtime_start_address
+        + (uintptr_t)offsets->debug_offsets.runtime_state.interpreters_head;
+    uintptr_t interpreter_id_offset =
+        (uintptr_t)offsets->debug_offsets.interpreter_state.id;
+    uintptr_t interpreter_next_offset =
+        (uintptr_t)offsets->debug_offsets.interpreter_state.next;
+
+    uintptr_t interpreter_state_addr;
+    if (_Py_RemoteDebug_ReadRemoteMemory(&offsets->handle,
+                                         interpreters_head_addr,
+                                         sizeof(void*),
+                                         &interpreter_state_addr) < 0) {
+        set_exception_cause(offsets, PyExc_RuntimeError, "Failed to read interpreter state address");
+        return -1;
+    }
+
+    if (interpreter_state_addr == 0) {
+        PyErr_SetString(PyExc_RuntimeError, "No interpreter state found");
+        return -1;
+    }
+
+    int64_t iid = 0;
+    static_assert(
+        sizeof((((PyInterpreterState*)NULL)->id)) == sizeof(iid),
+        "Sizeof of PyInterpreterState.id mismatch with local iid value");
+    while (interpreter_state_addr != 0) {
+        if (_Py_RemoteDebug_ReadRemoteMemory(
+                    &offsets->handle,
+                    interpreter_state_addr + interpreter_id_offset,
+                    sizeof(iid),
+                    &iid) < 0) {
+            set_exception_cause(offsets, PyExc_RuntimeError, "Failed to read interpreter id");
+            return -1;
+        }
+
+        if (processor(offsets, interpreter_state_addr, iid, context) < 0) {
+            return -1;
+        }
+
+        if (_Py_RemoteDebug_ReadRemoteMemory(
+                    &offsets->handle,
+                    interpreter_state_addr + interpreter_next_offset,
+                    sizeof(void*),
+                    &interpreter_state_addr) < 0) {
+            set_exception_cause(offsets, PyExc_RuntimeError, "Failed to read next interpreter state");
+            return -1;
+        }
+    }
+
+    return 0;
+}
@@ -8,6 +8,7 @@
 #include "_remote_debugging.h"
 #include "binary_io.h"
 #include "debug_offsets_validation.h"
+#include "gc_stats.h"

 /* Forward declarations for clinic-generated code */
 typedef struct {
@@ -132,6 +133,27 @@ PyStructSequence_Desc AwaitedInfo_desc = {
    2
 };

+// GCStatsInfo structseq type
+static PyStructSequence_Field GCStatsInfo_fields[] = {
+    {"gen", "GC generation number"},
+    {"iid", "Interpreter ID"},
+    {"ts_start", "Raw timestamp at collection start"},
+    {"ts_stop", "Raw timestamp at collection stop"},
+    {"collections", "Total number of collections"},
+    {"collected", "Total number of collected objects"},
+    {"uncollectable", "Total number of uncollectable objects"},
+    {"candidates", "Total objects considered and traversed"},
+    {"duration", "Total collection time, in seconds"},
+    {NULL}
+};
+
+PyStructSequence_Desc GCStatsInfo_desc = {
+    "_remote_debugging.GCStatsInfo",
+    "Information about a garbage collector stats sample",
+    GCStatsInfo_fields,
+    9
+};
+
 /* ============================================================================
 * UTILITY FUNCTIONS
 * ============================================================================ */
@@ -1100,6 +1122,159 @@ static PyType_Spec RemoteUnwinder_spec = {
    .slots = RemoteUnwinder_slots,
 };

+/* ============================================================================
+ * GCMONITOR CLASS IMPLEMENTATION
+ * ============================================================================ */
+
+static void
+cleanup_runtime_offsets(RuntimeOffsets *offsets)
+{
+    if (offsets->handle.pid != 0) {
+        _Py_RemoteDebug_ClearCache(&offsets->handle);
+        _Py_RemoteDebug_CleanupProcHandle(&offsets->handle);
+    }
+}
+
+static int
+init_runtime_offsets(RuntimeOffsets *offsets, int pid, int debug)
+{
+    offsets->debug = debug;
+    if (_Py_RemoteDebug_InitProcHandle(&offsets->handle, pid) < 0) {
+        set_exception_cause(offsets, PyExc_RuntimeError, "Failed to initialize process handle");
+        return -1;
+    }
+    offsets->runtime_start_address = _Py_RemoteDebug_GetPyRuntimeAddress(&offsets->handle);
+    if (offsets->runtime_start_address == 0) {
+        set_exception_cause(offsets, PyExc_RuntimeError, "Failed to get Python runtime address");
+        goto error;
+    }
+    if (_Py_RemoteDebug_ReadDebugOffsets(&offsets->handle,
+                                         &offsets->runtime_start_address,
+                                         &offsets->debug_offsets) < 0)
+    {
+        set_exception_cause(offsets, PyExc_RuntimeError, "Failed to read debug offsets");
+        goto error;
+    }
+    if (validate_debug_offsets(&offsets->debug_offsets) == -1) {
+        set_exception_cause(offsets, PyExc_RuntimeError, "Invalid debug offsets found");
+        goto error;
+    }
+    return 0;
+
+error:
+    cleanup_runtime_offsets(offsets);
+    return -1;
+}
+
+/*[clinic input]
+class _remote_debugging.GCMonitor "GCMonitorObject *" "&GCMonitor_Type"
+[clinic start generated code]*/
+/*[clinic end generated code: output=da39a3ee5e6b4b0d input=ebc229325a5e5154]*/
+
+/*[clinic input]
+@permit_long_summary
+@permit_long_docstring_body
+_remote_debugging.GCMonitor.__init__
+    pid: int
+    *
+    debug: bool = False
+
+Initialize a new GCMonitor object for monitoring GC events from remote process.
+
+Args:
+    pid: Process ID of the target Python process to monitor
+    debug: If True, chain exceptions to explain the sequence of events that
+           lead to the exception.
+
+The GCMonitor provides functionality to read GC statistics from a running
+Python process.
+
+Raises:
+    PermissionError: If access to the target process is denied
+    OSError: If unable to attach to the target process or access its memory
+    RuntimeError: If unable to read debug information from the target process
+[clinic start generated code]*/
+
+static int
+_remote_debugging_GCMonitor___init___impl(GCMonitorObject *self, int pid,
+                                          int debug)
+/*[clinic end generated code: output=2cdf351c2f6335db input=1185a48535b808be]*/
+{
+    return init_runtime_offsets(&self->offsets, pid, debug);
+}
+
+/*[clinic input]
+@critical_section
+_remote_debugging.GCMonitor.get_gc_stats
+
+    all_interpreters: bool = False
+        If True, return GC statistics from all interpreters.
+        If False, return only from main interpreter.
+
+Get garbage collector statistics from external Python process.
+
+Returns a list of GCStatsInfo objects with GC statistics data.
+
+Returns:
+    list of GCStatsInfo: A list of stats samples containing:
+        - gen: GC generation number.
+        - iid: Interpreter ID.
+        - ts_start: Raw timestamp at collection start.
+        - ts_stop: Raw timestamp at collection stop.
+        - collections: Total number of collections.
+        - collected: Total number of collected objects.
+        - uncollectable: Total number of uncollectable objects.
+        - candidates: Total objects considered and traversed.
+        - duration: Total collection time, in seconds.
+
+Raises:
+    RuntimeError: If the target process cannot be inspected or if its
+        debug offsets or GC stats layout are incompatible.
+[clinic start generated code]*/
+
+static PyObject *
+_remote_debugging_GCMonitor_get_gc_stats_impl(GCMonitorObject *self,
+                                              int all_interpreters)
+/*[clinic end generated code: output=f73f365725224f7a input=09e647719c65f9e4]*/
+{
+    RemoteDebuggingState *st = RemoteDebugging_GetStateFromType(Py_TYPE(self));
+    return get_gc_stats(&self->offsets, all_interpreters, st->GCStatsInfo_Type);
+}
+
+static PyMethodDef GCMonitor_methods[] = {
+    _REMOTE_DEBUGGING_GCMONITOR_GET_GC_STATS_METHODDEF
+    {NULL, NULL}
+};
+
+static void
+GCMonitor_dealloc(PyObject *op)
+{
+    GCMonitorObject *self = GCMonitor_CAST(op);
+    PyTypeObject *tp = Py_TYPE(self);
+
+    cleanup_runtime_offsets(&self->offsets);
+    PyObject_Del(self);
+    Py_DECREF(tp);
+}
+
+static PyType_Slot GCMonitor_slots[] = {
+    {Py_tp_doc, (void *)"GCMonitor(pid): Monitor GC events of a remote Python process."},
+    {Py_tp_methods, GCMonitor_methods},
+    {Py_tp_init, _remote_debugging_GCMonitor___init__},
+    {Py_tp_dealloc, GCMonitor_dealloc},
+    {0, NULL}
+};
+
+static PyType_Spec GCMonitor_spec = {
+    .name = "_remote_debugging.GCMonitor",
+    .basicsize = sizeof(GCMonitorObject),
+    .flags = (
+        Py_TPFLAGS_DEFAULT
+        | Py_TPFLAGS_IMMUTABLETYPE
+    ),
+    .slots = GCMonitor_slots,
+};
+
 /* Forward declarations for type specs defined later */
 static PyType_Spec BinaryWriter_spec;
 static PyType_Spec BinaryReader_spec;
@@ -1126,6 +1301,11 @@ _remote_debugging_exec(PyObject *m)
        return -1;
    }

+    CREATE_TYPE(m, st->GCMonitor_Type, &GCMonitor_spec);
+    if (PyModule_AddType(m, st->GCMonitor_Type) < 0) {
+        return -1;
+    }
+
    // Initialize structseq types
    st->TaskInfo_Type = PyStructSequence_NewType(&TaskInfo_desc);
    if (st->TaskInfo_Type == NULL) {
@@ -1183,6 +1363,14 @@ _remote_debugging_exec(PyObject *m)
        return -1;
    }

+    st->GCStatsInfo_Type = PyStructSequence_NewType(&GCStatsInfo_desc);
+    if (st->GCStatsInfo_Type == NULL) {
+        return -1;
+    }
+    if (PyModule_AddType(m, st->GCStatsInfo_Type) < 0) {
+        return -1;
+    }
+
    // Create BinaryWriter and BinaryReader types
    CREATE_TYPE(m, st->BinaryWriter_Type, &BinaryWriter_spec);
    if (PyModule_AddType(m, st->BinaryWriter_Type) < 0) {
@@ -1240,8 +1428,10 @@ remote_debugging_traverse(PyObject *mod, visitproc visit, void *arg)
    Py_VISIT(state->ThreadInfo_Type);
    Py_VISIT(state->InterpreterInfo_Type);
    Py_VISIT(state->AwaitedInfo_Type);
+    Py_VISIT(state->GCStatsInfo_Type);
    Py_VISIT(state->BinaryWriter_Type);
    Py_VISIT(state->BinaryReader_Type);
+    Py_VISIT(state->GCMonitor_Type);
    return 0;
 }

@@ -1257,8 +1447,10 @@ remote_debugging_clear(PyObject *mod)
    Py_CLEAR(state->ThreadInfo_Type);
    Py_CLEAR(state->InterpreterInfo_Type);
    Py_CLEAR(state->AwaitedInfo_Type);
+    Py_CLEAR(state->GCStatsInfo_Type);
    Py_CLEAR(state->BinaryWriter_Type);
    Py_CLEAR(state->BinaryReader_Type);
+    Py_CLEAR(state->GCMonitor_Type);
    return 0;
 }

@@ -1837,10 +2029,57 @@ _remote_debugging_is_python_process_impl(PyObject *module, int pid)
    Py_RETURN_TRUE;
 }

+/*[clinic input]
+_remote_debugging.get_gc_stats
+
+    pid: int
+    *
+    all_interpreters: bool = False
+        If True, return GC statistics from all interpreters.
+        If False, return only from main interpreter.
+
+Get garbage collector statistics from external Python process.
+
+Returns:
+    list of GCStatsInfo: A list of stats samples containing:
+        - gen: GC generation number.
+        - iid: Interpreter ID.
+        - ts_start: Raw timestamp at collection start.
+        - ts_stop: Raw timestamp at collection stop.
+        - collections: Total number of collections.
+        - collected: Total number of collected objects.
+        - uncollectable: Total number of uncollectable objects.
+        - candidates: Total objects considered and traversed.
+        - duration: Total collection time, in seconds.
+
+Raises:
+    RuntimeError: If the target process cannot be inspected or if its
+        debug offsets or GC stats layout are incompatible.
+[clinic start generated code]*/
+
+static PyObject *
+_remote_debugging_get_gc_stats_impl(PyObject *module, int pid,
+                                    int all_interpreters)
+/*[clinic end generated code: output=d9dce5f7add149bb input=a2a08a45a8f0b119]*/
+{
+    RuntimeOffsets offsets;
+    if (init_runtime_offsets(&offsets, pid, /*debug=*/1) < 0) {
+        return NULL;
+    }
+
+    RemoteDebuggingState *st = RemoteDebugging_GetState(module);
+    PyObject *result = get_gc_stats(&offsets, all_interpreters,
+                                    st->GCStatsInfo_Type);
+
+    cleanup_runtime_offsets(&offsets);
+    return result;
+}
+
 static PyMethodDef remote_debugging_methods[] = {
    _REMOTE_DEBUGGING_ZSTD_AVAILABLE_METHODDEF
    _REMOTE_DEBUGGING_GET_CHILD_PIDS_METHODDEF
    _REMOTE_DEBUGGING_IS_PYTHON_PROCESS_METHODDEF
+    _REMOTE_DEBUGGING_GET_GC_STATS_METHODDEF
    {NULL, NULL, 0, NULL},
 };

@@ -99,6 +99,7 @@
 </ItemDefinitionGroup>
  <ItemGroup>
    <ClCompile Include="..\Modules\_remote_debugging\module.c" />
+    <ClCompile Include="..\Modules\_remote_debugging\gc_stats.c" />
    <ClCompile Include="..\Modules\_remote_debugging\object_reading.c" />
    <ClCompile Include="..\Modules\_remote_debugging\code_objects.c" />
    <ClCompile Include="..\Modules\_remote_debugging\frames.c" />
@@ -108,10 +109,12 @@
    <ClCompile Include="..\Modules\_remote_debugging\binary_io_writer.c" />
    <ClCompile Include="..\Modules\_remote_debugging\binary_io_reader.c" />
    <ClCompile Include="..\Modules\_remote_debugging\subprocess.c" />
+    <ClCompile Include="..\Modules\_remote_debugging\interpreters.c" />
  </ItemGroup>
  <ItemGroup>
    <ClInclude Include="..\Modules\_remote_debugging\_remote_debugging.h" />
    <ClInclude Include="..\Modules\_remote_debugging\binary_io.h" />
+    <ClInclude Include="..\Modules\_remote_debugging\gc_stats.h" />
  </ItemGroup>
  <ItemGroup>
    <ResourceCompile Include="..\PC\python_nt.rc" />
@@ -15,6 +15,9 @@
    <ClCompile Include="..\Modules\_remote_debugging\module.c">
      <Filter>Source Files</Filter>
    </ClCompile>
+    <ClCompile Include="..\Modules\_remote_debugging\gc_stats.c">
+      <Filter>Source Files</Filter>
+    </ClCompile>
    <ClCompile Include="..\Modules\_remote_debugging\object_reading.c">
      <Filter>Source Files</Filter>
    </ClCompile>
@@ -42,6 +45,9 @@
    <ClCompile Include="..\Modules\_remote_debugging\subprocess.c">
      <Filter>Source Files</Filter>
    </ClCompile>
+    <ClCompile Include="..\Modules\_remote_debugging\interpreters.c">
+      <Filter>Source Files</Filter>
+    </ClCompile>
  </ItemGroup>
  <ItemGroup>
    <ClInclude Include="..\Modules\_remote_debugging\_remote_debugging.h">
@@ -50,6 +56,9 @@
    <ClInclude Include="..\Modules\_remote_debugging\binary_io.h">
      <Filter>Header Files</Filter>
    </ClInclude>
+    <ClInclude Include="..\Modules\_remote_debugging\gc_stats.h">
+      <Filter>Header Files</Filter>
+    </ClInclude>
  </ItemGroup>
  <ItemGroup>
    <ResourceCompile Include="..\PC\python_nt.rc">
@@ -1405,7 +1405,6 @@ add_stats(GCState *gcstate, int gen, struct gc_generation_stats *stats)
    memcpy(cur_stats, prev_stats, sizeof(struct gc_generation_stats));

    cur_stats->ts_start = stats->ts_start;
-    cur_stats->ts_stop = stats->ts_stop;

    cur_stats->collections += 1;
    cur_stats->collected += stats->collected;
@@ -1413,6 +1412,9 @@ add_stats(GCState *gcstate, int gen, struct gc_generation_stats *stats)
    cur_stats->candidates += stats->candidates;

    cur_stats->duration += stats->duration;
+    /* Publish ts_stop last so remote readers do not select a partially
+       updated stats record as the latest collection. */
+    cur_stats->ts_stop = stats->ts_stop;
 }

 /* This is the main function.  Read this to understand how the
@@ -2492,6 +2492,8 @@ gc_collect_main(PyThreadState *tstate, int generation, _PyGC_Reason reason)

    /* Update stats */
    struct gc_generation_stats *stats = get_stats(gcstate, generation);
+    stats->ts_start = start;
+    stats->ts_stop = stop;
    stats->collections++;
    stats->collected += m;
    stats->uncollectable += n;
@@ -318,6 +318,7 @@ MAX_SIZES = {
    _abs('Modules/_hacl/*.c'): (200_000, 500),
    _abs('Modules/posixmodule.c'): (20_000, 500),
    _abs('Modules/termios.c'): (10_000, 800),
+    _abs('Modules/_remote_debugging/debug_offsets_validation.h'): (25_000, 1000),
    _abs('Modules/_remote_debugging/*.h'): (20_000, 1000),
    _abs('Modules/_testcapimodule.c'): (20_000, 400),
    _abs('Modules/expat/expat.h'): (10_000, 400),
@@ -346,7 +347,7 @@ MAX_SIZES = {
    _abs('Modules/_ssl_data_300.h'): (80_000, 10_000),
    _abs('Modules/_ssl_data_111.h'): (80_000, 10_000),
    _abs('Modules/cjkcodecs/mappings_*.h'): (160_000, 2_000),
-    _abs('Modules/clinic/_testclinic.c.h'): (120_000, 5_000),
+    _abs('Modules/clinic/_testclinic.c.h'): (125_000, 5_000),
    _abs('Modules/unicodedata_db.h'): (180_000, 3_000),
    _abs('Modules/unicodename_db.h'): (1_200_000, 15_000),
    _abs('Objects/unicodetype_db.h'): (240_000, 3_000),