o i@sXdZddlZddlZddlZddlZddlmZGdddeZedkr*e dSdS)u Tests for greenlet behavior during interpreter shutdown (Py_FinalizeEx). During interpreter shutdown, several greenlet code paths can access partially-destroyed Python state, leading to SIGSEGV. Two independent guards protect against this on ALL Python versions: 1. g_greenlet_shutting_down — set by an atexit handler registered at greenlet import time (LIFO = runs before other cleanup). Covers the atexit phase of Py_FinalizeEx, where _Py_IsFinalizing() is still False on all Python versions. 2. Py_IsFinalizing() — covers the GC collection and later phases of Py_FinalizeEx, where __del__ methods and destructor code run. 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. N)TestCasec@seZdZddZddZddZddZd d Zd d Zd dZ ddZ ddZ ddZ ddZ ddZddZddZddZdd Zd!d"Zd#d$Zd%d&Zd'd(Zd)d*Zd+d,Zd-S).TestInterpreterShutdowncCs6t|}tjtjd|gddddd}|j|j|jfS)z Run a Python script in a subprocess that exercises greenlet during interpreter shutdown. Returns (returncode, stdout, stderr). z-cTF)capture_outputtexttimeoutcheck) textwrapdedent subprocessrunsys executable returncodestdoutstderr)self script_body full_scriptresultrh/var/www/html/arapca_proje/venv/lib/python3.10/site-packages/greenlet/tests/test_interpreter_shutdown.py_run_shutdown_script)s  z,TestInterpreterShutdown._run_shutdown_scriptc C>|d\}}}||dd|d|||d|dS)a8 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. aT 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") rProcess crashed (rc=): z OK: exiting with active greenletNr assertEqualassertInrrcrrrrr)test_active_greenlet_at_shutdown_no_crash<s  zATestInterpreterShutdown.test_active_greenlet_at_shutdown_no_crashc Cr)zm Multiple suspended greenlets at shutdown should all be cleaned up without crashing. a 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") rrrz#OK: 10 active greenlets at shutdownNrrrrr*test_multiple_active_greenlets_at_shutdownTszBTestInterpreterShutdown.test_multiple_active_greenlets_at_shutdownc Cr)zQ Nested (chained parent) greenlets at shutdown should not crash. a 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") rrrz OK: nested greenlets at shutdownNrrrrr!test_nested_greenlets_at_shutdownksz9TestInterpreterShutdown.test_nested_greenlets_at_shutdownc Cr)zm Greenlets in worker threads that are still referenced at shutdown should not crash. a 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") rrrz$OK: 3 threaded greenlets at shutdownNrrrrr#test_threaded_greenlets_at_shutdownsz;TestInterpreterShutdown.test_threaded_greenlets_at_shutdownc CJ|d\}}}||dd|d|||d||d|dS)z 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). a4 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") rrrOK: thread cleanup donezCLEANUP: GreenletExit caughtNrrrrr(test_greenlet_cleanup_during_thread_exits z@TestInterpreterShutdown.test_greenlet_cleanup_during_thread_exitc Cr%)zl try/finally blocks in active greenlets run correctly when the owning thread exits. aR 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") rrrr&zFINALLY: cleanup executedNrrrrr%test_finally_block_during_thread_exits z=TestInterpreterShutdown.test_finally_block_during_thread_exitc Cr)z Stress test: many active greenlets with cleanup code at shutdown. Ensures no crashes regardless of deallocation order. a 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. rrrz#OK: 50 greenlets about to shut downNrrrrr,test_many_greenlets_with_cleanup_at_shutdownszDTestInterpreterShutdown.test_many_greenlets_with_cleanup_at_shutdownc Cr)z Deeply nested greenlet parent chains at shutdown. Tests that the deallocation order doesn't cause issues. a 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}") rrrzOK: nested to depth 10Nrrrrr(test_deeply_nested_greenlets_at_shutdowns z@TestInterpreterShutdown.test_deeply_nested_greenlets_at_shutdownc Cr)z A greenlet that has an active exception context when it's suspended should not crash during shutdown cleanup. a 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") rrrz.OK: greenlet with active exception at shutdownNrrrrr(test_greenlet_with_traceback_at_shutdown&sz@TestInterpreterShutdown.test_greenlet_with_traceback_at_shutdownc CL|d\}}}||dd|d|||d||d|ddS) z 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. a 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") rrrOK: atexit registered&OK: getcurrent returned valid greenletigetcurrent() should return a valid greenlet when called before greenlet's cleanup handler (LIFO ordering)Nrrrrr&test_getcurrent_during_atexit_no_crashJs  z>TestInterpreterShutdown.test_getcurrent_during_atexit_no_crashc Cr)zK Calling greenlet.gettrace() during atexit must not crash. a 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") rrrOK: registeredNrrrrr$test_gettrace_during_atexit_no_crashiztest_greenlet_construction_with_active_greenlets_during_atexitszVTestInterpreterShutdown.test_greenlet_construction_with_active_greenlets_during_atexitc Cr)u 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. a 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") rrrz.OK: cross-thread setup done, atexit registeredNrrrrrCtest_greenlet_construction_with_cross_thread_deleteme_during_atexits #z[TestInterpreterShutdown.test_greenlet_construction_with_cross_thread_deleteme_during_atexitc Cr,) u greenlet.getcurrent() must return None 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). ax import gc import os import greenlet gc.disable() class CleanupChecker: def __del__(self): try: 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") # 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") rrrzOK: deferred cycle createdGUARDED: getcurrent=Nonetgetcurrent() must return None during GC finalization; returned a live object instead (missing Py_IsFinalizing guard)Nrrrrr3test_getcurrent_returns_none_during_gc_finalization:s   zKTestInterpreterShutdown.test_getcurrent_returns_none_during_gc_finalizationc Cr,) z Same as above but with active greenlets at shutdown, which increases the amount of C++ destructor work during finalization. a import gc import os import greenlet gc.disable() class CleanupChecker: def __del__(self): try: 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") # 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") rrrz'OK: 10 active greenlets, cycle deferredr:r;NrrrrrItest_getcurrent_returns_none_during_gc_finalization_with_active_greenletses $ zaTestInterpreterShutdown.test_getcurrent_returns_none_during_gc_finalization_with_active_greenletsc Cr,) a0 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. a{ import gc import os import threading import greenlet gc.disable() class CleanupChecker: def __del__(self): try: 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") # 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") rrrz)OK: cross-thread cleanup + cycle deferredr:r;Nrrrrr@test_getcurrent_returns_none_during_gc_finalization_cross_threads * zXTestInterpreterShutdown.test_getcurrent_returns_none_during_gc_finalization_cross_threadc Cr,) a greenlet.getcurrent() must return None when called from an atexit handler that runs AFTER greenlet's own atexit handler. This tests the g_greenlet_shutting_down flag, which is needed because _Py_IsFinalizing() is still False during the atexit phase on ALL Python versions. Without g_greenlet_shutting_down, getcurrent() proceeds unguarded into partially-torn-down state. u 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") rrrz,OK: atexit registered before greenlet importr:getcurrent() must return None during atexit phase; returned a live object instead (missing g_greenlet_shutting_down atexit handler)Nrrrrr0test_getcurrent_returns_none_during_atexit_phases   zHTestInterpreterShutdown.test_getcurrent_returns_none_during_atexit_phasec Cr,) z Same as above but with active greenlets, ensuring the atexit guard works even when there is greenlet state to clean up. a 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") rrrr7r:r?NrrrrrFtest_getcurrent_returns_none_during_atexit_phase_with_active_greenlets s  z^TestInterpreterShutdown.test_getcurrent_returns_none_during_atexit_phase_with_active_greenletsN)__name__ __module__ __qualname__rr!r"r#r$r'r(r)r*r+r0r2r4r5r6r8r9r<r=r>r@rArrrrr's.+% $(#@+/M )r__main__) __doc__r r unittestr greenlet.testsrrrBmainrrrrs