https://github.com/python/cpython/commit/20c07473f494cbd184ab43ac1ba9fc5ef7774f6c
commit: 20c07473f494cbd184ab43ac1ba9fc5ef7774f6c
branch: 3.13
author: Serhiy Storchaka <[email protected]>
committer: Yhg1s <[email protected]>
date: 2026-09-29T23:51:11+02:00
summary:
[3.13] gh-75876: Run bigmem tests in a subprocess (GH-155302) (#156014)
A test which really allocates the memory it asks for (that is, run with -M)
now runs in a subprocess, so that the memory it uses and the address space
it fragments are released when it ends. A dummy run stays in the process.
The separate watchdog process is no longer used. Unlike in the main branch,
the parent process does not report the memory usage of the subprocess:
get_process_memory_usage() (gh-150114) is not available on this branch.
(cherry picked from commit 66d7c891044915ccb7a12e0cc2c78991508e816e)
files:
A Misc/NEWS.d/next/Tests/2026-08-18-10-16-44.gh-issue-75876.PEZplM.rst
D Lib/test/memory_watchdog.py
M Lib/test/_isolated_sample.py
M Lib/test/support/__init__.py
M Lib/test/support/isolation.py
M Lib/test/test_support.py
diff --git a/Lib/test/_isolated_sample.py b/Lib/test/_isolated_sample.py
index 5853b654fc28cb6..3aa58835c883677 100644
--- a/Lib/test/_isolated_sample.py
+++ b/Lib/test/_isolated_sample.py
@@ -10,6 +10,7 @@
import sys
import time
import unittest
+from test import support
from test.support import isolation
# DurationSample sleeps this long in the subprocess; a parent-reported duration
@@ -178,3 +179,12 @@ class TimeoutSample(unittest.TestCase):
@isolation.runInSubprocess(timeout=TIMEOUT)
def test_hang(self):
time.sleep(TIMEOUT_HANG)
+
+
+class BigmemSample(unittest.TestCase):
+
+ @support.bigmemtest(size=1024, memuse=1)
+ def test_where_it_runs(self, size):
+ # A real run is isolated by bigmemtest() itself, a dummy run is not.
+ self.assertEqual(isolation.runningInSubprocess,
+ bool(support.real_max_memuse))
diff --git a/Lib/test/memory_watchdog.py b/Lib/test/memory_watchdog.py
deleted file mode 100644
index fee062ecc9b3001..000000000000000
--- a/Lib/test/memory_watchdog.py
+++ /dev/null
@@ -1,21 +0,0 @@
-"""Memory watchdog: periodically read the memory usage of the main test process
-and print it out, until terminated."""
-# stdin should refer to the process' /proc/<PID>/statm: we don't pass the
-# process' PID to avoid a race condition in case of - unlikely - PID recycling.
-# If the process crashes, reading from the /proc entry will fail with ESRCH.
-
-
-import sys
-import time
-from test.support import get_pagesize
-
-
-while True:
- page_size = get_pagesize()
- sys.stdin.seek(0)
- statm = sys.stdin.read()
- data = int(statm.split()[5])
- sys.stdout.write(" ... process data size: {data:.1f}G\n"
- .format(data=data * page_size / (1024 ** 3)))
- sys.stdout.flush()
- time.sleep(1)
diff --git a/Lib/test/support/__init__.py b/Lib/test/support/__init__.py
index b82c9355b484a64..abc525b80e0abf3 100644
--- a/Lib/test/support/__init__.py
+++ b/Lib/test/support/__init__.py
@@ -1149,38 +1149,6 @@ def set_memlimit(limit: str) -> None:
max_memuse = memlimit
-class _MemoryWatchdog:
- """An object which periodically watches the process' memory consumption
- and prints it out.
- """
-
- def __init__(self):
- self.procfile = '/proc/{pid}/statm'.format(pid=os.getpid())
- self.started = False
-
- def start(self):
- import warnings
- try:
- f = open(self.procfile, 'r')
- except OSError as e:
- logging.getLogger(__name__).warning('/proc not available for
stats: %s', e, exc_info=e)
- sys.stderr.flush()
- return
-
- import subprocess
- with f:
- watchdog_script = findfile("memory_watchdog.py")
- self.mem_watchdog = subprocess.Popen([sys.executable,
watchdog_script],
- stdin=f,
- stderr=subprocess.DEVNULL)
- self.started = True
-
- def stop(self):
- if self.started:
- self.mem_watchdog.terminate()
- self.mem_watchdog.wait()
-
-
def bigmemtest(size, memuse, dry_run=True):
"""Decorator for bigmem tests.
@@ -1193,8 +1161,14 @@ def bigmemtest(size, memuse, dry_run=True):
extra argument. If 'dry_run' is true, the value passed to the test method
may be less than the requested value. If 'dry_run' is false, it means the
test doesn't support dummy runs when -M is not specified.
+
+ A test that actually allocates the requested memory (that is, one run with
+ -M) runs in a subprocess, so that the memory it uses and the address space
+ it fragments are released when it ends. A dummy run stays in the process.
"""
def decorator(f):
+ from test.support import isolation
+
@functools.wraps(f)
def wrapper(self):
size = wrapper.size
@@ -1210,20 +1184,21 @@ def wrapper(self):
"not enough memory: %.1fG minimum needed"
% (size * memuse / (1024 ** 3)))
- if real_max_memuse and verbose:
+ if (real_max_memuse and verbose
+ and not isolation.runningInSubprocess):
print()
- print(" ... expected peak memory use: {peak:.1f}G"
- .format(peak=size * memuse / (1024 ** 3)))
- watchdog = _MemoryWatchdog()
- watchdog.start()
- else:
- watchdog = None
+ peak = (size * memuse) / (1024 ** 3)
+ print(f" ... expected peak memory use: {peak:.1f} GiB")
+
+ if (real_max_memuse and has_subprocess_support
+ and not isolation.runningInSubprocess):
+ cls = type(self)
+ qualname = f'{cls.__qualname__}.{f.__name__}'
+ proc = isolation._start_test(cls.__module__, qualname)
+ isolation._replay_test(self, *proc.wait())
+ return
- try:
- return f(self, maxsize)
- finally:
- if watchdog:
- watchdog.stop()
+ return f(self, maxsize)
wrapper.size = size
wrapper.memuse = memuse
diff --git a/Lib/test/support/isolation.py b/Lib/test/support/isolation.py
index bb4fa6b003cc20c..3cfd406b0f2f0e1 100644
--- a/Lib/test/support/isolation.py
+++ b/Lib/test/support/isolation.py
@@ -108,12 +108,75 @@ def _child_environ(env):
return environ
-def _run_in_subprocess(module, qualname, options, env, timeout):
- """Run module.qualname (a test method or class) in a fresh subprocess.
+class _SubprocessTest:
+ """A test running in a subprocess, started by _start_test().
- Return ``(payload, output, returncode)``, where *payload* is the decoded
- ``{'outcomes': ..., 'durations': ...}`` mapping from the subprocess, or
- ``None`` if it did not run to completion (crash, import error, ...).
+ The parent can watch the subprocess (its pid) while the test runs, and
+ must wait() for it.
+ """
+
+ def __init__(self, proc, result_path):
+ self._proc = proc
+ self._result_path = result_path
+
+ @property
+ def pid(self):
+ return self._proc.pid
+
+ def wait(self, timeout=None, tick=None, interval=1.0):
+ """Wait for the test to finish, calling *tick* every *interval*
seconds.
+
+ Return ``(payload, output, returncode)``, where *payload* is the
+ decoded ``{'outcomes': ..., 'durations': ...}`` mapping from the
+ subprocess, or ``None`` if it did not run to completion (crash,
+ import error, ...).
+ """
+ import marshal
+ import subprocess
+ import time
+ deadline = None if timeout is None else time.monotonic() + timeout
+ try:
+ while True:
+ step = None if deadline is None else max(
+ 0.0, deadline - time.monotonic())
+ # Wake up for the next tick, unless the timeout comes first.
+ ticking = tick is not None and (step is None or step >
interval)
+ try:
+ # communicate(), not wait(): a test writing more than a
+ # pipe buffer would block. Retrying keeps what it read.
+ stdout, stderr = self._proc.communicate(
+ timeout=interval if ticking else step)
+ break
+ except subprocess.TimeoutExpired:
+ if ticking:
+ tick()
+ continue
+ # Report the hang rather than leaving the runner stuck.
+ self._proc.kill()
+ stdout, stderr = self._proc.communicate()
+ raise _SubprocessTestError(
+ f'test did not complete in a subprocess '
+ f'within {timeout} seconds'
+ ) from _remote(_decode(stdout) + _decode(stderr))
+ try:
+ with open(self._result_path, 'rb') as f:
+ payload = marshal.load(f)
+ except (OSError, EOFError, ValueError):
+ payload = None
+ output = _decode(stdout) + _decode(stderr)
+ return payload, output, self._proc.returncode
+ finally:
+ try:
+ os.unlink(self._result_path)
+ except OSError:
+ pass
+
+
+def _start_test(module, qualname, options=(), env=None):
+ """Start module.qualname (a test method or class) in a fresh subprocess.
+
+ Return a _SubprocessTest. Its wait() is what removes the temporary file
+ the subprocess writes its result to.
"""
import marshal
import subprocess
@@ -129,26 +192,16 @@ def _run_in_subprocess(module, qualname, options, env,
timeout):
cmd = [sys.executable, *options, '-m',
'test.support.subprocess_runner',
module, qualname, result_path,
marshal.dumps(_child_config()).hex()]
- try:
- proc = subprocess.run(cmd, capture_output=True,
- env=_child_environ(env), timeout=timeout)
- except subprocess.TimeoutExpired as exc:
- # Report the hang rather than leaving the test runner stuck.
- output = _decode(exc.stdout) + _decode(exc.stderr)
- raise _SubprocessTestError(
- f'test did not complete in a subprocess '
- f'within {timeout} seconds') from _remote(output)
- try:
- with open(result_path, 'rb') as f:
- payload = marshal.load(f)
- except (OSError, EOFError, ValueError):
- payload = None
- finally:
+ proc = subprocess.Popen(cmd, stdout=subprocess.PIPE,
+ stderr=subprocess.PIPE,
env=_child_environ(env))
+ except BaseException:
try:
os.unlink(result_path)
except OSError:
pass
- return payload, _decode(proc.stdout) + _decode(proc.stderr),
proc.returncode
+ raise
+ return _SubprocessTest(proc, result_path)
+
def _replay_outcome(test, outcome):
@@ -200,6 +253,19 @@ def _check_returncode(returncode, output, what):
raise exc from _remote(output)
+def _replay_test(test, payload, output, returncode):
+ """Reproduce in *test* the result that _SubprocessTest.wait() returned."""
+ if payload is None:
+ exc = _SubprocessTestError(
+ f'test did not complete in a subprocess (exit code {returncode})')
+ raise exc from _remote(output)
+ # The parent measures the test method's own duration (the real cost of the
+ # isolated run, subprocess startup included), so nothing to forward here.
+ # Replay the outcomes first: a failure of the test itself is more useful.
+ _replay_outcomes(test, payload['outcomes'])
+ _check_returncode(returncode, output, 'test')
+
+
def _isolate_method(func, options, env, timeout):
@functools.wraps(func)
def wrapper(self, /, *args, **kwargs):
@@ -209,18 +275,8 @@ def wrapper(self, /, *args, **kwargs):
_check_subprocess_support()
cls = type(self)
qualname = f'{cls.__qualname__}.{func.__name__}'
- payload, output, returncode = _run_in_subprocess(cls.__module__,
- qualname, options,
- env, timeout)
- if payload is None:
- exc = _SubprocessTestError(
- f'test did not complete in a subprocess (exit code
{returncode})')
- raise exc from _remote(output)
- # The parent measures this method's own duration (the real cost of the
- # isolated run, subprocess startup included), so nothing to forward
here.
- # Replay the outcomes first: a failure of the test itself is more
useful.
- _replay_outcomes(self, payload['outcomes'])
- _check_returncode(returncode, output, 'test')
+ proc = _start_test(cls.__module__, qualname, options, env)
+ _replay_test(self, *proc.wait(timeout))
return wrapper
@@ -244,9 +300,8 @@ def setUpClass(cls):
_check_subprocess_support()
# Run the whole class in a single subprocess and stash the outcomes
# for the test methods to replay.
- payload, output, returncode = _run_in_subprocess(cls.__module__,
- cls.__qualname__,
- options, env, timeout)
+ proc = _start_test(cls.__module__, cls.__qualname__, options, env)
+ payload, output, returncode = proc.wait(timeout)
if payload is None:
exc = _SubprocessTestError(
f'class did not complete in a subprocess (exit code
{returncode})')
diff --git a/Lib/test/test_support.py b/Lib/test/test_support.py
index 789aca08180e148..547f31df904f1e4 100644
--- a/Lib/test/test_support.py
+++ b/Lib/test/test_support.py
@@ -965,17 +965,28 @@ def test_timeout_reported_as_error(self):
self.assertEqual(len(result.errors), 1)
self.assertIn(f'within {TIMEOUT} seconds', result.errors[0][1])
+ @support.requires_subprocess()
+ def test_bigmemtest_isolates_a_real_run(self):
+ # A dummy run (no -M) stays in this process, a real run does not.
+ for memlimit in (0, support._1G):
+ with self.subTest(real_max_memuse=memlimit):
+ with support.swap_attr(support, 'real_max_memuse', memlimit):
+ result = self._run('BigmemSample')
+ self.assertEqual(result.testsRun, 1)
+ self.assertEqual(self._names(result.failures), [])
+ self.assertEqual(self._names(result.errors), [])
+
def test_skipped_without_subprocess_support(self):
# On a platform without subprocess support the test is skipped in the
# parent, before any subprocess is spawned.
calls = []
- orig = isolation._run_in_subprocess
+ orig = isolation._start_test
with support.swap_attr(support, 'has_subprocess_support', False):
- isolation._run_in_subprocess = lambda *a, **k: calls.append(a)
+ isolation._start_test = lambda *a, **k: calls.append(a)
try:
result = self._run('MethodSample.test_pass')
finally:
- isolation._run_in_subprocess = orig
+ isolation._start_test = orig
self.assertEqual(result.testsRun, 1)
self.assertEqual(len(result.skipped), 1)
self.assertEqual(calls, [])
diff --git
a/Misc/NEWS.d/next/Tests/2026-08-18-10-16-44.gh-issue-75876.PEZplM.rst
b/Misc/NEWS.d/next/Tests/2026-08-18-10-16-44.gh-issue-75876.PEZplM.rst
new file mode 100644
index 000000000000000..438911cd3877795
--- /dev/null
+++ b/Misc/NEWS.d/next/Tests/2026-08-18-10-16-44.gh-issue-75876.PEZplM.rst
@@ -0,0 +1,5 @@
+A test decorated with :func:`~test.support.bigmemtest` now runs in a
+subprocess if it really allocates the memory it asks for (that is, if the
+``-M`` option is used), so that the memory it uses and the address space it
+fragments are released when it ends. A dummy run stays in the process. The
+separate memory watchdog process is no longer used.
_______________________________________________
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]
