(self)
| 165 | file.unlink() |
| 166 | |
| 167 | def test_pre_fork_compile(self): |
| 168 | code = """if 1: |
| 169 | import sys |
| 170 | import os |
| 171 | import sysconfig |
| 172 | from _testinternalcapi import ( |
| 173 | compile_perf_trampoline_entry, |
| 174 | perf_trampoline_set_persist_after_fork, |
| 175 | ) |
| 176 | |
| 177 | def foo_fork(): |
| 178 | pass |
| 179 | |
| 180 | def bar_fork(): |
| 181 | foo_fork() |
| 182 | |
| 183 | def foo(): |
| 184 | import time; time.sleep(1) |
| 185 | |
| 186 | def bar(): |
| 187 | foo() |
| 188 | |
| 189 | def compile_trampolines_for_all_functions(): |
| 190 | perf_trampoline_set_persist_after_fork(1) |
| 191 | for _, obj in globals().items(): |
| 192 | if callable(obj) and hasattr(obj, '__code__'): |
| 193 | compile_perf_trampoline_entry(obj.__code__) |
| 194 | |
| 195 | if __name__ == "__main__": |
| 196 | compile_trampolines_for_all_functions() |
| 197 | pid = os.fork() |
| 198 | if pid == 0: |
| 199 | print(os.getpid()) |
| 200 | bar_fork() |
| 201 | else: |
| 202 | bar() |
| 203 | """ |
| 204 | |
| 205 | with temp_dir() as script_dir: |
| 206 | script = make_script(script_dir, "perftest", code) |
| 207 | env = {**os.environ, "PYTHON_JIT": "0"} |
| 208 | with subprocess.Popen( |
| 209 | [sys.executable, "-Xperf", script], |
| 210 | universal_newlines=True, |
| 211 | stderr=subprocess.PIPE, |
| 212 | stdout=subprocess.PIPE, |
| 213 | env=env, |
| 214 | ) as process: |
| 215 | stdout, stderr = process.communicate() |
| 216 | |
| 217 | self.assertEqual(process.returncode, 0) |
| 218 | self.assertNotIn("Error:", stderr) |
| 219 | child_pid = int(stdout.strip()) |
| 220 | perf_file = pathlib.Path(f"/tmp/perf-{process.pid}.map") |
| 221 | perf_child_file = pathlib.Path(f"/tmp/perf-{child_pid}.map") |
| 222 | self.assertTrue(perf_file.exists()) |
| 223 | self.assertTrue(perf_child_file.exists()) |
| 224 |
nothing calls this directly
no test coverage detected