From 712751fa6e4249ab3d030a2d2e2425cd77941fd3 Mon Sep 17 00:00:00 2001 From: Serhiy Storchaka Date: Tue, 18 Aug 2026 15:55:49 +0300 Subject: [PATCH] [3.13] gh-75876: Run bigmem tests in a subprocess (GH-155302) 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) Co-authored-by: Serhiy Storchaka --- Lib/test/_isolated_sample.py | 10 ++ Lib/test/memory_watchdog.py | 21 --- Lib/test/support/__init__.py | 63 +++------ Lib/test/support/isolation.py | 127 +++++++++++++----- Lib/test/test_support.py | 17 ++- ...6-08-18-10-16-44.gh-issue-75876.PEZplM.rst | 5 + 6 files changed, 139 insertions(+), 104 deletions(-) delete mode 100644 Lib/test/memory_watchdog.py create mode 100644 Misc/NEWS.d/next/Tests/2026-08-18-10-16-44.gh-issue-75876.PEZplM.rst 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//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 74a2a1ba39a3232..fabddb2e1482e4a 100644 --- a/Lib/test/test_support.py +++ b/Lib/test/test_support.py @@ -960,17 +960,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.