Загрузить файлы в «venv/Lib/site-packages/greenlet/tests»

venv/Lib/site-packages/greenlet/tests/test_greenlet_trash.py

@@ -0,0 +1,195 @@
+# -*- coding: utf-8 -*-
+"""
+Tests for greenlets interacting with the CPython trash can API.
+
+The CPython trash can API is not designed to be re-entered from a
+single thread. But this can happen using greenlets, if something
+during the object deallocation process switches greenlets, and this second
+greenlet then causes the trash can to get entered again. Here, we do this
+very explicitly, but in other cases (like gevent) it could be arbitrarily more
+complicated: for example, a weakref callback might try to acquire a lock that's
+already held by another greenlet; that would allow a greenlet switch to occur.
+
+See https://github.com/gevent/gevent/issues/1909
+
+This test is fragile and relies on details of the CPython
+implementation (like most of the rest of this package):
+
+    - We enter the trashcan and deferred deallocation after
+      ``_PyTrash_UNWIND_LEVEL`` calls. This constant, defined in
+      CPython's object.c, is generally 50. That's basically how many objects are required to
+      get us into the deferred deallocation situation.
+
+    - The test fails by hitting an ``assert()`` in object.c; if the
+      build didn't enable assert, then we don't catch this.
+
+    - If the test fails in that way, the interpreter crashes.
+"""
+
+import unittest
+
+
+class TestTrashCanReEnter(unittest.TestCase):
+
+    def test_it(self):
+        try:
+            # pylint:disable-next=no-name-in-module
+            from greenlet._greenlet import get_tstate_trash_delete_nesting
+        except ImportError:
+            import sys
+            # Python 3.13 has no "trash delete nesting" anymore (but
+            # "delete later") Our test as written won't be able to check
+            # the nesting depth anymore, but on debug builds it should
+            # still be able to trigger a crash if we get things wrong.
+            # However, at least on 3.14, the greenlet switch in the test
+            # below never finds an active value for
+            # ``tstate->delete_later``, meaning this test isn't testing
+            # what we want it to.
+            assert sys.version_info[:2] >= (3, 13)
+            def get_tstate_trash_delete_nesting():
+                return 0
+
+        # Try several times to trigger it, because it isn't 100%
+        # reliable.
+        for i in range(30):
+            with self.subTest(i=i):
+                self.check_it(get_tstate_trash_delete_nesting)
+
+    def check_it(self, get_tstate_trash_delete_nesting): # pylint:disable=too-many-statements
+        import greenlet
+        main = greenlet.getcurrent()
+
+        assert get_tstate_trash_delete_nesting() == 0
+
+        # We expect to be in deferred deallocation after this many
+        # deallocations have occurred. TODO: I wish we had a better way to do
+        # this --- that was before get_tstate_trash_delete_nesting; perhaps
+        # we can use that API to do better? (Probably not, it is non-functional in
+        # 3.13+)
+        TRASH_UNWIND_LEVEL = 500 # 50 is the nominal default on 3.12
+        # How many objects to put in a container; it's the container that
+        # queues objects for deferred deallocation.
+        OBJECTS_PER_CONTAINER = 1000
+
+        class Dealloc: # define the class here because we alter class variables each time we run.
+            """
+            An object with a ``__del__`` method. When it starts getting deallocated
+            from a deferred trash can run, it switches greenlets, allocates more objects
+            which then also go in the trash can. If we don't save state appropriately,
+            nesting gets out of order and we can crash the interpreter.
+            """
+
+            #: Has our deallocation actually run and switched greenlets?
+            #: When it does, this will be set to the current greenlet. This should
+            #: be happening in the main greenlet, so we check that down below.
+            SPAWNED = False
+
+            #: Has the background greenlet run?
+            BG_RAN = False
+
+            BG_GLET = None
+
+            #: How many of these things have ever been allocated.
+            CREATED = 0
+
+            #: How many of these things have ever been deallocated.
+            DESTROYED = 0
+
+            #: How many were destroyed not in the main greenlet. There should always
+            #: be some.
+            #: If the test is broken or things change in the trashcan implementation,
+            #: this may not be correct.
+            DESTROYED_BG = 0
+
+            def __init__(self, sequence_number):
+                """
+                :param sequence_number: The ordinal of this object during
+                   one particular creation run. This is used to detect (guess, really)
+                   when we have entered the trash can's deferred deallocation.
+                """
+                self.i = sequence_number
+                Dealloc.CREATED += 1
+
+            def __del__(self):
+                if self.i == TRASH_UNWIND_LEVEL and not self.SPAWNED:
+                    Dealloc.SPAWNED = greenlet.getcurrent()
+                    other = Dealloc.BG_GLET = greenlet.greenlet(background_greenlet)
+                    x = other.switch()
+                    assert x == 42
+                    # It's important that we don't switch back to the greenlet,
+                    # we leave it hanging there in an incomplete state. But we don't let it
+                    # get collected, either. If we complete it now, while we're still
+                    # in the scope of the initial trash can, things work out and we
+                    # don't see the problem. We need this greenlet to complete
+                    # at some point in the future, after we've exited this trash can invocation.
+                    del other
+                elif self.i == 40 and greenlet.getcurrent() is not main:
+                    Dealloc.BG_RAN = True
+                    try:
+                        main.switch(42)
+                    except greenlet.GreenletExit as ex:
+                        # We expect this; all references to us go away
+                        # while we're still running, and we need to finish deleting
+                        # ourself.
+                        Dealloc.BG_RAN = type(ex)
+                        del ex
+
+                # Record the fact that we're dead last of all. This ensures that
+                # we actually get returned too.
+                Dealloc.DESTROYED += 1
+                if greenlet.getcurrent() is not main:
+                    Dealloc.DESTROYED_BG += 1
+
+
+        def background_greenlet():
+            # We direct through a second function, instead of
+            # directly calling ``make_some()``, so that we have complete
+            # control over when these objects are destroyed: we need them
+            # to be destroyed in the context of the background greenlet
+            t = make_some()
+            del t # Triggere deletion.
+
+        def make_some():
+            t = ()
+            i = OBJECTS_PER_CONTAINER
+            while i:
+                # Nest the tuples; it's the recursion that gets us
+                # into trash.
+                t = (Dealloc(i), t)
+                i -= 1
+            return t
+
+
+        some = make_some()
+        self.assertEqual(Dealloc.CREATED, OBJECTS_PER_CONTAINER)
+        self.assertEqual(Dealloc.DESTROYED, 0)
+
+        # If we're going to crash, it should be on the following line.
+        # We only crash if ``assert()`` is enabled, of course.
+        del some
+
+        # For non-debug builds of CPython, we won't crash. The best we can do is check
+        # the nesting level explicitly.
+        self.assertEqual(0, get_tstate_trash_delete_nesting())
+
+        # Discard this, raising GreenletExit into where it is waiting.
+        Dealloc.BG_GLET = None
+        # The same nesting level maintains.
+        self.assertEqual(0, get_tstate_trash_delete_nesting())
+
+        # We definitely cleaned some up in the background
+        self.assertGreater(Dealloc.DESTROYED_BG, 0)
+
+        # Make sure all the cleanups happened.
+        self.assertIs(Dealloc.SPAWNED, main)
+        self.assertTrue(Dealloc.BG_RAN)
+        self.assertEqual(Dealloc.BG_RAN, greenlet.GreenletExit)
+        self.assertEqual(Dealloc.CREATED, Dealloc.DESTROYED )
+        self.assertEqual(Dealloc.CREATED, OBJECTS_PER_CONTAINER * 2)
+
+        import gc
+        gc.collect()
+
+
+if __name__ == '__main__':
+    unittest.main()

venv/Lib/site-packages/greenlet/tests/test_interpreter_shutdown.py

@@ -0,0 +1,821 @@
+# -*- coding: utf-8 -*-
+"""
+Tests for greenlet behavior during interpreter shutdown (Py_FinalizeEx).
+
+These tests are organized into four groups:
+
+  A. Core safety (smoke): no crashes with active greenlets at shutdown.
+  B. Cleanup semantics: GreenletExit / finally still works during
+     normal thread exit (the standard production path).
+  C. Atexit "still works" tests: getcurrent() / greenlet construction
+     during atexit handlers registered AFTER greenlet import (i.e.
+     BEFORE greenlet's cleanup handler in LIFO order) must return
+     valid objects — verifies the guards don't over-block.
+  D. TDD-certified regression tests: getcurrent() must return None
+     when called AFTER greenlet's cleanup (GC finalization phase
+     or late atexit phase).  These tests fail on greenlet 3.3.2
+     and pass with the fix across Python 3.10-3.14.
+"""
+import sys
+import subprocess
+import unittest
+import textwrap
+
+from greenlet.tests import TestCase
+
+
+class TestInterpreterShutdown(TestCase): # pylint:disable=too-many-public-methods
+
+    def _run_shutdown_script(self, script_body):
+        """
+        Run a Python script in a subprocess that exercises greenlet
+        during interpreter shutdown. Returns (returncode, stdout, stderr).
+        """
+        full_script = textwrap.dedent(script_body)
+        result = subprocess.run(
+            [sys.executable, '-c', full_script],
+            capture_output=True,
+            text=True,
+            timeout=30,
+            check=False,
+        )
+        return result.returncode, result.stdout, result.stderr
+
+    # -----------------------------------------------------------------
+    # Group A: Core safety — no crashes with active greenlets at exit
+    # -----------------------------------------------------------------
+
+    def test_active_greenlet_at_shutdown_no_crash(self):
+
+        # An active (suspended) greenlet that is deallocated during
+        # interpreter shutdown should not crash the process.
+
+        # Before the fix, this would SIGSEGV on Python < 3.11 because
+        # _green_dealloc_kill_started_non_main_greenlet tried to call
+        # g_switch() during Py_FinalizeEx.
+
+        rc, stdout, stderr = self._run_shutdown_script("""\
+            import greenlet
+
+            def worker():
+                greenlet.getcurrent().parent.switch("from worker")
+                return "done"
+
+            g = greenlet.greenlet(worker)
+            result = g.switch()
+            assert result == "from worker", result
+            print("OK: exiting with active greenlet")
+        """)
+        self.assertEqual(rc, 0, f"Process crashed (rc={rc}):\n{stdout}{stderr}")
+        self.assertIn("OK: exiting with active greenlet", stdout)
+
+    def test_multiple_active_greenlets_at_shutdown(self):
+        # Multiple suspended greenlets at shutdown should all be cleaned
+        # up without crashing.
+
+        rc, stdout, stderr = self._run_shutdown_script("""\
+            import greenlet
+
+            def worker(name):
+                greenlet.getcurrent().parent.switch(f"hello from {name}")
+                return "done"
+
+            greenlets = []
+            for i in range(10):
+                g = greenlet.greenlet(worker)
+                result = g.switch(f"g{i}")
+                greenlets.append(g)
+
+            print(f"OK: {len(greenlets)} active greenlets at shutdown")
+        """)
+        self.assertEqual(rc, 0, f"Process crashed (rc={rc}):\n{stdout}{stderr}")
+        self.assertIn("OK: 10 active greenlets at shutdown", stdout)
+
+    def test_nested_greenlets_at_shutdown(self):
+        # Nested (chained parent) greenlets at shutdown should not crash.
+
+        rc, stdout, stderr = self._run_shutdown_script("""\
+            import greenlet
+
+            def inner():
+                greenlet.getcurrent().parent.switch("inner done")
+
+            def outer():
+                g_inner = greenlet.greenlet(inner)
+                g_inner.switch()
+                greenlet.getcurrent().parent.switch("outer done")
+
+            g = greenlet.greenlet(outer)
+            result = g.switch()
+            assert result == "outer done", result
+            print("OK: nested greenlets at shutdown")
+        """)
+        self.assertEqual(rc, 0, f"Process crashed (rc={rc}):\n{stdout}{stderr}")
+        self.assertIn("OK: nested greenlets at shutdown", stdout)
+
+    def test_threaded_greenlets_at_shutdown(self):
+        # Greenlets in worker threads that are still referenced at
+        # shutdown should not crash.
+
+        rc, stdout, stderr = self._run_shutdown_script("""\
+            import greenlet
+            import threading
+
+            results = []
+
+            def thread_worker():
+                def greenlet_func():
+                    greenlet.getcurrent().parent.switch("from thread greenlet")
+                    return "done"
+
+                g = greenlet.greenlet(greenlet_func)
+                val = g.switch()
+                results.append((g, val))
+
+            threads = []
+            for _ in range(3):
+                t = threading.Thread(target=thread_worker)
+                t.start()
+                threads.append(t)
+
+            for t in threads:
+                t.join()
+
+            print(f"OK: {len(results)} threaded greenlets at shutdown")
+        """)
+        self.assertEqual(rc, 0, f"Process crashed (rc={rc}):\n{stdout}{stderr}")
+        self.assertIn("OK: 3 threaded greenlets at shutdown", stdout)
+
+    # -----------------------------------------------------------------
+    # Group B: Cleanup semantics — thread exit
+    # -----------------------------------------------------------------
+    #
+    # These tests verify that GreenletExit / try-finally still work
+    # correctly during normal thread exit (the standard production
+    # path, e.g. uWSGI worker threads finishing a request).  This is
+    # NOT interpreter shutdown; the guards do not fire here.
+
+    def test_greenlet_cleanup_during_thread_exit(self):
+        # When a thread exits normally while holding active greenlets,
+        # GreenletExit IS thrown and cleanup code runs.  This is the
+        # standard cleanup path used in production (e.g. uWSGI worker
+        # threads finishing a request).
+
+        rc, stdout, stderr = self._run_shutdown_script("""\
+            import os
+            import threading
+            import greenlet
+
+            _write = os.write
+
+            def thread_func():
+                def worker(_w=_write,
+                           _GreenletExit=greenlet.GreenletExit):
+                    try:
+                        greenlet.getcurrent().parent.switch("suspended")
+                    except _GreenletExit:
+                        _w(1, b"CLEANUP: GreenletExit caught\\n")
+                        raise
+
+                g = greenlet.greenlet(worker)
+                g.switch()
+                # Thread exits with active greenlet -> thread-state
+                # cleanup triggers GreenletExit
+
+            t = threading.Thread(target=thread_func)
+            t.start()
+            t.join()
+            print("OK: thread cleanup done")
+        """)
+        self.assertEqual(rc, 0, f"Process crashed (rc={rc}):\n{stdout}{stderr}")
+        self.assertIn("OK: thread cleanup done", stdout)
+        self.assertIn("CLEANUP: GreenletExit caught", stdout)
+
+    def test_finally_block_during_thread_exit(self):
+        # try/finally blocks in active greenlets run correctly when the
+        # owning thread exits.
+        rc, stdout, stderr = self._run_shutdown_script("""\
+            import os
+            import threading
+            import greenlet
+
+            _write = os.write
+
+            def thread_func():
+                def worker(_w=_write):
+                    try:
+                        greenlet.getcurrent().parent.switch("suspended")
+                    finally:
+                        _w(1, b"FINALLY: cleanup executed\\n")
+
+                g = greenlet.greenlet(worker)
+                g.switch()
+
+            t = threading.Thread(target=thread_func)
+            t.start()
+            t.join()
+            print("OK: thread cleanup done")
+        """)
+        self.assertEqual(rc, 0, f"Process crashed (rc={rc}):\n{stdout}{stderr}")
+        self.assertIn("OK: thread cleanup done", stdout)
+        self.assertIn("FINALLY: cleanup executed", stdout)
+
+    def test_many_greenlets_with_cleanup_at_shutdown(self):
+        # Stress test: many active greenlets with cleanup code at shutdown.
+        # Ensures no crashes regardless of deallocation order.
+
+        rc, stdout, stderr = self._run_shutdown_script("""\
+            import sys
+            import greenlet
+
+            cleanup_count = 0
+
+            def worker(idx):
+                global cleanup_count
+                try:
+                    greenlet.getcurrent().parent.switch(f"ready-{idx}")
+                except greenlet.GreenletExit:
+                    cleanup_count += 1
+                    raise
+
+            greenlets = []
+            for i in range(50):
+                g = greenlet.greenlet(worker)
+                result = g.switch(i)
+                greenlets.append(g)
+
+            print(f"OK: {len(greenlets)} greenlets about to shut down")
+            # Note: we can't easily print cleanup_count during shutdown
+            # since it happens after the main module's code runs.
+        """)
+        self.assertEqual(rc, 0, f"Process crashed (rc={rc}):\n{stdout}{stderr}")
+        self.assertIn("OK: 50 greenlets about to shut down", stdout)
+
+    def test_deeply_nested_greenlets_at_shutdown(self):
+        # Deeply nested greenlet parent chains at shutdown.
+        # Tests that the deallocation order doesn't cause issues.
+
+        rc, stdout, stderr = self._run_shutdown_script("""\
+            import greenlet
+
+            def level(depth, max_depth):
+                if depth < max_depth:
+                    g = greenlet.greenlet(level)
+                    g.switch(depth + 1, max_depth)
+                greenlet.getcurrent().parent.switch(f"depth-{depth}")
+
+            g = greenlet.greenlet(level)
+            result = g.switch(0, 10)
+            print(f"OK: nested to depth 10, got {result}")
+        """)
+        self.assertEqual(rc, 0, f"Process crashed (rc={rc}):\n{stdout}{stderr}")
+        self.assertIn("OK: nested to depth 10", stdout)
+
+    def test_greenlet_with_traceback_at_shutdown(self):
+        # A greenlet that has an active exception context when it's
+        # suspended should not crash during shutdown cleanup.
+
+        rc, stdout, stderr = self._run_shutdown_script("""\
+            import greenlet
+
+            def worker():
+                try:
+                    raise ValueError("test error")
+                except ValueError:
+                    # Suspend while an exception is active on the stack
+                    greenlet.getcurrent().parent.switch("suspended with exc")
+                return "done"
+
+            g = greenlet.greenlet(worker)
+            result = g.switch()
+            assert result == "suspended with exc"
+            print("OK: greenlet with active exception at shutdown")
+        """)
+        self.assertEqual(rc, 0, f"Process crashed (rc={rc}):\n{stdout}{stderr}")
+        self.assertIn("OK: greenlet with active exception at shutdown", stdout)
+
+    # -----------------------------------------------------------------
+    # Group C: getcurrent() / construction / gettrace() / settrace()
+    # during atexit — registered AFTER greenlet import
+    # -----------------------------------------------------------------
+
+    def test_getcurrent_during_atexit_no_crash(self):
+        # getcurrent() in an atexit handler registered AFTER greenlet
+        # import must return a valid greenlet (not None), because LIFO
+        # ordering means this handler runs BEFORE greenlet's cleanup.
+
+        rc, stdout, stderr = self._run_shutdown_script("""\
+            import atexit
+            import greenlet
+
+            def call_getcurrent_at_exit():
+                try:
+                    g = greenlet.getcurrent()
+                    if g is not None and type(g).__name__ == 'greenlet':
+                        print(f"OK: getcurrent returned valid greenlet")
+                    elif g is None:
+                        print("FAIL: getcurrent returned None (over-blocked)")
+                    else:
+                        print(f"FAIL: unexpected {g!r}")
+                except Exception as e:
+                    print(f"OK: getcurrent raised {type(e).__name__}: {e}")
+
+            atexit.register(call_getcurrent_at_exit)
+            print("OK: atexit registered")
+        """)
+        self.assertEqual(rc, 0, f"Process crashed (rc={rc}):\n{stdout}{stderr}")
+        self.assertIn("OK: atexit registered", stdout)
+        self.assertIn("OK: getcurrent returned valid greenlet", stdout,
+                      "getcurrent() should return a valid greenlet when called "
+                      "before greenlet's cleanup handler (LIFO ordering)")
+
+    def test_gettrace_during_atexit_no_crash(self):
+        # Calling greenlet.gettrace() during atexit must not crash.
+
+        rc, stdout, stderr = self._run_shutdown_script("""\
+            import atexit
+            import greenlet
+
+            def check_at_exit():
+                try:
+                    result = greenlet.gettrace()
+                    print(f"OK: gettrace returned {result!r}")
+                except Exception as e:
+                    print(f"OK: gettrace raised {type(e).__name__}: {e}")
+
+            atexit.register(check_at_exit)
+            print("OK: registered")
+        """)
+        self.assertEqual(rc, 0, f"Process crashed (rc={rc}):\n{stdout}{stderr}")
+        self.assertIn("OK: registered", stdout)
+
+    def test_settrace_during_atexit_no_crash(self):
+        # Calling greenlet.settrace() during atexit must not crash.
+
+        rc, stdout, stderr = self._run_shutdown_script("""\
+            import atexit
+            import greenlet
+
+            def check_at_exit():
+                try:
+                    greenlet.settrace(lambda *args: None)
+                    print("OK: settrace succeeded")
+                except Exception as e:
+                    print(f"OK: settrace raised {type(e).__name__}: {e}")
+
+            atexit.register(check_at_exit)
+            print("OK: registered")
+        """)
+        self.assertEqual(rc, 0, f"Process crashed (rc={rc}):\n{stdout}{stderr}")
+        self.assertIn("OK: registered", stdout)
+
+    def test_getcurrent_with_active_greenlets_during_atexit(self):
+        # getcurrent() during atexit (registered after import) with active
+        # greenlets must still return a valid greenlet, since LIFO means
+        # this runs before greenlet's cleanup.
+
+        rc, stdout, stderr = self._run_shutdown_script("""\
+            import atexit
+            import greenlet
+
+            def worker():
+                greenlet.getcurrent().parent.switch("ready")
+
+            greenlets = []
+            for i in range(5):
+                g = greenlet.greenlet(worker)
+                result = g.switch()
+                greenlets.append(g)
+
+            def check_at_exit():
+                try:
+                    g = greenlet.getcurrent()
+                    if g is not None and type(g).__name__ == 'greenlet':
+                        print(f"OK: getcurrent returned valid greenlet")
+                    elif g is None:
+                        print("FAIL: getcurrent returned None (over-blocked)")
+                    else:
+                        print(f"FAIL: unexpected {g!r}")
+                except Exception as e:
+                    print(f"OK: getcurrent raised {type(e).__name__}: {e}")
+
+            atexit.register(check_at_exit)
+            print(f"OK: {len(greenlets)} active greenlets, atexit registered")
+        """)
+        self.assertEqual(rc, 0, f"Process crashed (rc={rc}):\n{stdout}{stderr}")
+        self.assertIn("OK: 5 active greenlets, atexit registered", stdout)
+        self.assertIn("OK: getcurrent returned valid greenlet", stdout,
+                      "getcurrent() should return a valid greenlet when called "
+                      "before greenlet's cleanup handler (LIFO ordering)")
+
+    def test_greenlet_construction_during_atexit_no_crash(self):
+        # Constructing a new greenlet during atexit (registered after
+        # import) must succeed, since this runs before greenlet's cleanup.
+
+        rc, stdout, stderr = self._run_shutdown_script("""\
+            import atexit
+            import greenlet
+
+            def create_greenlets_at_exit():
+                try:
+                    def noop():
+                        pass
+                    g = greenlet.greenlet(noop)
+                    if g is not None:
+                        print(f"OK: created greenlet successfully")
+                    else:
+                        print("FAIL: greenlet() returned None")
+                except Exception as e:
+                    print(f"OK: construction raised {type(e).__name__}: {e}")
+
+            atexit.register(create_greenlets_at_exit)
+            print("OK: atexit registered")
+        """)
+        self.assertEqual(rc, 0, f"Process crashed (rc={rc}):\n{stdout}{stderr}")
+        self.assertIn("OK: atexit registered", stdout)
+        self.assertIn("OK: created greenlet successfully", stdout)
+
+    def test_greenlet_construction_with_active_greenlets_during_atexit(self):
+        # Constructing new greenlets during atexit when other active
+        # greenlets already exist (maximizes the chance of a non-empty
+        # deleteme list).
+
+        rc, stdout, stderr = self._run_shutdown_script("""\
+            import atexit
+            import greenlet
+
+            def worker():
+                greenlet.getcurrent().parent.switch("ready")
+
+            greenlets = []
+            for i in range(10):
+                g = greenlet.greenlet(worker)
+                g.switch()
+                greenlets.append(g)
+
+            def create_at_exit():
+                try:
+                    new_greenlets = []
+                    for i in range(5):
+                        g = greenlet.greenlet(lambda: None)
+                        new_greenlets.append(g)
+                    print(f"OK: created {len(new_greenlets)} greenlets at exit")
+                except Exception as e:
+                    print(f"OK: raised {type(e).__name__}: {e}")
+
+            atexit.register(create_at_exit)
+            print(f"OK: {len(greenlets)} active greenlets, atexit registered")
+        """)
+        self.assertEqual(rc, 0, f"Process crashed (rc={rc}):\n{stdout}{stderr}")
+        self.assertIn("OK: 10 active greenlets, atexit registered", stdout)
+
+    def test_greenlet_construction_with_cross_thread_deleteme_during_atexit(self):
+        # Create greenlets in a worker thread, transfer them to the main
+        # thread, then drop them — populating the deleteme list. Then
+        # construct a new greenlet during atexit. On Python < 3.11
+        # clear_deleteme_list() could previously crash if the
+        # PythonAllocator vector copy failed during early Py_FinalizeEx;
+        # using std::swap eliminates that allocation.
+        rc, stdout, stderr = self._run_shutdown_script("""\
+            import atexit
+            import greenlet
+            import threading
+
+            cross_thread_refs = []
+
+            def thread_worker():
+                # Create greenlets in this thread
+                def gl_body():
+                    greenlet.getcurrent().parent.switch("ready")
+                for _ in range(20):
+                    g = greenlet.greenlet(gl_body)
+                    g.switch()
+                    cross_thread_refs.append(g)
+
+            t = threading.Thread(target=thread_worker)
+            t.start()
+            t.join()
+
+            # Dropping these references in the main thread
+            # causes them to be added to the main thread's
+            # deleteme list (deferred cross-thread dealloc).
+            cross_thread_refs.clear()
+
+            def create_at_exit():
+                try:
+                    g = greenlet.greenlet(lambda: None)
+                    print(f"OK: created greenlet at exit {g!r}")
+                except Exception as e:
+                    print(f"OK: raised {type(e).__name__}: {e}")
+
+            atexit.register(create_at_exit)
+            print("OK: cross-thread setup done, atexit registered")
+        """)
+        self.assertEqual(rc, 0, f"Process crashed (rc={rc}):\n{stdout}{stderr}")
+        self.assertIn("OK: cross-thread setup done, atexit registered", stdout)
+
+
+    # -----------------------------------------------------------------
+    # Group D.1: TDD-certified — getcurrent() during GC finalization
+    #
+    # These tests use gc.disable() + reference cycles to force __del__
+    # to run during Py_FinalizeEx's GC pass, where Py_IsFinalizing()
+    # is True.  Without the fix, getcurrent() returns a live greenlet
+    # (unguarded); with the fix, it returns None.
+    #
+    # TDD verification (greenlet 3.3.2 = RED, patched = GREEN):
+    #   Python 3.10: RED (UNGUARDED) → GREEN (GUARDED)
+    #   Python 3.11: RED (UNGUARDED) → GREEN (GUARDED)
+    #   Python 3.12: RED (UNGUARDED) → GREEN (GUARDED)
+    #   Python 3.13: RED (UNGUARDED) → GREEN (GUARDED)
+    #   Python 3.14: RED (UNGUARDED) → GREEN (GUARDED)
+    # -----------------------------------------------------------------
+
+    def test_getcurrent_returns_none_during_gc_finalization(self):
+        # greenlet.getcurrent() must raise an exception when called from a
+        # __del__ method during Py_FinalizeEx's GC collection pass.
+
+        # On Python >= 3.11, _Py_IsFinalizing() is True during this
+        # phase.  Without the Py_IsFinalizing() guard in mod_getcurrent,
+        # this would return a greenlet — the same unguarded code path
+        # that leads to SIGSEGV in production (uWSGI worker recycling).
+        rc, stdout, stderr = self._run_shutdown_script("""\
+            import gc
+            import os
+            import greenlet
+
+            gc.disable()
+
+            class CleanupChecker:
+                def __del__(self):
+                    try:
+                        try:
+                            greenlet.getcurrent()
+                        except RuntimeError:
+                            os.write(1, b"GUARDED: getcurrent=None\\n")
+                        else:
+                            os.write(1, b"UNGUARDED: getcurrent="
+                                     + type(cur).__name__.encode() + b"\\n")
+                    except Exception as e:
+                        os.write(1, b"EXCEPTION: " + str(e).encode() + b"\\n")
+
+            # Reference cycle: only collected during Py_FinalizeEx GC pass
+            a = CleanupChecker()
+            b = {"ref": a}
+            a._cycle = b
+            del a, b
+
+            print("OK: deferred cycle created")
+        """)
+        self.assertEqual(rc, 0, f"Process crashed (rc={rc}):\n{stdout}{stderr}")
+        self.assertIn("OK: deferred cycle created", stdout)
+        self.assertIn("GUARDED: getcurrent=None", stdout)
+
+    def test_getcurrent_returns_none_during_gc_finalization_with_active_greenlets(self):
+        # Same as above but with active greenlets at shutdown, which
+        # increases the amount of C++ destructor work during finalization.
+
+        rc, stdout, stderr = self._run_shutdown_script("""\
+            import gc
+            import os
+            import greenlet
+
+            gc.disable()
+
+            class CleanupChecker:
+                def __del__(self):
+                    try:
+                        try:
+                            greenlet.getcurrent()
+                        except RuntimeError:
+                            os.write(1, b"GUARDED: getcurrent=None\\n")
+                        else:
+                            os.write(1, b"UNGUARDED: getcurrent="
+                                     + type(cur).__name__.encode() + b"\\n")
+                    except Exception as e:
+                        os.write(1, b"EXCEPTION: " + str(e).encode() + b"\\n")
+
+            # Create active greenlets
+            greenlets = []
+            for _ in range(10):
+                def worker():
+                    greenlet.getcurrent().parent.switch("suspended")
+                g = greenlet.greenlet(worker)
+                g.switch()
+                greenlets.append(g)
+
+            # Reference cycle deferred to Py_FinalizeEx
+            a = CleanupChecker()
+            b = {"ref": a}
+            a._cycle = b
+            del a, b
+
+            print(f"OK: {len(greenlets)} active greenlets, cycle deferred")
+        """)
+        self.assertEqual(rc, 0, f"Process crashed (rc={rc}):\n{stdout}{stderr}")
+        self.assertIn("OK: 10 active greenlets, cycle deferred", stdout)
+        self.assertIn("GUARDED: getcurrent=None", stdout)
+
+    def test_getcurrent_returns_none_during_gc_finalization_cross_thread(self):
+        # Combines cross-thread greenlet deallocation (deleteme list)
+        # with the GC finalization check.  This simulates the production
+        # scenario where uWSGI worker threads create greenlets that are
+        # transferred to the main thread, then cleaned up during
+        # Py_FinalizeEx.
+
+        rc, stdout, stderr = self._run_shutdown_script("""\
+            import gc
+            import os
+            import threading
+            import greenlet
+
+            gc.disable()
+
+            class CleanupChecker:
+                def __del__(self):
+                    try:
+                        try:
+                            greenlet.getcurrent()
+                        except RuntimeError:
+                            os.write(1, b"GUARDED: getcurrent=None\\n")
+                        else:
+                            os.write(1, b"UNGUARDED: getcurrent="
+                                     + type(cur).__name__.encode() + b"\\n")
+                    except Exception as e:
+                        os.write(1, b"EXCEPTION: " + str(e).encode() + b"\\n")
+
+            # Create cross-thread greenlet references
+            cross_refs = []
+            def thread_fn():
+                for _ in range(20):
+                    def body():
+                        greenlet.getcurrent().parent.switch("x")
+                    g = greenlet.greenlet(body)
+                    g.switch()
+                    cross_refs.append(g)
+            t = threading.Thread(target=thread_fn)
+            t.start()
+            t.join()
+            cross_refs.clear()
+
+            # Reference cycle deferred to Py_FinalizeEx
+            a = CleanupChecker()
+            b = {"ref": a}
+            a._cycle = b
+            del a, b
+
+            print("OK: cross-thread cleanup + cycle deferred")
+        """)
+        self.assertEqual(rc, 0, f"Process crashed (rc={rc}):\n{stdout}{stderr}")
+        self.assertIn("OK: cross-thread cleanup + cycle deferred", stdout)
+        self.assertIn("GUARDED: getcurrent=None", stdout)
+
+
+    # -----------------------------------------------------------------
+    # Group D.2: TDD-certified — getcurrent() during atexit phase
+    #
+    # These tests register the checker BEFORE importing greenlet.
+    # Python's atexit is LIFO, so greenlet's handler (registered at
+    # import) runs FIRST and sets g_greenlet_shutting_down=1; then
+    # the checker runs SECOND and observes getcurrent() → None.
+    #
+    # This covers the atexit phase where _Py_IsFinalizing() is still
+    # False on ALL Python versions — the exact window that causes
+    # SIGSEGV in production (uWSGI worker recycling → Py_FinalizeEx).
+    #
+    # TDD verification (greenlet 3.3.2 = RED, patched = GREEN):
+    #   Python 3.10: RED (UNGUARDED) → GREEN (GUARDED)
+    #   Python 3.11: RED (UNGUARDED) → GREEN (GUARDED)
+    #   Python 3.12: RED (UNGUARDED) → GREEN (GUARDED)
+    #   Python 3.13: RED (UNGUARDED) → GREEN (GUARDED)
+    #   Python 3.14: RED (UNGUARDED) → GREEN (GUARDED)
+    # -----------------------------------------------------------------
+
+    def test_getcurrent_returns_none_during_atexit_phase(self):
+        # greenlet.getcurrent() must NOT return None when called from an
+        # atexit handler that runs AFTER greenlet's own atexit handler.
+
+        rc, stdout, stderr = self._run_shutdown_script("""\
+            import atexit
+            import os
+
+            def late_checker():
+                try:
+                    import greenlet
+                    cur = greenlet.getcurrent()
+                    if cur is None:
+                        os.write(1, b"GUARDED: getcurrent=None\\n")
+                    else:
+                        os.write(1, b"UNGUARDED: getcurrent="
+                                 + type(cur).__name__.encode() + b"\\n")
+                except Exception as e:
+                    os.write(1, b"EXCEPTION: " + str(e).encode() + b"\\n")
+
+            # Register BEFORE importing greenlet.  LIFO order:
+            # greenlet's handler (registered at import) runs FIRST,
+            # late_checker runs SECOND — seeing the flag already set.
+            atexit.register(late_checker)
+
+            import greenlet
+            print("OK: atexit registered before greenlet import")
+        """)
+        self.assertEqual(rc, 0, f"Process crashed (rc={rc}):\n{stdout}{stderr}")
+        self.assertIn("OK: atexit registered before greenlet import", stdout)
+        self.assertIn("UNGUARDED", stdout)
+
+
+    def test_getcurrent_returns_none_during_atexit_phase_with_active_greenlets(self):
+        # Same as above but with active greenlets
+        rc, stdout, stderr = self._run_shutdown_script("""\
+            import atexit
+            import os
+
+            def late_checker():
+                try:
+                    import greenlet
+                    cur = greenlet.getcurrent()
+                    if cur is None:
+                        os.write(1, b"GUARDED: getcurrent=None\\n")
+                    else:
+                        os.write(1, b"UNGUARDED: getcurrent="
+                                 + type(cur).__name__.encode() + b"\\n")
+                except Exception as e:
+                    os.write(1, b"EXCEPTION: " + str(e).encode() + b"\\n")
+
+            atexit.register(late_checker)
+
+            import greenlet
+
+            greenlets = []
+            for _ in range(10):
+                def worker():
+                    greenlet.getcurrent().parent.switch("parked")
+                g = greenlet.greenlet(worker)
+                g.switch()
+                greenlets.append(g)
+
+            print(f"OK: {len(greenlets)} active greenlets, atexit registered")
+        """)
+        self.assertEqual(rc, 0, f"Process crashed (rc={rc}):\n{stdout}{stderr}")
+        self.assertIn("OK: 10 active greenlets, atexit registered", stdout)
+        self.assertIn("UNGUARDED", stdout)
+
+    def test_api_getcurrent_no_system_error_at_module_gc_time(self):
+        # If we use the C API directly to return a greenlet AFTER
+        # atexit threads have been run, we don't crash, we get a
+        # specific error. We arrange for this by putting a __del__ on
+        # an object that lives in greenlet's own (extension module)
+        # dict; this is cleaned out sometime during the module cleanup
+        # steps.
+        rc, stdout, stderr = self._run_shutdown_script("""\
+            import greenlet
+            from greenlet.tests import _test_extension
+
+            class WithDel:
+                # must cache the method we want, because by the time we
+                # run, module globals may have been cleaned up.
+                def __del__(self, gc=_test_extension.getcurrent_api):
+                    print('Destructor running')
+                    gc() # Should print an unraisable RuntimeException
+
+            greenlet._greenlet.with_del = WithDel()
+        """)
+        self.assertEqual(rc, 0, f"Process crashed (rc={rc}):\n{stdout}{stderr}")
+        self.assertIn('Destructor running', stdout)
+        self.assertIn('RuntimeError: greenlet is being finalized', stderr)
+
+
+    def test_switch_no_error_at_module_gc_time(self):
+        # Switching to a greenlet we've captured during
+        # module tear down doesn't cause a crash
+        rc, stdout, stderr = self._run_shutdown_script("""\
+            import greenlet
+            from greenlet.tests import _test_extension
+
+            gs = []
+            # must cache the objects we want, because by the time we
+            # run, module globals may have been cleaned up.
+            def do_it(gs=gs):
+                print('current', gs)
+                gs[0].parent.switch(1)
+
+
+            gs.append(greenlet.greenlet(do_it))
+            gs.append(greenlet.greenlet(do_it))
+            gs[1].switch()
+
+            class WithDel:
+                def __del__(self, gs=gs):
+                    print('Destructor running')
+                    r = gs[0].switch()
+                    print('Result', r)
+
+            greenlet._greenlet.with_del = WithDel()
+        """)
+        self.assertEqual(rc, 0, f"Process crashed (rc={rc}):\n{stdout}{stderr}")
+        self.assertIn('Destructor running', stdout)
+        self.assertIn('Result 1', stdout)
+
+
+if __name__ == '__main__':
+    unittest.main()

venv/Lib/site-packages/greenlet/tests/test_leaks.py

@@ -0,0 +1,497 @@
+# -*- coding: utf-8 -*-
+"""
+Testing scenarios that may have leaked.
+"""
+import sys
+import gc
+
+import time
+import weakref
+import threading
+
+
+import greenlet
+from . import TestCase
+from . import PY314
+from . import RUNNING_ON_FREETHREAD_BUILD
+from . import WIN
+from .leakcheck import fails_leakcheck
+from .leakcheck import ignores_leakcheck
+from .leakcheck import RUNNING_ON_MANYLINUX
+
+
+# pylint:disable=protected-access
+
+assert greenlet.GREENLET_USE_GC # Option to disable this was removed in 1.0
+
+class HasFinalizerTracksInstances(object):
+    EXTANT_INSTANCES = set()
+    def __init__(self, msg):
+        self.msg = sys.intern(msg)
+        self.EXTANT_INSTANCES.add(id(self))
+    def __del__(self):
+        self.EXTANT_INSTANCES.remove(id(self))
+    def __repr__(self):
+        return "<HasFinalizerTracksInstances at 0x%x %r>" % (
+            id(self), self.msg
+        )
+    @classmethod
+    def reset(cls):
+        cls.EXTANT_INSTANCES.clear()
+
+
+def fails_leakcheck_except_on_free_thraded(func):
+    if RUNNING_ON_FREETHREAD_BUILD:
+        # These all seem to pass on free threading because
+        # of the changes to the garbage collector
+        return func
+    return fails_leakcheck(func)
+
+
+class TestLeaks(TestCase):
+
+    def test_arg_refs(self):
+        args = ('a', 'b', 'c')
+        refcount_before = sys.getrefcount(args)
+        # pylint:disable=unnecessary-lambda
+        g = greenlet.greenlet(
+            lambda *args: greenlet.getcurrent().parent.switch(*args))
+        for _ in range(100):
+            g.switch(*args)
+        self.assertEqual(sys.getrefcount(args), refcount_before)
+
+    def test_kwarg_refs(self):
+        kwargs = {}
+        self.assertEqual(sys.getrefcount(kwargs), 2 if not PY314 else 1)
+        # pylint:disable=unnecessary-lambda
+        g = greenlet.greenlet(
+            lambda **gkwargs: greenlet.getcurrent().parent.switch(**gkwargs))
+        for _ in range(100):
+            g.switch(**kwargs)
+        # Python 3.14 elides reference counting operations
+        # in some cases. See https://github.com/python/cpython/pull/130708
+        self.assertEqual(sys.getrefcount(kwargs), 2 if not PY314 else 1)
+
+
+    @staticmethod
+    def __recycle_threads():
+        # By introducing a thread that does sleep we allow other threads,
+        # that have triggered their __block condition, but did not have a
+        # chance to deallocate their thread state yet, to finally do so.
+        # The way it works is by requiring a GIL switch (different thread),
+        # which does a GIL release (sleep), which might do a GIL switch
+        # to finished threads and allow them to clean up.
+        def worker():
+            time.sleep(0.001)
+        t = threading.Thread(target=worker)
+        t.start()
+        time.sleep(0.001)
+        t.join(10)
+
+    def test_threaded_leak(self):
+        gg = []
+        def worker():
+            # only main greenlet present
+            gg.append(weakref.ref(greenlet.getcurrent()))
+        for _ in range(2):
+            t = threading.Thread(target=worker)
+            t.start()
+            t.join(10)
+            del t
+        greenlet.getcurrent() # update ts_current
+        self.__recycle_threads()
+        greenlet.getcurrent() # update ts_current
+        gc.collect()
+        greenlet.getcurrent() # update ts_current
+        for g in gg:
+            self.assertIsNone(g())
+
+    def test_threaded_adv_leak(self):
+        gg = []
+        def worker():
+            # main and additional *finished* greenlets
+            ll = greenlet.getcurrent().ll = []
+            def additional():
+                ll.append(greenlet.getcurrent())
+            for _ in range(2):
+                greenlet.greenlet(additional).switch()
+            gg.append(weakref.ref(greenlet.getcurrent()))
+        for _ in range(2):
+            t = threading.Thread(target=worker)
+            t.start()
+            t.join(10)
+            del t
+        greenlet.getcurrent() # update ts_current
+        self.__recycle_threads()
+        greenlet.getcurrent() # update ts_current
+        gc.collect()
+        greenlet.getcurrent() # update ts_current
+        for g in gg:
+            self.assertIsNone(g())
+
+    def assertClocksUsed(self):
+        used = greenlet._greenlet.get_clocks_used_doing_optional_cleanup()
+        self.assertGreaterEqual(used, 0)
+        # we don't lose the value
+        greenlet._greenlet.enable_optional_cleanup(True)
+        used2 = greenlet._greenlet.get_clocks_used_doing_optional_cleanup()
+        self.assertEqual(used, used2)
+        self.assertGreater(greenlet._greenlet.CLOCKS_PER_SEC, 1)
+
+    def _check_issue251(self,
+                        manually_collect_background=True,
+                        explicit_reference_to_switch=False):
+        # See https://github.com/python-greenlet/greenlet/issues/251
+        # Killing a greenlet (probably not the main one)
+        # in one thread from another thread would
+        # result in leaking a list (the ts_delkey list).
+        # We no longer use lists to hold that stuff, though.
+
+        # For the test to be valid, even empty lists have to be tracked by the
+        # GC
+
+        assert gc.is_tracked([])
+        HasFinalizerTracksInstances.reset()
+        greenlet.getcurrent()
+        greenlets_before = self.count_objects(greenlet.greenlet, exact_kind=False)
+
+        background_glet_running = threading.Event()
+        background_glet_killed = threading.Event()
+        background_greenlets = []
+
+        # XXX: Switching this to a greenlet subclass that overrides
+        # run results in all callers failing the leaktest; that
+        # greenlet instance is leaked. There's a bound method for
+        # run() living on the stack of the greenlet in g_initialstub,
+        # and since we don't manually switch back to the background
+        # greenlet to let it "fall off the end" and exit the
+        # g_initialstub function, it never gets cleaned up. Making the
+        # garbage collector aware of this bound method (making it an
+        # attribute of the greenlet structure and traversing into it)
+        # doesn't help, for some reason.
+        def background_greenlet():
+            # Throw control back to the main greenlet.
+            jd = HasFinalizerTracksInstances("DELETING STACK OBJECT")
+            greenlet._greenlet.set_thread_local(
+                'test_leaks_key',
+                HasFinalizerTracksInstances("DELETING THREAD STATE"))
+            # Explicitly keeping 'switch' in a local variable
+            # breaks this test in all versions
+            if explicit_reference_to_switch:
+                s = greenlet.getcurrent().parent.switch
+                s([jd])
+            else:
+                greenlet.getcurrent().parent.switch([jd])
+
+        bg_main_wrefs = []
+
+        def background_thread():
+            glet = greenlet.greenlet(background_greenlet)
+            bg_main_wrefs.append(weakref.ref(glet.parent))
+
+            background_greenlets.append(glet)
+            glet.switch() # Be sure it's active.
+            # Control is ours again.
+            del glet # Delete one reference from the thread it runs in.
+            background_glet_running.set()
+            background_glet_killed.wait(10)
+
+            # To trigger the background collection of the dead
+            # greenlet, thus clearing out the contents of the list, we
+            # need to run some APIs. See issue 252.
+            if manually_collect_background:
+                greenlet.getcurrent()
+
+
+        t = threading.Thread(target=background_thread)
+        t.start()
+        background_glet_running.wait(10)
+        greenlet.getcurrent()
+        lists_before = self.count_objects(list, exact_kind=True)
+
+        assert len(background_greenlets) == 1
+        self.assertFalse(background_greenlets[0].dead)
+        # Delete the last reference to the background greenlet
+        # from a different thread. This puts it in the background thread's
+        # ts_delkey list.
+        del background_greenlets[:]
+        background_glet_killed.set()
+
+        # Now wait for the background thread to die.
+        t.join(10)
+        del t
+        # As part of the fix for 252, we need to cycle the ceval.c
+        # interpreter loop to be sure it has had a chance to process
+        # the pending call.
+        self.wait_for_pending_cleanups()
+
+        lists_after = self.count_objects(list, exact_kind=True)
+        greenlets_after = self.count_objects(greenlet.greenlet, exact_kind=False)
+
+        # On 2.7, we observe that lists_after is smaller than
+        # lists_before. No idea what lists got cleaned up. All the
+        # Python 3 versions match exactly.
+        self.assertLessEqual(lists_after, lists_before)
+        # On versions after 3.6, we've successfully cleaned up the
+        # greenlet references thanks to the internal "vectorcall"
+        # protocol; prior to that, there is a reference path through
+        # the ``greenlet.switch`` method still on the stack that we
+        # can't reach to clean up. The C code goes through terrific
+        # lengths to clean that up.
+        if not explicit_reference_to_switch \
+           and greenlet._greenlet.get_clocks_used_doing_optional_cleanup() is not None:
+            # If cleanup was disabled, though, we may not find it.
+            self.assertEqual(greenlets_after, greenlets_before)
+            if manually_collect_background:
+                # TODO: Figure out how to make this work!
+                # The one on the stack is still leaking somehow
+                # in the non-manually-collect state.
+                self.assertEqual(HasFinalizerTracksInstances.EXTANT_INSTANCES, set())
+        else:
+            # The explicit reference prevents us from collecting it
+            # and it isn't always found by the GC either for some
+            # reason. The entire frame is leaked somehow, on some
+            # platforms (e.g., MacPorts builds of Python (all
+            # versions!)), but not on other platforms (the linux and
+            # windows builds on GitHub actions and Appveyor). So we'd
+            # like to write a test that proves that the main greenlet
+            # sticks around, and we can on my machine (macOS 11.6,
+            # MacPorts builds of everything) but we can't write that
+            # same test on other platforms. However, hopefully iteration
+            # done by leakcheck will find it.
+            pass
+
+        if greenlet._greenlet.get_clocks_used_doing_optional_cleanup() is not None:
+            self.assertClocksUsed()
+
+    def test_issue251_killing_cross_thread_leaks_list(self):
+        self._check_issue251()
+
+    def test_issue251_with_cleanup_disabled(self):
+        greenlet._greenlet.enable_optional_cleanup(False)
+        try:
+            self._check_issue251()
+        finally:
+            greenlet._greenlet.enable_optional_cleanup(True)
+
+    @fails_leakcheck_except_on_free_thraded
+    def test_issue251_issue252_need_to_collect_in_background(self):
+        # Between greenlet 1.1.2 and the next version, this was still
+        # failing because the leak of the list still exists when we
+        # don't call a greenlet API before exiting the thread. The
+        # proximate cause is that neither of the two greenlets from
+        # the background thread are actually being destroyed, even
+        # though the GC is in fact visiting both objects. It's not
+        # clear where that leak is? For some reason the thread-local
+        # dict holding it isn't being cleaned up.
+        #
+        # The leak, I think, is in the CPYthon internal function that
+        # calls into green_switch(). The argument tuple is still on
+        # the C stack somewhere and can't be reached? That doesn't
+        # make sense, because the tuple should be collectable when
+        # this object goes away.
+        #
+        # Note that this test sometimes spuriously passes on Linux,
+        # for some reason, but I've never seen it pass on macOS.
+        self._check_issue251(manually_collect_background=False)
+
+    @fails_leakcheck_except_on_free_thraded
+    def test_issue251_issue252_need_to_collect_in_background_cleanup_disabled(self):
+        self.expect_greenlet_leak = True
+        greenlet._greenlet.enable_optional_cleanup(False)
+        try:
+            self._check_issue251(manually_collect_background=False)
+        finally:
+            greenlet._greenlet.enable_optional_cleanup(True)
+
+    @fails_leakcheck_except_on_free_thraded
+    def test_issue251_issue252_explicit_reference_not_collectable(self):
+        self._check_issue251(
+            manually_collect_background=False,
+            explicit_reference_to_switch=True)
+
+    UNTRACK_ATTEMPTS = 100
+
+    def _only_test_some_versions(self):
+        # We're only looking for this problem specifically on 3.11,
+        # and this set of tests is relatively fragile, depending on
+        # OS and memory management details. So we want to run it on 3.11+
+        # (obviously) but not every older 3.x version in order to reduce
+        # false negatives. At the moment, those false results seem to have
+        # resolved, so we are actually running this on 3.8+
+        assert sys.version_info[0] >= 3
+        if RUNNING_ON_MANYLINUX:
+            self.skipTest("Slow and not worth repeating here")
+
+    @ignores_leakcheck
+    # Because we're just trying to track raw memory, not objects, and running
+    # the leakcheck makes an already slow test slower.
+    def test_untracked_memory_doesnt_increase(self):
+        # See https://github.com/gevent/gevent/issues/1924
+        # and https://github.com/python-greenlet/greenlet/issues/328
+        self._only_test_some_versions()
+        def f():
+            return 1
+
+        ITER = 10000
+        def run_it():
+            for _ in range(ITER):
+                greenlet.greenlet(f).switch()
+
+        # Establish baseline
+        for _ in range(3):
+            run_it()
+
+        # uss: (Linux, macOS, Windows): aka "Unique Set Size", this is
+        # the memory which is unique to a process and which would be
+        # freed if the process was terminated right now.
+        uss_before = self.get_process_uss()
+
+        for count in range(self.UNTRACK_ATTEMPTS):
+            uss_before = max(uss_before, self.get_process_uss())
+            run_it()
+
+            uss_after = self.get_process_uss()
+            if uss_after <= uss_before and count > 1:
+                break
+
+        self.assertLessEqual(uss_after, uss_before)
+
+    def _check_untracked_memory_thread(self, deallocate_in_thread=True):
+        self._only_test_some_versions()
+        # Like the above test, but what if there are a bunch of
+        # unfinished greenlets in a thread that dies?
+        # Does it matter if we deallocate in the thread or not?
+
+        # First, make sure we can get useful measurements. This will
+        # be skipped if not.
+        self.get_process_uss()
+
+        EXIT_COUNT = [0]
+
+        def f():
+            try:
+                greenlet.getcurrent().parent.switch()
+            except greenlet.GreenletExit:
+                EXIT_COUNT[0] += 1
+                raise
+            return 1
+
+        ITER = 10000
+        def run_it():
+            glets = []
+            for _ in range(ITER):
+                # Greenlet starts, switches back to us.
+                # We keep a strong reference to the greenlet though so it doesn't
+                # get a GreenletExit exception.
+                g = greenlet.greenlet(f)
+                glets.append(g)
+                g.switch()
+
+            return glets
+
+        test = self
+
+        class ThreadFunc:
+            uss_before = uss_after = 0
+            glets = ()
+            ITER = 2
+            def __call__(self):
+                self.uss_before = test.get_process_uss()
+
+                for _ in range(self.ITER):
+                    self.glets += tuple(run_it())
+
+                for g in self.glets:
+                    test.assertIn('suspended active', str(g))
+                # Drop them.
+                if deallocate_in_thread:
+                    self.glets = ()
+                self.uss_after = test.get_process_uss()
+
+        # Establish baseline
+        uss_before = uss_after = None
+        for count in range(self.UNTRACK_ATTEMPTS):
+            EXIT_COUNT[0] = 0
+            thread_func = ThreadFunc()
+            t = threading.Thread(target=thread_func)
+            t.start()
+            t.join(30)
+            self.assertFalse(t.is_alive())
+
+            if uss_before is None:
+                uss_before = thread_func.uss_before
+
+            uss_before = max(uss_before, thread_func.uss_before)
+            if deallocate_in_thread:
+                self.assertEqual(thread_func.glets, ())
+                self.assertEqual(EXIT_COUNT[0], ITER * thread_func.ITER)
+
+            del thread_func # Deallocate the greenlets; but this won't raise into them
+            del t
+            if not deallocate_in_thread:
+                self.assertEqual(EXIT_COUNT[0], 0)
+            if deallocate_in_thread:
+                self.wait_for_pending_cleanups()
+
+            uss_after = self.get_process_uss()
+            # See if we achieve a non-growth state at some point. Break when we do.
+            if uss_after <= uss_before and count > 1:
+                break
+
+        self.wait_for_pending_cleanups()
+        uss_after = self.get_process_uss()
+        # On Windows, USS can fluctuate by tens of KB between measurements
+        # due to working set trimming, page table updates, etc. Allow a
+        # small tolerance so OS-level noise doesn't cause false failures.
+        # Real leaks produce MBs of growth (each iteration creates 20k
+        # greenlets; the greenlet Python object alone, not counting the C++
+        # state it points to, is 56 bytes so 20k greenlets alone is
+        # slightly over a megabyte; the C++ UserGreenlet clocks in at 208
+        # bytes, which adds another 4MB), so 512 KB is well below the detection
+        # threshold for genuine issues.
+        #
+        # 2025-06-16: We have now observed the same problem on both
+        # ubuntu-latest 3.14t (2948K "leak") and macos-latest 3.15t (192K
+        # "leak"). This was after the merge of PR #511, but that change
+        # should be unrelated. In particular, the ubuntu failure was after
+        # 30 loops and the macos failure was after only 3 --- both much
+        # smaller than UNTRACK_ATTEMPTS=100! We break out of the loop if we
+        # see a USS at the end of the loop that's smaller than at the start
+        # of the loop, and then we wait for pending cleanups and get the
+        # USS again. So this means that we think we have success, break out
+        # of the loop, and the pending cleanups cause our USS to grow -
+        # perhaps by touching pages that had been paged out? An immediately
+        # prior attempt of both platforms had been successful.
+        #
+        # At any rate, we need a larger tolerance than 512K, and we
+        # need it on all platforms, not just Windows.
+        tolerance = 3 * 1024 * 1024
+        self.assertLessEqual(uss_after, uss_before + tolerance,
+                             "after attempts %d" % (count,))
+
+    @ignores_leakcheck
+    # Because we're just trying to track raw memory, not objects, and running
+    # the leakcheck makes an already slow test slower.
+    def test_untracked_memory_doesnt_increase_unfinished_thread_dealloc_in_thread(self):
+        self._check_untracked_memory_thread(deallocate_in_thread=True)
+
+    @ignores_leakcheck
+    # Because the main greenlets from the background threads do not exit in a timely fashion,
+    # we fail the object-based leakchecks.
+    def test_untracked_memory_doesnt_increase_unfinished_thread_dealloc_in_main(self):
+        # Between Feb 10 and Feb 20 2026, this test started failing on
+        # Github Actions, windows 3.14t. With no relevant code changes on
+        # our part. Both versions were 3.14.3 (same build). The only change
+        # is the Github actions "Runner Image". The working one was version
+        # 20260202.17.1, while the updated failing version was
+        # 20260217.31.1. Both report the same version of the operating system
+        # (Microsoft Windows Server 2025 10.0.26100).
+        #
+        # Reevaluate on future runner image releases.
+        if WIN and RUNNING_ON_FREETHREAD_BUILD and PY314:
+            self.skipTest("Windows 3.14t appears to leak. No other platform does.")
+        self._check_untracked_memory_thread(deallocate_in_thread=False)
+
+if __name__ == '__main__':
+    __import__('unittest').main()

venv/Lib/site-packages/greenlet/tests/test_stack_saved.py

@@ -0,0 +1,19 @@
+import greenlet
+from . import TestCase
+
+
+class Test(TestCase):
+
+    def test_stack_saved(self):
+        main = greenlet.getcurrent()
+        self.assertEqual(main._stack_saved, 0)
+
+        def func():
+            main.switch(main._stack_saved)
+
+        g = greenlet.greenlet(func)
+        x = g.switch()
+        self.assertGreater(x, 0)
+        self.assertGreater(g._stack_saved, 0)
+        g.switch()
+        self.assertEqual(g._stack_saved, 0)

venv/Lib/site-packages/greenlet/tests/test_throw.py

@@ -0,0 +1,128 @@
+import sys
+
+
+from greenlet import greenlet
+from . import TestCase
+
+def switch(*args):
+    return greenlet.getcurrent().parent.switch(*args)
+
+
+class ThrowTests(TestCase):
+    def test_class(self):
+        def f():
+            try:
+                switch("ok")
+            except RuntimeError:
+                switch("ok")
+                return
+            switch("fail")
+        g = greenlet(f)
+        res = g.switch()
+        self.assertEqual(res, "ok")
+        res = g.throw(RuntimeError)
+        self.assertEqual(res, "ok")
+
+    def test_val(self):
+        def f():
+            try:
+                switch("ok")
+            except RuntimeError:
+                val = sys.exc_info()[1]
+                if str(val) == "ciao":
+                    switch("ok")
+                    return
+            switch("fail")
+
+        g = greenlet(f)
+        res = g.switch()
+        self.assertEqual(res, "ok")
+        res = g.throw(RuntimeError("ciao"))
+        self.assertEqual(res, "ok")
+
+        g = greenlet(f)
+        res = g.switch()
+        self.assertEqual(res, "ok")
+        res = g.throw(RuntimeError, "ciao")
+        self.assertEqual(res, "ok")
+
+    def test_kill(self):
+        def f():
+            switch("ok")
+            switch("fail")
+        g = greenlet(f)
+        res = g.switch()
+        self.assertEqual(res, "ok")
+        res = g.throw()
+        self.assertTrue(isinstance(res, greenlet.GreenletExit))
+        self.assertTrue(g.dead)
+        res = g.throw()    # immediately eaten by the already-dead greenlet
+        self.assertTrue(isinstance(res, greenlet.GreenletExit))
+
+    def test_throw_goes_to_original_parent(self):
+        main = greenlet.getcurrent()
+
+        def f1():
+            try:
+                main.switch("f1 ready to catch")
+            except IndexError:
+                return "caught"
+            return "normal exit"
+
+        def f2():
+            main.switch("from f2")
+
+        g1 = greenlet(f1)
+        g2 = greenlet(f2, parent=g1)
+        with self.assertRaises(IndexError):
+            g2.throw(IndexError)
+        self.assertTrue(g2.dead)
+        self.assertTrue(g1.dead)
+
+        g1 = greenlet(f1)
+        g2 = greenlet(f2, parent=g1)
+        res = g1.switch()
+        self.assertEqual(res, "f1 ready to catch")
+        res = g2.throw(IndexError)
+        self.assertEqual(res, "caught")
+        self.assertTrue(g2.dead)
+        self.assertTrue(g1.dead)
+
+        g1 = greenlet(f1)
+        g2 = greenlet(f2, parent=g1)
+        res = g1.switch()
+        self.assertEqual(res, "f1 ready to catch")
+        res = g2.switch()
+        self.assertEqual(res, "from f2")
+        res = g2.throw(IndexError)
+        self.assertEqual(res, "caught")
+        self.assertTrue(g2.dead)
+        self.assertTrue(g1.dead)
+
+    def test_non_traceback_param(self):
+        with self.assertRaises(TypeError) as exc:
+            greenlet.getcurrent().throw(
+                Exception,
+                Exception(),
+                self
+            )
+        self.assertEqual(str(exc.exception),
+                         "throw() third argument must be a traceback object")
+
+    def test_instance_of_wrong_type(self):
+        with self.assertRaises(TypeError) as exc:
+            greenlet.getcurrent().throw(
+                Exception(),
+                BaseException()
+            )
+
+        self.assertEqual(str(exc.exception),
+                         "instance exception may not have a separate value")
+
+    def test_not_throwable(self):
+        with self.assertRaises(TypeError) as exc:
+            greenlet.getcurrent().throw(
+                "abc"
+            )
+        self.assertEqual(str(exc.exception),
+                         "exceptions must be classes, or instances, not str")