@@ -167,6 +167,48 @@ def baz():
167167 self .assertNotIn (f"py::bar:{ script } " , child_perf_file_contents )
168168 self .assertNotIn (f"py::baz:{ script } " , child_perf_file_contents )
169169
170+ @unittest .skipIf (support .check_bolt_optimized (), "fails on BOLT instrumented binaries" )
171+ def test_trampoline_works_after_fork_with_many_code_objects (self ):
172+ code = """if 1:
173+ import gc, os, sys, signal
174+
175+ # Create many code objects so trampoline_refcount > 1
176+ for i in range(50):
177+ exec(compile(f"def _dummy_{i}(): pass", f"<test{i}>", "exec"))
178+
179+ pid = os.fork()
180+ if pid == 0:
181+ # Child: create and destroy new code objects,
182+ # then collect garbage. If the old code watcher
183+ # survived the fork, the double-decrement of
184+ # trampoline_refcount will cause a SIGSEGV.
185+ for i in range(50):
186+ exec(compile(f"def _child_{i}(): pass", f"<child{i}>", "exec"))
187+ gc.collect()
188+ os._exit(0)
189+ else:
190+ _, status = os.waitpid(pid, 0)
191+ if os.WIFSIGNALED(status):
192+ print(f"FAIL: child killed by signal {os.WTERMSIG(status)}", file=sys.stderr)
193+ sys.exit(1)
194+ sys.exit(os.WEXITSTATUS(status))
195+ """
196+ with temp_dir () as script_dir :
197+ script = make_script (script_dir , "perftest" , code )
198+ env = {** os .environ , "PYTHON_JIT" : "0" }
199+ with subprocess .Popen (
200+ [sys .executable , "-Xperf" , script ],
201+ text = True ,
202+ stderr = subprocess .PIPE ,
203+ stdout = subprocess .PIPE ,
204+ env = env ,
205+ ) as process :
206+ stdout , stderr = process .communicate ()
207+
208+ self .assertEqual (process .returncode , 0 , stderr )
209+ self .assertEqual (stderr , "" )
210+
211+ @unittest .skipIf (support .check_bolt_optimized (), "fails on BOLT instrumented binaries" )
170212 def test_sys_api (self ):
171213 code = """if 1:
172214 import sys
0 commit comments