https://github.com/python/cpython/commit/6fd15f0ea315af9883207518f692b1d3f3503350
commit: 6fd15f0ea315af9883207518f692b1d3f3503350
branch: main
author: Victor Stinner <[email protected]>
committer: vstinner <[email protected]>
date: 2026-10-21T12:56:35-05:00
summary:
gh-159534: Fix race condition in test_sampling_profiler (#160028)
Fix test_is_python_process_non_python_subprocess(). Wait until the
forked Python process calls exec() before calling is_python_process().
Previously, is_python_build() returned True if exec() was not called
yet in the forked process.
files:
M Lib/test/test_profiling/test_sampling_profiler/test_children.py
diff --git a/Lib/test/test_profiling/test_sampling_profiler/test_children.py
b/Lib/test/test_profiling/test_sampling_profiler/test_children.py
index e64d917eedde56b..4b27c6dbb1896c1 100644
--- a/Lib/test/test_profiling/test_sampling_profiler/test_children.py
+++ b/Lib/test/test_profiling/test_sampling_profiler/test_children.py
@@ -16,6 +16,7 @@
SHORT_TIMEOUT,
reap_children,
requires_remote_subprocess_debugging,
+ sleeping_retry,
)
# Guard imports that require _remote_debugging module.
@@ -84,29 +85,13 @@ def reader():
return result[0]
-def _wait_for_process_ready(proc, timeout):
- # Wait for a subprocess to be ready using polling instead of fixed sleep.
- # Returns True if process is ready, False if it exited or timeout.
- deadline = time.time() + timeout
- poll_interval = 0.01
-
- while time.time() < deadline:
- if proc.poll() is not None:
- return False
-
- try:
- if sys.platform == "linux":
- if os.path.exists(f"/proc/{proc.pid}/exe"):
- return True
- else:
- return True
- except OSError:
- pass
-
- time.sleep(poll_interval)
- poll_interval = min(poll_interval * 2, 0.1)
-
- return proc.poll() is None
+def _wait_for_process_exec(proc, name, timeout):
+ # Wait until a child process calls exec() to replace the forked process
+ # executable. Raise AssertionError on timeout.
+ for _ in sleeping_retry(timeout, max_delay=0.1):
+ exe = os.readlink(f"/proc/{proc.pid}/exe")
+ if os.path.basename(exe) == name:
+ break
@unittest.skipIf(
@@ -634,17 +619,13 @@ def test_is_python_process_non_python_subprocess(self):
"""Test that a non-Python subprocess is not detected as Python."""
# Start a non-Python subprocess (sleep command)
proc = subprocess.Popen(
- ["sleep", "10"],
+ ["sleep", "60"],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
try:
- # Wait for process to be ready using polling
- self.assertTrue(
- _wait_for_process_ready(proc, SHORT_TIMEOUT),
- f"Process {proc.pid} should be ready within {SHORT_TIMEOUT}s",
- )
+ _wait_for_process_exec(proc, "sleep", SHORT_TIMEOUT)
self.assertFalse(
is_python_process(proc.pid),
_______________________________________________
Python-checkins mailing list -- [email protected]
To unsubscribe send an email to [email protected]
https://mail.python.org/mailman3//lists/python-checkins.python.org
Member address: [email protected]
