fix(proxy): tune macOS libmalloc and trim allocator pages so long-lived RSS stays bounded (#2879)
## Summary Fixes #2820. Prevents long-lived macOS proxies from retaining every largest transient request-body allocation in libmalloc. The reporter’s production A/B isolated the allocator behavior and verified the two pre-main libmalloc knobs; this PR applies them through a one-time Darwin-only re-exec and adds periodic per-worker pressure relief. - `MallocAggressiveMadvise=1` returns freed pages eagerly. - `MallocLargeCache=0` disables the large-allocation death-row cache. - Operator-set allocator variables are preserved; `HEADROOM_MALLOC_TUNING=0` is the kill switch. - Periodic trim defaults on only for macOS, runs off the event loop, performs no forced Python GC, validates its interval, and is retained/cancelled through the app lifecycle. - Non-Darwin behavior remains unchanged unless explicitly enabled. - Semantically rebased onto current `main`, retaining startup dependency validation, MCP SDK v1 compatibility, and all newer proxy behavior. ## Verification - 147 proxy CLI/config/malloc/MCP-contract tests pass; 1 platform skip. - Ruff check and formatting clean; `git diff --check` clean. - The reporter’s macOS A/B reduced dirty empty malloc regions to zero and lowered steady/startup RSS; the control flow and shutdown lifecycle are covered locally. ## Safety The re-exec is Darwin-only, PID-preserving, loop-guarded, and opt-out. The trim task is per worker because allocator state is per process, and shutdown cancels it explicitly.
This commit is contained in:
Abhay Singh
@@ -113,6 +113,35 @@ def _get_env_bool_optional(name: str) -> bool | None:
|
||||
return _get_env_bool(name, False)
|
||||
|
||||
|
||||
# libmalloc reads these before main() runs, so they cannot be set from inside
|
||||
# the current process — the proxy re-execs itself once to apply them. Without
|
||||
# them, freed pages from large concurrent request bodies stay resident
|
||||
# (``vmmap`` shows whole "MALLOC_LARGE (empty)" regions) and long-lived proxy
|
||||
# RSS only ratchets upward (#2820). Vars the operator already set are left
|
||||
# untouched; HEADROOM_MALLOC_TUNING=0 disables the re-exec entirely.
|
||||
_MALLOC_TUNING = {
|
||||
"MallocAggressiveMadvise": "1", # madvise freed pages back to the OS eagerly
|
||||
"MallocLargeCache": "0", # no death-row cache for freed large allocations
|
||||
}
|
||||
|
||||
|
||||
def _reexec_with_malloc_tuning() -> None:
|
||||
if sys.platform != "darwin":
|
||||
return
|
||||
if not _get_env_bool("HEADROOM_MALLOC_TUNING", True):
|
||||
return
|
||||
if os.environ.get("_HEADROOM_MALLOC_TUNED") == "1":
|
||||
return
|
||||
missing = {k: v for k, v in _MALLOC_TUNING.items() if k not in os.environ}
|
||||
# Set the loop guard before the re-exec so the replacement process (which
|
||||
# inherits this environment) skips this path instead of re-execing forever.
|
||||
os.environ["_HEADROOM_MALLOC_TUNED"] = "1"
|
||||
if not missing:
|
||||
return
|
||||
os.environ.update(missing)
|
||||
os.execv(sys.executable, [sys.executable, "-m", "headroom.cli", *sys.argv[1:]])
|
||||
|
||||
|
||||
def _get_env_int_optional(name: str) -> int | None:
|
||||
val = os.environ.get(name)
|
||||
if val is None or val == "":
|
||||
@@ -1065,6 +1094,7 @@ def proxy(
|
||||
Usage with OpenAI-compatible clients:
|
||||
OPENAI_BASE_URL=http://localhost:8787/v1 your-app
|
||||
"""
|
||||
_reexec_with_malloc_tuning()
|
||||
ensure_proxy_dependencies()
|
||||
|
||||
# Import here to avoid slow startup
|
||||
@@ -1261,6 +1291,10 @@ def proxy(
|
||||
rate_limit_requests_per_minute=rpm if rpm is not None else 60,
|
||||
rate_limit_tokens_per_minute=tpm if tpm is not None else 100_000,
|
||||
compress_user_messages=_get_env_bool("HEADROOM_COMPRESS_USER_MESSAGES", False),
|
||||
periodic_malloc_trim_enabled=_get_env_bool(
|
||||
"HEADROOM_MALLOC_TRIM", sys.platform == "darwin"
|
||||
),
|
||||
malloc_trim_interval_seconds=_get_env_int("HEADROOM_MALLOC_TRIM_INTERVAL_SECONDS", 60),
|
||||
min_tokens_to_crush=_get_env_int("HEADROOM_MIN_TOKENS", 500),
|
||||
max_items_after_crush=_get_env_int("HEADROOM_MAX_ITEMS", 50),
|
||||
exclude_tools=_parse_exclude_tools(None) or None,
|
||||
|
||||
@@ -0,0 +1,135 @@
|
||||
"""Return freed-but-retained allocator pages to the OS on long-lived proxies.
|
||||
|
||||
Large concurrent Anthropic bodies (0.5-1 MB of JSON parsed, deep-copied and
|
||||
re-serialized per in-flight request) drive libmalloc and pymalloc to a
|
||||
high-water mark that is never returned to the OS: after a burst the malloc
|
||||
zones keep entire regions resident but empty (``vmmap`` lists them as
|
||||
``MALLOC_LARGE (empty)`` / ``MALLOC_SMALL (empty)``), so process RSS only
|
||||
ratchets upward. Over a multi-day proxy lifetime under Claude Code traffic
|
||||
this reaches double-digit GB and starves the host.
|
||||
|
||||
Neither runtime returns these pages on its own. macOS exposes
|
||||
``malloc_zone_pressure_relief(NULL, 0)`` to purge every zone's free pages;
|
||||
glibc has ``malloc_trim(0)``. ``trim()`` calls that entry point directly: it is
|
||||
a C call that releases the GIL and reclaims whatever is already on the
|
||||
allocator's free lists. It deliberately does not run a Python ``gc.collect()``
|
||||
-- a full cyclic collection holds the GIL, and this periodic task runs off the
|
||||
event-loop thread precisely so it cannot stall request handling; freeing cyclic
|
||||
garbage is left to CPython's own automatic collection.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import ctypes
|
||||
import logging
|
||||
import sys
|
||||
import time
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Interval bounds for the periodic trim task. A non-positive interval would make
|
||||
# ``asyncio.sleep`` return immediately and spin a continuous collect/trim loop,
|
||||
# so anything below the minimum falls back to the default.
|
||||
_DEFAULT_TRIM_INTERVAL_SECONDS = 60
|
||||
_MIN_TRIM_INTERVAL_SECONDS = 1
|
||||
|
||||
# Lazily resolved (platform_tag, foreign_function | None). ``None`` function
|
||||
# means the platform has no supported trim call and trim() is a no-op.
|
||||
_relief: tuple[str, object | None] | None = None
|
||||
|
||||
|
||||
def _resolve() -> tuple[str, object | None]:
|
||||
global _relief
|
||||
if _relief is not None:
|
||||
return _relief
|
||||
try:
|
||||
libc = ctypes.CDLL(None)
|
||||
if sys.platform == "darwin":
|
||||
fn = libc.malloc_zone_pressure_relief
|
||||
fn.argtypes = [ctypes.c_void_p, ctypes.c_size_t]
|
||||
fn.restype = ctypes.c_size_t
|
||||
_relief = ("darwin", fn)
|
||||
else:
|
||||
fn = libc.malloc_trim
|
||||
fn.argtypes = [ctypes.c_size_t]
|
||||
fn.restype = ctypes.c_int
|
||||
_relief = ("glibc", fn)
|
||||
except (OSError, AttributeError):
|
||||
_relief = ("unsupported", None)
|
||||
return _relief
|
||||
|
||||
|
||||
def trim() -> int:
|
||||
"""Return allocator free pages to the OS.
|
||||
|
||||
Calls the platform's allocator pressure-relief entry point
|
||||
(``malloc_zone_pressure_relief`` on macOS, ``malloc_trim`` on glibc). This
|
||||
is a C call that releases the GIL for its duration and reclaims pages
|
||||
already on the allocator's free lists. It deliberately does *not* run a
|
||||
Python ``gc.collect()`` (a full cyclic collection holds the GIL); cyclic
|
||||
garbage is left to CPython's automatic collection, so this off-thread
|
||||
periodic task never holds the GIL for a full-heap traversal.
|
||||
|
||||
Returns the number of bytes freed on macOS (glibc's ``malloc_trim``
|
||||
reports only success, so 0 is returned there and on unsupported
|
||||
platforms).
|
||||
"""
|
||||
kind, fn = _resolve()
|
||||
if fn is None:
|
||||
return 0
|
||||
if kind == "darwin":
|
||||
return int(fn(None, 0)) # type: ignore[operator]
|
||||
fn(0) # type: ignore[operator]
|
||||
return 0
|
||||
|
||||
|
||||
async def trim_periodically(interval_seconds: int = 60) -> None:
|
||||
"""Background task that periodically returns allocator free pages to the OS.
|
||||
|
||||
Runs in every worker process (allocator state is per-process). The trim is
|
||||
the platform's allocator pressure-relief C call
|
||||
(``malloc_zone_pressure_relief``/``malloc_trim``), dispatched via
|
||||
``asyncio.to_thread`` so it runs off the event-loop thread. Because it is a
|
||||
C call that releases the GIL and runs no Python ``gc.collect()``, it holds
|
||||
the GIL only as briefly as the to_thread hand-off, so a slow purge on a
|
||||
large heap does not stall request handling. The task exits immediately on
|
||||
platforms with no supported trim call, so it is a true no-op there.
|
||||
|
||||
Args:
|
||||
interval_seconds: How often to trim (default: 60 seconds). A value below
|
||||
``_MIN_TRIM_INTERVAL_SECONDS`` (which would busy-loop) falls back to
|
||||
the default.
|
||||
"""
|
||||
_, fn = _resolve()
|
||||
if fn is None:
|
||||
# No supported allocator-trim call on this platform (Windows, musl, ...);
|
||||
# do not spin a wakeup task that can only ever no-op.
|
||||
logger.debug("MallocTrim: no supported trim on %s; task disabled", sys.platform)
|
||||
return
|
||||
|
||||
if interval_seconds < _MIN_TRIM_INTERVAL_SECONDS:
|
||||
logger.warning(
|
||||
"MallocTrim: interval %ss is below the %ds minimum; using default %ds",
|
||||
interval_seconds,
|
||||
_MIN_TRIM_INTERVAL_SECONDS,
|
||||
_DEFAULT_TRIM_INTERVAL_SECONDS,
|
||||
)
|
||||
interval_seconds = _DEFAULT_TRIM_INTERVAL_SECONDS
|
||||
|
||||
while True:
|
||||
await asyncio.sleep(interval_seconds)
|
||||
try:
|
||||
start = time.perf_counter()
|
||||
# Off the event-loop thread: the C-level purge can pause for a while
|
||||
# on a large heap, and that pause must not stall proxy traffic.
|
||||
freed = await asyncio.to_thread(trim)
|
||||
elapsed_ms = (time.perf_counter() - start) * 1000
|
||||
log = logger.info if freed >= (16 << 20) else logger.debug
|
||||
log(
|
||||
"MallocTrim: returned %.1f MB to OS in %.0f ms",
|
||||
freed / 1048576,
|
||||
elapsed_ms,
|
||||
)
|
||||
except Exception as e:
|
||||
logger.debug("MallocTrim failed: %s", e)
|
||||
@@ -7,6 +7,7 @@ Extracted from server.py to keep the codebase maintainable.
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import sys
|
||||
from dataclasses import InitVar, dataclass, field
|
||||
from datetime import datetime
|
||||
from typing import Any, Literal
|
||||
@@ -439,6 +440,17 @@ class ProxyConfig:
|
||||
# Env: HEADROOM_PERIODIC_TOIN_STATS=0.
|
||||
periodic_toin_stats_enabled: bool = True
|
||||
|
||||
# Periodic allocator trim. Long-lived proxies processing large concurrent
|
||||
# request bodies ratchet RSS through freed-but-retained allocator pages;
|
||||
# this returns them to the OS (malloc_zone_pressure_relief on macOS,
|
||||
# malloc_trim on glibc). Default-on only on macOS, where the retained-page
|
||||
# ratchet is the documented failure (#2820); an opt-in elsewhere via
|
||||
# HEADROOM_MALLOC_TRIM=1 so glibc deployments do not silently take on a
|
||||
# once-a-minute allocator purge they did not ask for. Envs:
|
||||
# HEADROOM_MALLOC_TRIM=0/1, HEADROOM_MALLOC_TRIM_INTERVAL_SECONDS.
|
||||
periodic_malloc_trim_enabled: bool = field(default_factory=lambda: sys.platform == "darwin")
|
||||
malloc_trim_interval_seconds: int = 60
|
||||
|
||||
# Stateless mode — disable all filesystem writes for read-only / container deployments
|
||||
stateless: bool = False
|
||||
|
||||
|
||||
@@ -144,6 +144,7 @@ from headroom.proxy.helpers import (
|
||||
)
|
||||
from headroom.proxy.loop_callback_failure_policy import is_known_websocket_callback_failure
|
||||
from headroom.proxy.loopback_guard import is_loopback_host
|
||||
from headroom.proxy.malloc_trim import trim_periodically
|
||||
from headroom.proxy.memory_handler import MemoryConfig, MemoryHandler
|
||||
|
||||
# Data models (extracted to headroom/proxy/models.py for maintainability)
|
||||
@@ -2601,6 +2602,7 @@ def create_app(config: ProxyConfig | None = None) -> FastAPI:
|
||||
app.state.ready = False
|
||||
app.state.startup_error = None
|
||||
app.state.periodic_toin_stats_task = None
|
||||
app.state.periodic_malloc_trim_task = None
|
||||
|
||||
try:
|
||||
try:
|
||||
@@ -2611,6 +2613,12 @@ def create_app(config: ProxyConfig | None = None) -> FastAPI:
|
||||
app.state.periodic_toin_stats_task = asyncio.create_task(
|
||||
_log_toin_stats_periodically()
|
||||
)
|
||||
# Per-worker on purpose: allocator state is per-process, so
|
||||
# every worker must trim its own zones (no beacon-owner gate).
|
||||
if config.periodic_malloc_trim_enabled:
|
||||
app.state.periodic_malloc_trim_task = asyncio.create_task(
|
||||
trim_periodically(config.malloc_trim_interval_seconds)
|
||||
)
|
||||
if proxy.usage_reporter:
|
||||
await proxy.usage_reporter.start(proxy)
|
||||
if proxy.traffic_learner:
|
||||
@@ -2670,6 +2678,16 @@ def create_app(config: ProxyConfig | None = None) -> FastAPI:
|
||||
)
|
||||
app.state.periodic_toin_stats_task = None
|
||||
|
||||
periodic_malloc_trim_task = app.state.periodic_malloc_trim_task
|
||||
if periodic_malloc_trim_task is not None:
|
||||
periodic_malloc_trim_task.cancel()
|
||||
await _timed(
|
||||
asyncio.gather(periodic_malloc_trim_task, return_exceptions=True),
|
||||
label="periodic_malloc_trim.stop",
|
||||
timeout=3.0,
|
||||
)
|
||||
app.state.periodic_malloc_trim_task = None
|
||||
|
||||
if _cc_reconciler is not None:
|
||||
await _timed(_cc_reconciler.stop(), label="cc_reconciler.stop", timeout=3.0)
|
||||
if _beacon_is_owner[0]:
|
||||
@@ -5160,6 +5178,10 @@ def _proxy_config_from_env() -> ProxyConfig:
|
||||
http2=_get_env_bool("HEADROOM_HTTP2", True),
|
||||
http_proxy=os.environ.get("HEADROOM_HTTP_PROXY") or None,
|
||||
periodic_toin_stats_enabled=_get_env_bool("HEADROOM_PERIODIC_TOIN_STATS", True),
|
||||
periodic_malloc_trim_enabled=_get_env_bool(
|
||||
"HEADROOM_MALLOC_TRIM", sys.platform == "darwin"
|
||||
),
|
||||
malloc_trim_interval_seconds=_get_env_int("HEADROOM_MALLOC_TRIM_INTERVAL_SECONDS", 60),
|
||||
proxy_token=os.environ.get("HEADROOM_PROXY_TOKEN") or None,
|
||||
offline=_get_env_bool("HEADROOM_OFFLINE", False),
|
||||
# Default mode is CACHE (Headroom's coding posture): delta-only compression
|
||||
|
||||
@@ -0,0 +1,288 @@
|
||||
"""macOS libmalloc tuning: pre-main re-exec gating + periodic allocator trim (#2820)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
|
||||
import pytest
|
||||
|
||||
import headroom.cli.proxy as proxy_cli
|
||||
from headroom.proxy import malloc_trim
|
||||
|
||||
|
||||
class _ExecCalled(Exception):
|
||||
"""Sentinel so a fake execv can stop execution the way real execv would."""
|
||||
|
||||
|
||||
def _fake_execv(recorder: dict):
|
||||
def _execv(path, argv): # noqa: ANN001
|
||||
recorder["path"] = path
|
||||
recorder["argv"] = list(argv)
|
||||
raise _ExecCalled
|
||||
|
||||
return _execv
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _clean_malloc_env(monkeypatch):
|
||||
for var in (
|
||||
"HEADROOM_MALLOC_TUNING",
|
||||
"_HEADROOM_MALLOC_TUNED",
|
||||
"MallocAggressiveMadvise",
|
||||
"MallocLargeCache",
|
||||
):
|
||||
monkeypatch.delenv(var, raising=False)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# _reexec_with_malloc_tuning
|
||||
# --------------------------------------------------------------------------- #
|
||||
def test_reexec_noop_off_darwin(monkeypatch):
|
||||
monkeypatch.setattr(proxy_cli.sys, "platform", "linux")
|
||||
rec: dict = {}
|
||||
monkeypatch.setattr(proxy_cli.os, "execv", _fake_execv(rec))
|
||||
proxy_cli._reexec_with_malloc_tuning() # must not raise / exec
|
||||
assert rec == {}
|
||||
|
||||
|
||||
def test_reexec_respects_opt_out(monkeypatch):
|
||||
monkeypatch.setattr(proxy_cli.sys, "platform", "darwin")
|
||||
monkeypatch.setenv("HEADROOM_MALLOC_TUNING", "0")
|
||||
rec: dict = {}
|
||||
monkeypatch.setattr(proxy_cli.os, "execv", _fake_execv(rec))
|
||||
proxy_cli._reexec_with_malloc_tuning()
|
||||
assert rec == {}
|
||||
|
||||
|
||||
def test_reexec_guard_prevents_loop(monkeypatch):
|
||||
monkeypatch.setattr(proxy_cli.sys, "platform", "darwin")
|
||||
monkeypatch.setenv("_HEADROOM_MALLOC_TUNED", "1")
|
||||
rec: dict = {}
|
||||
monkeypatch.setattr(proxy_cli.os, "execv", _fake_execv(rec))
|
||||
proxy_cli._reexec_with_malloc_tuning()
|
||||
assert rec == {}
|
||||
|
||||
|
||||
def test_reexec_skips_when_operator_already_set_vars(monkeypatch):
|
||||
monkeypatch.setattr(proxy_cli.sys, "platform", "darwin")
|
||||
monkeypatch.setenv("MallocAggressiveMadvise", "1")
|
||||
monkeypatch.setenv("MallocLargeCache", "0")
|
||||
rec: dict = {}
|
||||
monkeypatch.setattr(proxy_cli.os, "execv", _fake_execv(rec))
|
||||
proxy_cli._reexec_with_malloc_tuning()
|
||||
# No re-exec (vars present), but the guard is still stamped.
|
||||
assert rec == {}
|
||||
assert proxy_cli.os.environ.get("_HEADROOM_MALLOC_TUNED") == "1"
|
||||
|
||||
|
||||
def test_reexec_sets_vars_and_execs_once(monkeypatch):
|
||||
monkeypatch.setattr(proxy_cli.sys, "platform", "darwin")
|
||||
monkeypatch.setattr(proxy_cli.sys, "executable", "/usr/bin/python3")
|
||||
monkeypatch.setattr(proxy_cli.sys, "argv", ["headroom", "proxy", "--port", "8787"])
|
||||
rec: dict = {}
|
||||
monkeypatch.setattr(proxy_cli.os, "execv", _fake_execv(rec))
|
||||
|
||||
with pytest.raises(_ExecCalled):
|
||||
proxy_cli._reexec_with_malloc_tuning()
|
||||
|
||||
# The tuning knobs and the loop guard are exported to the replacement process.
|
||||
assert proxy_cli.os.environ["MallocAggressiveMadvise"] == "1"
|
||||
assert proxy_cli.os.environ["MallocLargeCache"] == "0"
|
||||
assert proxy_cli.os.environ["_HEADROOM_MALLOC_TUNED"] == "1"
|
||||
# Re-exec normalizes to `python -m headroom.cli <args>`, preserving the PID.
|
||||
assert rec["path"] == "/usr/bin/python3"
|
||||
assert rec["argv"] == ["/usr/bin/python3", "-m", "headroom.cli", "proxy", "--port", "8787"]
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# malloc_trim.trim / trim_periodically
|
||||
# --------------------------------------------------------------------------- #
|
||||
def test_trim_calls_platform_fn(monkeypatch):
|
||||
def fake_fn(ptr, size): # noqa: ANN001 (mac signature)
|
||||
return 4096
|
||||
|
||||
monkeypatch.setattr(malloc_trim, "_resolve", lambda: ("darwin", fake_fn))
|
||||
assert malloc_trim.trim() == 4096
|
||||
|
||||
|
||||
def test_trim_never_runs_python_gc(monkeypatch):
|
||||
# The periodic trim must NOT trigger a full cyclic collection: gc.collect()
|
||||
# holds the GIL for a whole-heap traversal, which would stall the event loop
|
||||
# even though the C purge itself is dispatched off-thread. Only the
|
||||
# GIL-releasing allocator C call may run.
|
||||
import gc
|
||||
|
||||
ran: list[str] = []
|
||||
monkeypatch.setattr(gc, "collect", lambda *a, **k: ran.append("gc") or 0)
|
||||
monkeypatch.setattr(malloc_trim, "_resolve", lambda: ("glibc", lambda _size: 0))
|
||||
malloc_trim.trim()
|
||||
assert ran == []
|
||||
|
||||
|
||||
def test_trim_is_noop_on_unsupported_platform(monkeypatch):
|
||||
monkeypatch.setattr(malloc_trim, "_resolve", lambda: ("unsupported", None))
|
||||
assert malloc_trim.trim() == 0
|
||||
|
||||
|
||||
def test_trim_periodically_trims_each_interval(monkeypatch):
|
||||
monkeypatch.setattr(malloc_trim, "_resolve", lambda: ("glibc", object()))
|
||||
trims: list[int] = []
|
||||
monkeypatch.setattr(malloc_trim, "trim", lambda: trims.append(1) or 0)
|
||||
|
||||
async def fake_sleep(_seconds):
|
||||
if len(trims) >= 2: # let two ticks run, then break the loop
|
||||
raise asyncio.CancelledError
|
||||
|
||||
monkeypatch.setattr(malloc_trim.asyncio, "sleep", fake_sleep)
|
||||
|
||||
with pytest.raises(asyncio.CancelledError):
|
||||
asyncio.run(malloc_trim.trim_periodically(interval_seconds=1))
|
||||
assert len(trims) == 2
|
||||
|
||||
|
||||
def test_trim_periodically_is_disabled_on_unsupported_platform(monkeypatch):
|
||||
# No supported trim call: the task must return at once, never scheduling a
|
||||
# wakeup (so it is a true no-op on Windows/musl, not a 60s spinner).
|
||||
monkeypatch.setattr(malloc_trim, "_resolve", lambda: ("unsupported", None))
|
||||
trims: list[int] = []
|
||||
monkeypatch.setattr(malloc_trim, "trim", lambda: trims.append(1) or 0)
|
||||
|
||||
async def _no_sleep(_seconds):
|
||||
raise AssertionError("unsupported platform must not schedule a trim wakeup")
|
||||
|
||||
monkeypatch.setattr(malloc_trim.asyncio, "sleep", _no_sleep)
|
||||
asyncio.run(malloc_trim.trim_periodically(interval_seconds=60)) # returns, no raise
|
||||
assert trims == []
|
||||
|
||||
|
||||
@pytest.mark.parametrize("bad_interval", [0, -5])
|
||||
def test_trim_periodically_rejects_non_positive_interval(monkeypatch, bad_interval):
|
||||
# A non-positive interval would make asyncio.sleep return immediately and
|
||||
# spin a continuous collect/trim loop; it must fall back to the default.
|
||||
monkeypatch.setattr(malloc_trim, "_resolve", lambda: ("glibc", object()))
|
||||
monkeypatch.setattr(malloc_trim, "trim", lambda: 0)
|
||||
slept: list[float] = []
|
||||
|
||||
async def capture_sleep(seconds):
|
||||
slept.append(seconds)
|
||||
raise asyncio.CancelledError # stop after the first sleep
|
||||
|
||||
monkeypatch.setattr(malloc_trim.asyncio, "sleep", capture_sleep)
|
||||
with pytest.raises(asyncio.CancelledError):
|
||||
asyncio.run(malloc_trim.trim_periodically(interval_seconds=bad_interval))
|
||||
assert slept == [malloc_trim._DEFAULT_TRIM_INTERVAL_SECONDS]
|
||||
|
||||
|
||||
def test_trim_runs_off_the_event_loop_thread(monkeypatch):
|
||||
# The blocking trim must run in a worker thread (via asyncio.to_thread), not
|
||||
# on the event loop, so a slow trim cannot stall other async work.
|
||||
import threading
|
||||
|
||||
monkeypatch.setattr(malloc_trim, "_resolve", lambda: ("glibc", object()))
|
||||
seen: dict[str, int] = {}
|
||||
|
||||
def record():
|
||||
seen["thread"] = threading.get_ident()
|
||||
return 0
|
||||
|
||||
monkeypatch.setattr(malloc_trim, "trim", record)
|
||||
calls = {"n": 0}
|
||||
|
||||
async def sleeper(_seconds):
|
||||
calls["n"] += 1
|
||||
if calls["n"] >= 2: # first sleep returns; after the trim, stop
|
||||
raise asyncio.CancelledError
|
||||
|
||||
monkeypatch.setattr(malloc_trim.asyncio, "sleep", sleeper)
|
||||
|
||||
async def _run() -> int:
|
||||
loop_thread = threading.get_ident()
|
||||
with pytest.raises(asyncio.CancelledError):
|
||||
await malloc_trim.trim_periodically(interval_seconds=60)
|
||||
return loop_thread
|
||||
|
||||
loop_thread = asyncio.run(_run())
|
||||
assert "thread" in seen # trim actually ran
|
||||
assert seen["thread"] != loop_thread # ran off the event-loop thread
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_slow_trim_does_not_stop_unrelated_async_work(monkeypatch):
|
||||
# The periodic trim is dispatched off the event-loop thread via
|
||||
# asyncio.to_thread and runs no Python gc.collect(), so even a slow purge
|
||||
# must not freeze the loop. It is modeled here with a worker-thread park
|
||||
# which, like the real GIL-releasing allocator C call, does not hold the
|
||||
# GIL while it waits: unrelated coroutines keep making progress meanwhile.
|
||||
import threading
|
||||
|
||||
monkeypatch.setattr(malloc_trim, "_resolve", lambda: ("glibc", object()))
|
||||
started = threading.Event()
|
||||
release = threading.Event()
|
||||
|
||||
def slow_trim() -> int:
|
||||
started.set()
|
||||
release.wait(5.0) # hold the worker thread until the test lets go
|
||||
return 0
|
||||
|
||||
monkeypatch.setattr(malloc_trim, "trim", slow_trim)
|
||||
|
||||
# Fire the trim's interval immediately (the interval is >= 1s) while leaving
|
||||
# the counter's sub-second sleeps to behave normally.
|
||||
real_sleep = asyncio.sleep
|
||||
|
||||
async def smart_sleep(seconds):
|
||||
if seconds >= 1:
|
||||
return
|
||||
await real_sleep(seconds)
|
||||
|
||||
monkeypatch.setattr(malloc_trim.asyncio, "sleep", smart_sleep)
|
||||
|
||||
ticks = 0
|
||||
|
||||
async def counter() -> None:
|
||||
nonlocal ticks
|
||||
while True:
|
||||
await real_sleep(0.005)
|
||||
ticks += 1
|
||||
|
||||
counter_task = asyncio.create_task(counter())
|
||||
trim_task = asyncio.create_task(malloc_trim.trim_periodically(interval_seconds=60))
|
||||
try:
|
||||
# Wait for the trim to actually start blocking a worker thread.
|
||||
for _ in range(400):
|
||||
if started.is_set():
|
||||
break
|
||||
await real_sleep(0.005)
|
||||
assert started.is_set(), "trim never started"
|
||||
|
||||
# The trim is now parked off-loop. The event loop must keep ticking.
|
||||
ticks_before = ticks
|
||||
await real_sleep(0.2)
|
||||
ticks_during_trim = ticks - ticks_before
|
||||
finally:
|
||||
release.set()
|
||||
counter_task.cancel()
|
||||
trim_task.cancel()
|
||||
|
||||
# On-loop blocking would freeze the counter (~0 ticks); off-thread it keeps
|
||||
# ticking (~40 in 0.2s). Generous floor for scheduler jitter.
|
||||
assert ticks_during_trim >= 10
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# ProxyConfig wiring
|
||||
# --------------------------------------------------------------------------- #
|
||||
def test_proxy_config_malloc_trim_default_is_darwin_scoped(monkeypatch):
|
||||
# Default-on only on macOS (the platform with the documented RSS ratchet);
|
||||
# elsewhere it is opt-in, so glibc deployments do not silently take on a
|
||||
# once-a-minute allocator purge.
|
||||
from headroom.proxy import models
|
||||
|
||||
monkeypatch.setattr(models.sys, "platform", "darwin")
|
||||
assert models.ProxyConfig().periodic_malloc_trim_enabled is True
|
||||
|
||||
monkeypatch.setattr(models.sys, "platform", "linux")
|
||||
assert models.ProxyConfig().periodic_malloc_trim_enabled is False
|
||||
|
||||
# The interval knob is platform-independent.
|
||||
assert models.ProxyConfig().malloc_trim_interval_seconds == 60
|
||||
Reference in New Issue
Block a user