# Free-threaded build, fresh process: ONE first `import tempfile` (which imports # shutil, which imports _zstd; _zstd's module exec calls PyType_Freeze()) races # @dataclass class creation (dataclasses sets __eq__/__lt__/... on the class: # pinned TYPE_LOCK + stop-the-world) and automatic GC (stop-the-world). # # usage: python single_shot_tempfile.py [module] (default: tempfile) # e.g. `compression.zstd`, or `json` as a control. # Prints "ok" on success. It hangs only occasionally, so run it in a loop # with a timeout of a few seconds; a hang is reported by faulthandler after # 30 s ("Timeout (0:00:30)!", exit status 1). import faulthandler, importlib, sys # Exit status: 0 ok, 1 hang (faulthandler timeout), 2 GIL enabled, # 3 a worker raised or did not finish its work. # The bug exists only on a free-threaded build with the GIL disabled. print(f"sys._is_gil_enabled() = {sys._is_gil_enabled()}", flush=True) if sys._is_gil_enabled() is not False: print("FAIL: needs a free-threaded build with the GIL disabled " "(do not set PYTHON_GIL=1 or -X gil=1)", file=sys.stderr) sys.exit(2) faulthandler.dump_traceback_later(30, exit=True) import threading, dataclasses, time MODULE = sys.argv[1] if len(sys.argv) > 1 else "tempfile" done = False imported = [0] # Any unexpected exception in a worker fails the run (exit status 3) instead # of letting the other threads finish and print "ok". errors = [] def guarded(fn): def run(): try: fn() except BaseException as e: errors.append(f"{fn.__name__}: {e!r}") raise # threading.excepthook prints the traceback run.__name__ = fn.__name__ return run def check_errors(completed_ok): if errors or not completed_ok: faulthandler.cancel_dump_traceback_later() print("FAIL: " + ("; ".join(errors) or "importer did not finish its work"), file=sys.stderr) sys.exit(3) go = threading.Event() def importer(): global done go.wait(); time.sleep(0.02) try: importlib.import_module(MODULE) imported[0] += 1 finally: done = True def dc_maker(): go.wait() while not done: @dataclasses.dataclass(eq=True, order=True) class D: x: int = 0 def garbage(): go.wait() while not done: l = [] for i in range(2000): a = []; a.append(a); l.append(a) ts = [threading.Thread(target=guarded(t)) for t in (importer, dc_maker, garbage)] for t in ts: t.start() go.set() for t in ts: t.join() check_errors(imported[0] == 1) faulthandler.cancel_dump_traceback_later() print("ok")