https://github.com/python/cpython/commit/cb60f48ed88b27ccdde9ac067117cc4860aa9852
commit: cb60f48ed88b27ccdde9ac067117cc4860aa9852
branch: main
author: Victor Stinner <[email protected]>
committer: vstinner <[email protected]>
date: 2026-05-25T21:15:12Z
summary:
gh-150114: Use get_process_memory_usage() in memory watchdog (#150402)
In practice, this change adds Windows, FreeBSD and macOS support
to the memory watchdog.
files:
M Lib/test/memory_watchdog.py
M Lib/test/support/__init__.py
diff --git a/Lib/test/memory_watchdog.py b/Lib/test/memory_watchdog.py
index fee062ecc9b300..4a3f66e1f822ba 100644
--- a/Lib/test/memory_watchdog.py
+++ b/Lib/test/memory_watchdog.py
@@ -1,21 +1,40 @@
"""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)
+from test.libregrtest.utils import get_process_memory_usage
+
+
+ONE_GIB = (1024 ** 3)
+
+
+def watchdog(pid):
+ while True:
+ mem = get_process_memory_usage(pid)
+ if mem is None:
+ # get_process_memory_usage() is not supported on the platform,
+ # or something went wrong. Exit since the next call is likely to
+ # fail the same way.
+ return
+
+ # Prefer sys.stdout.write() to print() to use a single write() syscall.
+ # print(msg) calls write(msg.encode()) and then write(b"\n").
+ sys.stdout.write(f" ... process data size: {mem / ONE_GIB:.1f} GiB\n")
+ sys.stdout.flush()
+ time.sleep(1)
+
+def main():
+ if len(sys.argv) != 2:
+ print(f"usage: python {sys.argv[0]} pid")
+ sys.exit(1)
+ pid = int(sys.argv[1])
+
+ try:
+ watchdog(pid)
+ except KeyboardInterrupt:
+ pass
+
+if __name__ == "__main__":
+ main()
diff --git a/Lib/test/support/__init__.py b/Lib/test/support/__init__.py
index 2cac70f4ab2afb..87082ff37d1e58 100644
--- a/Lib/test/support/__init__.py
+++ b/Lib/test/support/__init__.py
@@ -1241,29 +1241,20 @@ class _MemoryWatchdog:
"""
def __init__(self):
- self.procfile = '/proc/{pid}/statm'.format(pid=os.getpid())
self.started = False
def start(self):
- 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)
+ watchdog_script = findfile("memory_watchdog.py")
+ cmd = [sys.executable, watchdog_script, str(os.getpid())]
+ self.mem_watchdog = subprocess.Popen(cmd)
self.started = True
def stop(self):
- if self.started:
- self.mem_watchdog.terminate()
- self.mem_watchdog.wait()
+ if not self.started:
+ return
+ self.mem_watchdog.terminate()
+ self.mem_watchdog.wait()
def bigmemtest(size, memuse, dry_run=True):
@@ -1296,8 +1287,8 @@ def wrapper(self):
if real_max_memuse and verbose:
print()
- print(" ... expected peak memory use: {peak:.1f}G"
- .format(peak=size * memuse / (1024 ** 3)))
+ peak = (size * memuse) / (1024 ** 3)
+ print(f" ... expected peak memory use: {peak:.1f} GiB")
watchdog = _MemoryWatchdog()
watchdog.start()
else:
_______________________________________________
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]
