https://github.com/python/cpython/commit/a064b323f4350305e7486c7b1090cf12b19e7738
commit: a064b323f4350305e7486c7b1090cf12b19e7738
branch: main
author: Victor Stinner <[email protected]>
committer: vstinner <[email protected]>
date: 2026-06-17T10:58:22+02:00
summary:
gh-151496: Use process groups in test_dtrace (#151512)
Create a new process group to run bpftrace commands, so it's possible
to kill also child processes on timeout.
files:
M Lib/test/test_dtrace.py
diff --git a/Lib/test/test_dtrace.py b/Lib/test/test_dtrace.py
index 6286b6d21b572e..592f59d77f9221 100644
--- a/Lib/test/test_dtrace.py
+++ b/Lib/test/test_dtrace.py
@@ -1,6 +1,7 @@
import dis
import os.path
import re
+import signal
import subprocess
import sys
import sysconfig
@@ -50,6 +51,24 @@ def normalize_trace_output(output):
)
+USE_PROCESS_GROUP = (hasattr(os, "setsid") and hasattr(os, "killpg"))
+
+def create_process_group(*args, **kwargs):
+ if USE_PROCESS_GROUP:
+ kwargs['start_new_session'] = True
+ return subprocess.Popen(*args, **kwargs)
+
+def kill_process_group(proc):
+ if USE_PROCESS_GROUP:
+ try:
+ os.killpg(proc.pid, signal.SIGKILL)
+ except ProcessLookupError:
+ pass
+ else:
+ proc.kill()
+ proc.communicate() # Clean up
+
+
class TraceBackend:
EXTENSION = None
COMMAND = None
@@ -205,7 +224,7 @@ def run_case(self, name, optimize_python=None):
program = self.PROGRAMS[name].format(python=sys.executable)
try:
- proc = subprocess.Popen(
+ proc = create_process_group(
["bpftrace", "-e", program, "-c", " ".join(subcommand)],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
@@ -213,7 +232,7 @@ def run_case(self, name, optimize_python=None):
)
stdout, stderr = proc.communicate(timeout=60)
except subprocess.TimeoutExpired:
- proc.kill()
+ kill_process_group(proc)
raise AssertionError("bpftrace timed out")
except (FileNotFoundError, PermissionError) as e:
raise unittest.SkipTest(f"bpftrace not available: {e}")
@@ -243,7 +262,7 @@ def assert_usable(self):
# Check if bpftrace is available and can attach to USDT probes
program = f'usdt:{sys.executable}:python:function__entry {{
printf("probe: success\\n"); exit(); }}'
try:
- proc = subprocess.Popen(
+ proc = create_process_group(
["bpftrace", "-e", program, "-c", f"{sys.executable} -c pass"],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
@@ -251,8 +270,7 @@ def assert_usable(self):
)
stdout, stderr = proc.communicate(timeout=10)
except subprocess.TimeoutExpired:
- proc.kill()
- proc.communicate() # Clean up
+ kill_process_group(proc)
raise unittest.SkipTest("bpftrace timed out during usability
check")
except OSError as e:
raise unittest.SkipTest(f"bpftrace not available: {e}")
_______________________________________________
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]
