fix(config): serve cached config reads without taking _CONFIG_LOCK
A cache hit in `_load_config_impl` costs microseconds, but it was served
from inside `_CONFIG_LOCK` — which `save_config()` holds across an atomic
YAML write. Measured on a clean checkout, driving the real functions
against a temp HERMES_HOME:
cache-hit read, uncontended median 0.0237ms
the SAME cached read while another
thread holds _CONFIG_LOCK 10010.2ms
On a gateway this lands on the event loop. `invoke_hook` calls
`_resolve_hook_callback_timeout`, which reads config, and a gateway fires
hooks on every inbound message — so one background config write stalls
every message for the full duration of that write. The same probe showed
the hook path doing 100 config reads across 100 hook invocations.
Two changes:
1. `_load_config_impl` gets a lock-free fast path for cache hits. The lock
never protected the cache dict: CPython dict get/setitem are atomic under
the GIL, and the cached tuple is replaced wholesale rather than mutated
in place, so a reader observes either the complete old tuple or the
complete new one. The lock's real job is serializing the rebuild
(parse + merge + expand) and the writers. Worst case on a race is a
redundant rebuild, which the locked path re-checks and collapses. The
existing `_load_config_cache_sig()` helper is reused, so the fast and
locked paths cannot drift on freshness.
2. `_resolve_hook_callback_timeout` is memoized on that same signature. The
value only changes when config.yaml does; every other call is a dict
lookup. Validation and clamping move unchanged into
`_resolve_hook_callback_timeout_uncached`.
After: the blocked read returns in 0.0ms and the hook path does 1 config
read per 100 invocations.
Tests (tests/hermes_cli/test_config_lock_free_cache_hit.py, 11 tests) drive
the real functions against a temp HERMES_HOME — no mocks of the code under
test. They cover the blocked-read case for both the readonly and deepcopy
entry points, that the fast path still sees a changed file, that
`load_config()` still returns an isolated object, a 4-reader + 1-writer
concurrency arm asserting no torn observation, and the hook memo's
freshness plus its clamp/fallback/zero-disables contract.
RED/GREEN on this base, impl reverted via git stash:
without the change : 8 failed, 3 passed (blocked read: 30.0s)
with the change : 11 passed
Neighbours green: tests/hermes_cli/test_config.py, test_plugins.py,
test_config_loader_e2e.py, test_managed_scope_loaders.py,
test_read_raw_config_readonly.py — 266 passed, 4 skipped.
(cherry picked from commit b787427bdaef81d8b114194779fe1de3b4efe7bd)
This commit is contained in:
Kyzcreig
kshitij
@@ -2285,6 +2285,37 @@ def _merge_managed_overlay(expanded: Dict[str, Any]) -> Tuple[Dict[str, Any], An
|
||||
|
||||
|
||||
def _load_config_impl(*, want_deepcopy: bool) -> Dict[str, Any]:
|
||||
# Lock-free fast path for cache hits.
|
||||
#
|
||||
# A cache hit costs ~0.024ms, but it used to sit behind `_CONFIG_LOCK` —
|
||||
# which `save_config()` holds across an atomic YAML write. Measured: the
|
||||
# same cached read takes 10010ms when another thread holds the lock. On a
|
||||
# gateway this lands on the event loop, because the per-message hook path
|
||||
# (`invoke_hook` -> `_resolve_hook_callback_timeout`) reads config, so one
|
||||
# background config write stalls every inbound message for the duration.
|
||||
#
|
||||
# The lock never protected the cache dict itself: CPython dict get/setitem
|
||||
# are atomic under the GIL, and the cached tuple is replaced wholesale
|
||||
# rather than mutated in place, so a reader sees either the complete old
|
||||
# tuple or the complete new one. The lock's real job is serializing the
|
||||
# rebuild (parse + merge + expand) and the writers. Worst case on a race
|
||||
# is a redundant rebuild, which the locked path below re-checks and
|
||||
# collapses.
|
||||
try:
|
||||
config_path = get_config_path()
|
||||
path_key = str(config_path)
|
||||
cached = _LOAD_CONFIG_CACHE.get(path_key)
|
||||
if cached is not None:
|
||||
_, fast_sig = _load_config_cache_sig(config_path)
|
||||
if fast_sig is not None and cached[:8] == fast_sig:
|
||||
env_snapshot = cached[9] if len(cached) > 9 else {}
|
||||
if all(_env_ref_lookup(k) == v for k, v in env_snapshot.items()):
|
||||
return copy.deepcopy(cached[8]) if want_deepcopy else cached[8]
|
||||
except Exception:
|
||||
# Any surprise here falls through to the locked path, which is the
|
||||
# original fully-defensive implementation.
|
||||
pass
|
||||
|
||||
with _CONFIG_LOCK:
|
||||
ensure_hermes_home()
|
||||
config_path = get_config_path()
|
||||
|
||||
@@ -1131,9 +1131,54 @@ for _name, _method in list(vars(PluginContext).items()):
|
||||
del _name, _method
|
||||
|
||||
|
||||
_HOOK_TIMEOUT_CACHE_LOCK = threading.Lock()
|
||||
# Sentinel distinct from every real signature AND from None (which
|
||||
# _load_config_cache_sig returns when there is no config file to key on).
|
||||
_UNRESOLVED_HOOK_TIMEOUT_SIG: Any = object()
|
||||
_HOOK_TIMEOUT_CACHE: Dict[str, Any] = {
|
||||
"sig": _UNRESOLVED_HOOK_TIMEOUT_SIG,
|
||||
"value": None,
|
||||
}
|
||||
|
||||
|
||||
def _reset_hook_callback_timeout_cache() -> None:
|
||||
"""Drop the memoized hook-callback timeout. For tests and config reloads."""
|
||||
with _HOOK_TIMEOUT_CACHE_LOCK:
|
||||
_HOOK_TIMEOUT_CACHE["sig"] = _UNRESOLVED_HOOK_TIMEOUT_SIG
|
||||
_HOOK_TIMEOUT_CACHE["value"] = None
|
||||
|
||||
|
||||
def _resolve_hook_callback_timeout() -> float:
|
||||
"""Effective hook-callback timeout from ``plugins.hook_callback_timeout`` (default 30s; ``<= 0``
|
||||
disables the threaded path; clamped to ``_MAX_HOOK_CALLBACK_TIMEOUT_SECS``)."""
|
||||
disables the threaded path; clamped to ``_MAX_HOOK_CALLBACK_TIMEOUT_SECS``).
|
||||
|
||||
Memoized on the config file's cache signature. ``invoke_hook`` calls this
|
||||
once per hook invocation, and a gateway fires hooks on every inbound
|
||||
message — so this was a full config read per message, on the event loop.
|
||||
The value only changes when config.yaml does; every other call is a dict
|
||||
lookup.
|
||||
"""
|
||||
try:
|
||||
from hermes_cli.config import _load_config_cache_sig, get_config_path
|
||||
_, sig = _load_config_cache_sig(get_config_path())
|
||||
except Exception:
|
||||
sig = None
|
||||
|
||||
if sig is not None and _HOOK_TIMEOUT_CACHE["sig"] == sig:
|
||||
value = _HOOK_TIMEOUT_CACHE["value"]
|
||||
if isinstance(value, float):
|
||||
return value
|
||||
|
||||
resolved = _resolve_hook_callback_timeout_uncached()
|
||||
if sig is not None:
|
||||
with _HOOK_TIMEOUT_CACHE_LOCK:
|
||||
_HOOK_TIMEOUT_CACHE["sig"] = sig
|
||||
_HOOK_TIMEOUT_CACHE["value"] = resolved
|
||||
return resolved
|
||||
|
||||
|
||||
def _resolve_hook_callback_timeout_uncached() -> float:
|
||||
"""Read + validate ``plugins.hook_callback_timeout``; see the memoized wrapper above."""
|
||||
default = _HOOK_CALLBACK_TIMEOUT_SECS
|
||||
try:
|
||||
from hermes_cli.config import load_config_readonly
|
||||
|
||||
281
tests/hermes_cli/test_config_lock_free_cache_hit.py
Normal file
281
tests/hermes_cli/test_config_lock_free_cache_hit.py
Normal file
@@ -0,0 +1,281 @@
|
||||
"""A cached config read must not block behind a config WRITER.
|
||||
|
||||
``_load_config_impl`` served cache hits from inside ``_CONFIG_LOCK``, which
|
||||
``save_config()`` holds across an atomic YAML write. A cache hit itself costs
|
||||
~0.024us-scale work, but measured on a clean tree the same cached read took
|
||||
**10010ms** while another thread held the lock.
|
||||
|
||||
On a gateway that lands on the event loop: the per-message hook path
|
||||
(``invoke_hook`` -> ``_resolve_hook_callback_timeout``) reads config, so one
|
||||
background config write stalls every inbound message for its whole duration.
|
||||
|
||||
These tests drive the real functions against a temp HERMES_HOME -- no mocks of
|
||||
the thing under test, no source reading.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import threading
|
||||
import time
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
# Generous: the point is 10s-vs-instant, not a tight timing assertion.
|
||||
MAX_BLOCKED_READ_SECS = 2.0
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def config_home(tmp_path, monkeypatch):
|
||||
home = tmp_path / "hermes-home"
|
||||
home.mkdir()
|
||||
lines = ["plugins:", " hook_callback_timeout: 30", "agent:", " max_turns: 500"]
|
||||
for i in range(100):
|
||||
lines += [f"section_{i}:", f" key_a: value_{i}"]
|
||||
(home / "config.yaml").write_text("\n".join(lines) + "\n", encoding="utf-8")
|
||||
monkeypatch.setenv("HERMES_HOME", str(home))
|
||||
|
||||
from hermes_cli import config as cfgmod
|
||||
|
||||
cfgmod._LOAD_CONFIG_CACHE.clear()
|
||||
yield home
|
||||
cfgmod._LOAD_CONFIG_CACHE.clear()
|
||||
|
||||
|
||||
def _bump_mtime(path):
|
||||
st = path.stat()
|
||||
os.utime(path, ns=(st.st_atime_ns, st.st_mtime_ns + 1_000_000))
|
||||
|
||||
|
||||
def test_cached_read_completes_while_another_thread_holds_the_config_lock(config_home):
|
||||
from hermes_cli import config as cfgmod
|
||||
|
||||
cfgmod.load_config_readonly() # prime the cache
|
||||
|
||||
holding = threading.Event()
|
||||
release = threading.Event()
|
||||
|
||||
def _hold():
|
||||
with cfgmod._CONFIG_LOCK:
|
||||
holding.set()
|
||||
release.wait(timeout=30)
|
||||
|
||||
thread = threading.Thread(target=_hold, daemon=True)
|
||||
thread.start()
|
||||
try:
|
||||
assert holding.wait(timeout=10), "lock holder never started"
|
||||
started = time.perf_counter()
|
||||
cfg = cfgmod.load_config_readonly()
|
||||
elapsed = time.perf_counter() - started
|
||||
finally:
|
||||
release.set()
|
||||
thread.join(timeout=10)
|
||||
|
||||
assert isinstance(cfg, dict) and cfg
|
||||
assert elapsed < MAX_BLOCKED_READ_SECS, (
|
||||
f"a cached load_config_readonly() blocked {elapsed:.1f}s behind a held "
|
||||
"_CONFIG_LOCK; cache hits must not serialize against config writers"
|
||||
)
|
||||
|
||||
|
||||
def test_deepcopy_read_also_completes_under_a_held_lock(config_home):
|
||||
from hermes_cli import config as cfgmod
|
||||
|
||||
cfgmod.load_config()
|
||||
|
||||
holding = threading.Event()
|
||||
release = threading.Event()
|
||||
|
||||
def _hold():
|
||||
with cfgmod._CONFIG_LOCK:
|
||||
holding.set()
|
||||
release.wait(timeout=30)
|
||||
|
||||
thread = threading.Thread(target=_hold, daemon=True)
|
||||
thread.start()
|
||||
try:
|
||||
assert holding.wait(timeout=10)
|
||||
started = time.perf_counter()
|
||||
cfg = cfgmod.load_config()
|
||||
elapsed = time.perf_counter() - started
|
||||
finally:
|
||||
release.set()
|
||||
thread.join(timeout=10)
|
||||
|
||||
assert cfg["agent"]["max_turns"] == 500
|
||||
assert elapsed < MAX_BLOCKED_READ_SECS
|
||||
|
||||
|
||||
def test_fast_path_still_sees_a_changed_config_file(config_home):
|
||||
"""Freshness is not sacrificed for the lock-free read."""
|
||||
from hermes_cli import config as cfgmod
|
||||
|
||||
assert cfgmod.load_config_readonly()["agent"]["max_turns"] == 500
|
||||
|
||||
path = config_home / "config.yaml"
|
||||
path.write_text(
|
||||
path.read_text(encoding="utf-8").replace("max_turns: 500", "max_turns: 123"),
|
||||
encoding="utf-8",
|
||||
)
|
||||
_bump_mtime(path)
|
||||
|
||||
assert cfgmod.load_config_readonly()["agent"]["max_turns"] == 123
|
||||
|
||||
|
||||
def test_load_config_still_returns_an_isolated_object(config_home):
|
||||
"""The deepcopy contract must survive the fast path."""
|
||||
from hermes_cli import config as cfgmod
|
||||
|
||||
first = cfgmod.load_config()
|
||||
first["agent"]["max_turns"] = -1
|
||||
assert cfgmod.load_config()["agent"]["max_turns"] == 500
|
||||
assert cfgmod.load_config_readonly()["agent"]["max_turns"] == 500
|
||||
|
||||
|
||||
def test_concurrent_readers_and_a_writer_never_see_a_torn_config(config_home):
|
||||
"""The fast path reads a tuple that is replaced wholesale, never mutated."""
|
||||
from hermes_cli import config as cfgmod
|
||||
|
||||
cfgmod.load_config_readonly()
|
||||
stop = threading.Event()
|
||||
errors: list[BaseException] = []
|
||||
seen: list[int] = []
|
||||
|
||||
def _writer():
|
||||
try:
|
||||
while not stop.is_set():
|
||||
cfg = cfgmod.load_config()
|
||||
cfg.setdefault("agent", {})["max_turns"] = 500
|
||||
cfgmod.save_config(cfg)
|
||||
except BaseException as exc: # noqa: BLE001
|
||||
errors.append(exc)
|
||||
|
||||
def _reader():
|
||||
try:
|
||||
for _ in range(300):
|
||||
cfg = cfgmod.load_config_readonly()
|
||||
# Every observation must be a complete, well-formed config.
|
||||
assert isinstance(cfg, dict)
|
||||
assert isinstance(cfg.get("agent"), dict)
|
||||
seen.append(int(cfg["agent"]["max_turns"]))
|
||||
except BaseException as exc: # noqa: BLE001
|
||||
errors.append(exc)
|
||||
|
||||
writer = threading.Thread(target=_writer, daemon=True)
|
||||
readers = [threading.Thread(target=_reader, daemon=True) for _ in range(4)]
|
||||
writer.start()
|
||||
for r in readers:
|
||||
r.start()
|
||||
for r in readers:
|
||||
r.join(timeout=60)
|
||||
stop.set()
|
||||
writer.join(timeout=30)
|
||||
|
||||
assert not errors, f"torn/failed read under concurrency: {errors[0]!r}"
|
||||
assert seen, "readers observed nothing"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Hook-callback timeout memoization.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_hook_timeout_does_not_read_config_on_every_invocation(config_home):
|
||||
from hermes_cli import config as cfgmod
|
||||
from hermes_cli import plugins as pluginsmod
|
||||
|
||||
pluginsmod._reset_hook_callback_timeout_cache()
|
||||
assert pluginsmod._resolve_hook_callback_timeout() == 30.0
|
||||
|
||||
calls = {"n": 0}
|
||||
real = cfgmod.load_config_readonly
|
||||
|
||||
def counting():
|
||||
calls["n"] += 1
|
||||
return real()
|
||||
|
||||
cfgmod.load_config_readonly = counting # type: ignore[assignment]
|
||||
try:
|
||||
for _ in range(100):
|
||||
pluginsmod._resolve_hook_callback_timeout()
|
||||
finally:
|
||||
cfgmod.load_config_readonly = real # type: ignore[assignment]
|
||||
|
||||
assert calls["n"] == 0, (
|
||||
f"hook-timeout resolution read the config {calls['n']}x across 100 hook "
|
||||
"invocations; a gateway fires hooks per inbound message, so this is a "
|
||||
"per-message config read on the event loop"
|
||||
)
|
||||
|
||||
|
||||
def test_hook_timeout_picks_up_a_changed_config(config_home):
|
||||
from hermes_cli import plugins as pluginsmod
|
||||
|
||||
pluginsmod._reset_hook_callback_timeout_cache()
|
||||
assert pluginsmod._resolve_hook_callback_timeout() == 30.0
|
||||
|
||||
path = config_home / "config.yaml"
|
||||
path.write_text(
|
||||
path.read_text(encoding="utf-8").replace(
|
||||
"hook_callback_timeout: 30", "hook_callback_timeout: 45"
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
_bump_mtime(path)
|
||||
|
||||
assert pluginsmod._resolve_hook_callback_timeout() == 45.0
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"raw,expected_attr",
|
||||
[
|
||||
("9999", "_MAX_HOOK_CALLBACK_TIMEOUT_SECS"),
|
||||
("-5", "_HOOK_CALLBACK_TIMEOUT_SECS"),
|
||||
("'not-a-number'", "_HOOK_CALLBACK_TIMEOUT_SECS"),
|
||||
],
|
||||
)
|
||||
def test_hook_timeout_validation_survives_memoization(config_home, raw, expected_attr):
|
||||
"""Clamping and the fallback must not be lost to the cache."""
|
||||
import re
|
||||
|
||||
from hermes_cli import plugins as pluginsmod
|
||||
|
||||
path = config_home / "config.yaml"
|
||||
path.write_text(
|
||||
re.sub(
|
||||
r" hook_callback_timeout: .*",
|
||||
f" hook_callback_timeout: {raw}",
|
||||
path.read_text(encoding="utf-8"),
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
_bump_mtime(path)
|
||||
pluginsmod._reset_hook_callback_timeout_cache()
|
||||
|
||||
assert pluginsmod._resolve_hook_callback_timeout() == getattr(
|
||||
pluginsmod, expected_attr
|
||||
)
|
||||
|
||||
|
||||
def test_hook_timeout_zero_disables_and_is_cached(config_home):
|
||||
"""``<= 0`` means 'no threaded timeout' -- a falsy value the memo must
|
||||
still return rather than treating as 'unresolved'."""
|
||||
import re
|
||||
|
||||
from hermes_cli import plugins as pluginsmod
|
||||
|
||||
path = config_home / "config.yaml"
|
||||
path.write_text(
|
||||
re.sub(
|
||||
r" hook_callback_timeout: .*",
|
||||
" hook_callback_timeout: 0",
|
||||
path.read_text(encoding="utf-8"),
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
_bump_mtime(path)
|
||||
pluginsmod._reset_hook_callback_timeout_cache()
|
||||
|
||||
assert pluginsmod._resolve_hook_callback_timeout() == 0.0
|
||||
assert pluginsmod._resolve_hook_callback_timeout() == 0.0
|
||||
Reference in New Issue
Block a user