fix: follow-ups for salvaged PR #80795
- Drain the eviction plan (pop + del) before trim_memory: the batch
thread previously held every evicted agent in its local list while
gc.collect + malloc_trim ran, so the in-pass trim freed almost
nothing, the next tick re-read a still-high RSS, and the valve
over-evicted an extra batch of warm prompt caches per cycle.
- Clear _db_flush_scan_prefix in _release_evicted_agent_soft: it is a
shallow copy of the flushed transcript (stamped on every successful
flush) sharing every message dict — and pressure-evictable agents
have flushed by definition, so it pinned the multi-MB content strings
on exactly the agents the valve targets.
- Config-read failure now falls back to resolve_agent_cache_bounds({})
instead of bare AgentCacheBounds(): the dataclass default disables
the pressure pass, but an absent config section means 'auto' — a
transient read failure must not permanently switch off the OOM valve.
- protect_recent: false (YAML bool; False == 0) keeps the default MRU
protection instead of silently disabling it.
- 'No evictable session' warning now distinguishes sessions blocked on
un-flushed persistence (e.g. session DB never initialized — NFS
HERMES_HOME) from mid-turn agents, so operators can diagnose why the
valve isn't shedding instead of being pointed at running turns.
- _cgroup_limit_bytes checks the process's own cgroup (via the existing
gateway.cgroup_cleanup._own_cgroup_path) before the root files, so a
systemd unit's MemoryHigh=/MemoryMax= is detected — the root
memory.high/max read 'max' on those deployments.
- Tests: 5 new guards; drain-before-trim and scan-prefix-clear
mutation-checked (revert each fix -> its guard fails).
This commit is contained in:
kshitij
parent
2d0c2682c7
commit
83bad5cdda
|
|
@ -93,13 +93,34 @@ def _cgroup_limit_bytes() -> Optional[int]:
|
|||
Prefers cgroup v2 ``memory.high`` (the throttling point — passing it is
|
||||
what stalled the reported shutdown) over ``memory.max``, and falls back to
|
||||
cgroup v1. ``max`` / absurd sentinel values mean "unlimited".
|
||||
|
||||
Checks the process's *own* cgroup first (where a systemd unit's
|
||||
``MemoryHigh=``/``MemoryMax=`` actually lands — the root files read
|
||||
``max`` on those deployments), then walks up to the root for
|
||||
container-style limits.
|
||||
"""
|
||||
if sys.platform != "linux":
|
||||
return None
|
||||
candidates = (
|
||||
"/sys/fs/cgroup/memory.high",
|
||||
"/sys/fs/cgroup/memory.max",
|
||||
"/sys/fs/cgroup/memory/memory.limit_in_bytes",
|
||||
candidates: list[str] = []
|
||||
try:
|
||||
from gateway.cgroup_cleanup import _own_cgroup_path
|
||||
|
||||
own = _own_cgroup_path()
|
||||
except Exception:
|
||||
own = None
|
||||
if own and own != "/":
|
||||
candidates.extend(
|
||||
(
|
||||
f"/sys/fs/cgroup{own}/memory.high",
|
||||
f"/sys/fs/cgroup{own}/memory.max",
|
||||
)
|
||||
)
|
||||
candidates.extend(
|
||||
(
|
||||
"/sys/fs/cgroup/memory.high",
|
||||
"/sys/fs/cgroup/memory.max",
|
||||
"/sys/fs/cgroup/memory/memory.limit_in_bytes",
|
||||
)
|
||||
)
|
||||
for candidate in candidates:
|
||||
try:
|
||||
|
|
@ -179,7 +200,15 @@ def resolve_agent_cache_bounds(config: Any) -> AgentCacheBounds:
|
|||
max_evictions = _positive_int(section.get("max_evictions_per_pass"))
|
||||
protect_recent = section.get("protect_recent")
|
||||
protect_parsed = _positive_int(protect_recent)
|
||||
if protect_parsed is None and protect_recent == 0:
|
||||
if (
|
||||
protect_parsed is None
|
||||
and isinstance(protect_recent, int)
|
||||
and not isinstance(protect_recent, bool)
|
||||
and protect_recent == 0
|
||||
):
|
||||
# 0 means "shed anything" — distinct from unset. The isinstance
|
||||
# guards keep `protect_recent: false` (a YAML-typo bool, False == 0)
|
||||
# on the default instead of silently disabling MRU protection.
|
||||
protect_parsed = 0
|
||||
|
||||
return AgentCacheBounds(
|
||||
|
|
|
|||
|
|
@ -23747,6 +23747,14 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
# persisted session JSON on the next turn, so dropping it here is safe.
|
||||
if hasattr(agent, "_session_messages"):
|
||||
agent._session_messages = []
|
||||
# _db_flush_scan_prefix is a shallow copy of the flushed transcript
|
||||
# (run_agent.py, stamped on every successful flush) — it shares every
|
||||
# message dict, so leaving it pins the multi-MB content strings the
|
||||
# eviction exists to free. Pressure-evictable agents have flushed by
|
||||
# definition, so this attribute is always populated on exactly the
|
||||
# agents the memory valve targets.
|
||||
if hasattr(agent, "_db_flush_scan_prefix"):
|
||||
agent._db_flush_scan_prefix = None
|
||||
|
||||
def _agent_cache_bounds(self):
|
||||
"""Operator-configured agent-cache bounds, resolved once per process.
|
||||
|
|
@ -23762,10 +23770,14 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
try:
|
||||
bounds = resolve_agent_cache_bounds(_load_gateway_config())
|
||||
except Exception as _e:
|
||||
from gateway.agent_cache_pressure import AgentCacheBounds
|
||||
|
||||
logger.debug("Agent cache bounds config read failed: %s", _e)
|
||||
bounds = AgentCacheBounds()
|
||||
# Resolve from an empty config rather than bare
|
||||
# AgentCacheBounds(): the dataclass default has
|
||||
# memory_high_mb=None (pressure pass OFF), but an *absent*
|
||||
# config section means "auto" — a transient config read
|
||||
# failure must not permanently disable the OOM valve this
|
||||
# feature exists to provide.
|
||||
bounds = resolve_agent_cache_bounds({})
|
||||
self._agent_cache_bounds_cache = bounds
|
||||
return bounds
|
||||
|
||||
|
|
@ -23850,18 +23862,34 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
|
||||
if not plan:
|
||||
_mid_turn = sum(1 for _, a in ordered if a is not None and id(a) in running_ids)
|
||||
_unflushed = sum(
|
||||
1
|
||||
for _, a in ordered
|
||||
if a is not None
|
||||
and a is not _AGENT_PENDING_SENTINEL
|
||||
and id(a) not in running_ids
|
||||
and not transcript_persistence_caught_up(a)
|
||||
)
|
||||
logger.warning(
|
||||
"Agent cache pressure: anon RSS %dMB over budget %dMB but no "
|
||||
"evictable session (%d cached, %d mid-turn) — memory will keep "
|
||||
"climbing until those turns finish.",
|
||||
rss_mb, bounds.memory_high_mb, len(ordered), _mid_turn,
|
||||
"evictable session (%d cached, %d mid-turn, %d blocked on "
|
||||
"un-flushed persistence)%s",
|
||||
rss_mb, bounds.memory_high_mb, len(ordered), _mid_turn, _unflushed,
|
||||
(
|
||||
" — transcripts are not reaching the session DB "
|
||||
"(session persistence disabled or failing?); the memory "
|
||||
"valve cannot shed sessions until they persist."
|
||||
if _unflushed and not _mid_turn
|
||||
else " — memory will keep climbing until those turns finish."
|
||||
),
|
||||
)
|
||||
return 0
|
||||
|
||||
evicted_count = len(plan)
|
||||
logger.warning(
|
||||
"Agent cache pressure: anon RSS %dMB over budget %dMB — evicting "
|
||||
"%d LRU session(s): %s",
|
||||
rss_mb, bounds.memory_high_mb, len(plan),
|
||||
rss_mb, bounds.memory_high_mb, evicted_count,
|
||||
", ".join(key for key, _ in plan),
|
||||
)
|
||||
try:
|
||||
|
|
@ -23873,7 +23901,10 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
).start()
|
||||
except Exception:
|
||||
self._release_pressure_batch(plan)
|
||||
return len(plan)
|
||||
# NOTE: _release_pressure_batch drains `plan` in place (so the trim
|
||||
# runs with no lingering agent references) — len(plan) is 0 by the
|
||||
# time the daemon thread finishes, hence the pre-captured count.
|
||||
return evicted_count
|
||||
|
||||
def _release_pressure_batch(self, plan: List[tuple]) -> None:
|
||||
"""Release a pressure-evicted batch, then return the heap to the OS.
|
||||
|
|
@ -23883,12 +23914,20 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
memory, not to race N teardowns. The trailing ``malloc_trim`` is what
|
||||
turns "Python dropped the transcript" into "RSS actually fell" —
|
||||
without it glibc keeps the freed arenas and the cgroup never notices.
|
||||
|
||||
The plan is drained (``pop`` + ``del``) rather than iterated so that
|
||||
no local reference pins the evicted agents when ``gc.collect`` +
|
||||
``malloc_trim`` run — otherwise the trim frees almost nothing in this
|
||||
pass, the next tick re-reads a still-high RSS, and the valve
|
||||
over-evicts an extra batch of warm prompt caches every cycle.
|
||||
"""
|
||||
for key, agent in plan:
|
||||
while plan:
|
||||
key, agent = plan.pop(0) # FIFO — evict LRU-first order preserved
|
||||
try:
|
||||
self._commit_then_release_soft(agent, key)
|
||||
except Exception as _e:
|
||||
logger.debug("Pressure release failed for %s: %s", key, _e)
|
||||
del agent
|
||||
try:
|
||||
from hermes_cli.mem_trim import trim_memory
|
||||
|
||||
|
|
|
|||
|
|
@ -393,3 +393,120 @@ def _wait_for(predicate, timeout: float = 3.0) -> None:
|
|||
return
|
||||
_t.sleep(0.02)
|
||||
assert predicate(), "background release did not complete in time"
|
||||
|
||||
|
||||
class TestSalvageFollowups:
|
||||
"""Follow-up behaviors added while salvaging PR #80795."""
|
||||
|
||||
def test_config_read_failure_still_resolves_auto_budget(self, monkeypatch):
|
||||
"""A transient config-read failure must not permanently disable the
|
||||
pressure valve — the fallback resolves an empty config, whose absent
|
||||
section means memory_high_mb='auto', not None."""
|
||||
import gateway.run as gw_run
|
||||
from gateway.run import GatewayRunner
|
||||
|
||||
monkeypatch.setattr(
|
||||
gw_run, "_load_gateway_config",
|
||||
lambda: (_ for _ in ()).throw(OSError("transient")),
|
||||
)
|
||||
import gateway.agent_cache_pressure as acp
|
||||
|
||||
monkeypatch.setattr(acp, "_cgroup_limit_bytes", lambda: 8 * 1024**3)
|
||||
|
||||
runner = GatewayRunner.__new__(GatewayRunner)
|
||||
bounds = runner._agent_cache_bounds()
|
||||
assert bounds.memory_high_mb is not None, (
|
||||
"config-read failure fell back to a disabled valve — "
|
||||
"the #80764 protection must survive a flaky config read"
|
||||
)
|
||||
|
||||
def test_protect_recent_yaml_false_keeps_default(self):
|
||||
"""protect_recent: false (YAML-typo bool; False == 0) must keep the
|
||||
default MRU protection, not silently disable it."""
|
||||
bounds = resolve_agent_cache_bounds(
|
||||
{"agent": {"agent_cache": {"protect_recent": False}}}
|
||||
)
|
||||
assert bounds.protect_recent > 0
|
||||
|
||||
def test_release_batch_drains_plan_before_trim(self, monkeypatch):
|
||||
"""The plan list must be empty when trim_memory runs, so no local
|
||||
reference pins the evicted agents during gc.collect + malloc_trim
|
||||
(otherwise the in-pass trim frees nothing and the next tick
|
||||
over-evicts another batch)."""
|
||||
from collections import OrderedDict as _OD
|
||||
|
||||
from gateway.run import GatewayRunner
|
||||
|
||||
runner = GatewayRunner.__new__(GatewayRunner)
|
||||
released = []
|
||||
runner._commit_then_release_soft = lambda agent, key: released.append(key)
|
||||
|
||||
plan_len_at_trim = {}
|
||||
|
||||
import hermes_cli.mem_trim as mem_trim_mod
|
||||
|
||||
plan = [(f"s{i}", MagicMock()) for i in range(3)]
|
||||
|
||||
def fake_trim(force=False, reason=None):
|
||||
plan_len_at_trim["len"] = len(plan)
|
||||
return True
|
||||
|
||||
monkeypatch.setattr(mem_trim_mod, "trim_memory", fake_trim)
|
||||
|
||||
runner._release_pressure_batch(plan)
|
||||
|
||||
assert sorted(released) == ["s0", "s1", "s2"]
|
||||
assert plan_len_at_trim["len"] == 0, (
|
||||
"plan still held agent references when trim_memory ran"
|
||||
)
|
||||
|
||||
def test_soft_release_clears_db_flush_scan_prefix(self):
|
||||
"""_db_flush_scan_prefix shallow-copies the flushed transcript and is
|
||||
populated on exactly the agents the valve targets — leaving it pins
|
||||
every message dict the eviction claims to free."""
|
||||
from gateway.run import GatewayRunner
|
||||
|
||||
runner = GatewayRunner.__new__(GatewayRunner)
|
||||
agent = MagicMock()
|
||||
transcript = [{"role": "user", "content": "x" * 1024}]
|
||||
agent._session_messages = transcript
|
||||
agent._db_flush_scan_prefix = transcript[:]
|
||||
|
||||
runner._release_evicted_agent_soft(agent)
|
||||
|
||||
assert agent._session_messages == []
|
||||
assert agent._db_flush_scan_prefix is None
|
||||
|
||||
def test_no_evictable_warning_distinguishes_unflushed_persistence(self, monkeypatch, caplog):
|
||||
"""When everything is blocked on un-flushed persistence (e.g. the
|
||||
session DB never initialized), the warning must say so instead of
|
||||
blaming mid-turn agents."""
|
||||
import logging as _logging
|
||||
|
||||
from collections import OrderedDict as _OD
|
||||
|
||||
import gateway.agent_cache_pressure as acp
|
||||
from gateway.run import GatewayRunner
|
||||
|
||||
runner = GatewayRunner.__new__(GatewayRunner)
|
||||
runner._agent_cache = _OD()
|
||||
runner._agent_cache_lock = threading.Lock()
|
||||
runner._running_agents = {}
|
||||
runner._agent_cache_bounds_cache = AgentCacheBounds(
|
||||
memory_high_mb=1000, max_evictions_per_pass=8, protect_recent=0
|
||||
)
|
||||
monkeypatch.setattr(acp, "read_anon_rss_mb", lambda: 4000)
|
||||
|
||||
for i in range(3):
|
||||
agent = MagicMock()
|
||||
agent._session_messages = [{"role": "user", "content": "x"}]
|
||||
agent._last_flushed_db_idx = 0 # never flushed
|
||||
runner._agent_cache[f"s{i}"] = (agent, "sig")
|
||||
|
||||
with caplog.at_level(_logging.WARNING, logger="gateway.run"):
|
||||
evicted = runner._sweep_agent_cache_under_pressure()
|
||||
|
||||
assert evicted == 0
|
||||
joined = " ".join(r.getMessage() for r in caplog.records)
|
||||
assert "blocked on un-flushed persistence" in joined
|
||||
assert "3 blocked" in joined
|
||||
|
|
|
|||
Loading…
Reference in New Issue