Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
39 changes: 10 additions & 29 deletions Lib/test/test_profiling/test_sampling_profiler/test_children.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
SHORT_TIMEOUT,
reap_children,
requires_remote_subprocess_debugging,
sleeping_retry,
)

# Guard imports that require _remote_debugging module.
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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),
Expand Down
Loading