mirror/nanobot
1
0
Fork 0
mirror of https://github.com/HKUDS/nanobot.git synced 2026-08-06 17:38:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(config): drop thread synchronization

Browse Source
This commit is contained in:
chengyongru 2026-07-14 11:34:18 +08:00
parent f19efcd990
commit f5371a6c5a
2 changed files with 35 additions and 87 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

88
nanobot/config/repository.py
View File

@ -7,7 +7,6 @@ import json
import os import os
import re import re
import stat import stat
import threading
import uuid import uuid
from collections.abc import Callable from collections.abc import Callable
from contextlib import suppress from contextlib import suppress
@ -18,15 +17,11 @@ from typing import Any
from loguru import logger from loguru import logger
from pydantic import BaseModel from pydantic import BaseModel
from nanobot.config.schema import Config, _resolve_tool_config_refs from nanobot.config.schema import Config
ConfigMutator = Callable[[Config], None] ConfigMutator = Callable[[Config], None]
_ENV_REF_PATTERN = re.compile(r"\$\{([A-Za-z_][A-Za-z0-9_]*)\}") _ENV_REF_PATTERN = re.compile(r"\$\{([A-Za-z_][A-Za-z0-9_]*)\}")
_schema_refs_ready = False
_schema_refs_lock = threading.Lock()
_path_locks_guard = threading.Lock()
_path_locks: dict[Path, threading.RLock] = {}
class ConfigConflictError(RuntimeError): class ConfigConflictError(RuntimeError):
@ -60,22 +55,6 @@ class ConfigCommit:
changed_paths: frozenset[str] changed_paths: frozenset[str]
def _ensure_schema_refs() -> None:
global _schema_refs_ready
if _schema_refs_ready:
return
with _schema_refs_lock:
if not _schema_refs_ready:
_resolve_tool_config_refs()
_schema_refs_ready = True
def _lock_for_path(path: Path) -> threading.RLock:
key = path.expanduser().resolve(strict=False)
with _path_locks_guard:
return _path_locks.setdefault(key, threading.RLock())
def _revision_for_bytes(raw: bytes | None) -> str: def _revision_for_bytes(raw: bytes | None) -> str:
if raw is None: if raw is None:
return "missing" return "missing"
@ -92,7 +71,6 @@ def _config_data(config: Config) -> dict[str, Any]:
def _validate_config_data(data: dict[str, Any], path: Path) -> Config: def _validate_config_data(data: dict[str, Any], path: Path) -> Config:
_ensure_schema_refs()
try: try:
return Config.model_validate(data) return Config.model_validate(data)
except ValueError as exc: except ValueError as exc:
@ -100,7 +78,6 @@ def _validate_config_data(data: dict[str, Any], path: Path) -> Config:
def _read_snapshot(path: Path) -> PersistedConfigSnapshot: def _read_snapshot(path: Path) -> PersistedConfigSnapshot:
_ensure_schema_refs()
if not path.exists(): if not path.exists():
return PersistedConfigSnapshot(Config(), path, "missing") return PersistedConfigSnapshot(Config(), path, "missing")
@ -158,18 +135,15 @@ def _changed_paths(before: Any, after: Any, prefix: str = "") -> set[str]:
class FileConfigRepository: class FileConfigRepository:
"""Read and atomically update one configuration file. """Read and atomically update one configuration file.
The repository does not cache. Every read returns a new validated snapshot, The repository does not cache. Every read returns a new validated snapshot.
while updates for the same path are serialized within this process.
""" """
def __init__(self, path: str | Path): def __init__(self, path: str | Path):
self.path = Path(path).expanduser().resolve(strict=False) self.path = Path(path).expanduser().resolve(strict=False)
self._lock = _lock_for_path(self.path)
def load_raw(self) -> PersistedConfigSnapshot: def load_raw(self) -> PersistedConfigSnapshot:
"""Load the persisted form without resolving secret references.""" """Load the persisted form without resolving secret references."""
with self._lock: return _read_snapshot(self.path)
return _read_snapshot(self.path)
def load_effective(self) -> EffectiveConfigSnapshot: def load_effective(self) -> EffectiveConfigSnapshot:
"""Load an isolated runtime snapshot with ``${VAR}`` references resolved.""" """Load an isolated runtime snapshot with ``${VAR}`` references resolved."""
@ -184,17 +158,16 @@ class FileConfigRepository:
expected_revision: str | None = None, expected_revision: str | None = None,
) -> PersistedConfigSnapshot: ) -> PersistedConfigSnapshot:
"""Atomically save a complete config, optionally rejecting stale writes.""" """Atomically save a complete config, optionally rejecting stale writes."""
with self._lock: current = _read_snapshot(self.path)
current = _read_snapshot(self.path) if expected_revision is not None and current.revision != expected_revision:
if expected_revision is not None and current.revision != expected_revision: raise ConfigConflictError(
raise ConfigConflictError( f"Config changed since revision {expected_revision}; "
f"Config changed since revision {expected_revision}; " f"current revision is {current.revision}"
f"current revision is {current.revision}" )
) data = _config_data(config)
data = _config_data(config) _validate_config_data(data, self.path)
_validate_config_data(data, self.path) _write_config_atomic(self.path, data)
_write_config_atomic(self.path, data) return _read_snapshot(self.path)
return _read_snapshot(self.path)
def update( def update(
self, self,
@ -203,26 +176,25 @@ class FileConfigRepository:
expected_revision: str | None = None, expected_revision: str | None = None,
) -> ConfigCommit: ) -> ConfigCommit:
"""Atomically apply a mutation to the latest persisted config.""" """Atomically apply a mutation to the latest persisted config."""
with self._lock: before = _read_snapshot(self.path)
before = _read_snapshot(self.path) if expected_revision is not None and before.revision != expected_revision:
if expected_revision is not None and before.revision != expected_revision: raise ConfigConflictError(
raise ConfigConflictError( f"Config changed since revision {expected_revision}; "
f"Config changed since revision {expected_revision}; " f"current revision is {before.revision}"
f"current revision is {before.revision}" )
)
before_data = _config_data(before.config) before_data = _config_data(before.config)
draft = before.config.model_copy(deep=True) draft = before.config.model_copy(deep=True)
mutator(draft) mutator(draft)
after_data = _config_data(draft) after_data = _config_data(draft)
changed = frozenset(_changed_paths(before_data, after_data)) changed = frozenset(_changed_paths(before_data, after_data))
if not changed: if not changed:
return ConfigCommit(before, before, changed) return ConfigCommit(before, before, changed)
_validate_config_data(after_data, self.path) _validate_config_data(after_data, self.path)
_write_config_atomic(self.path, after_data) _write_config_atomic(self.path, after_data)
after = _read_snapshot(self.path) after = _read_snapshot(self.path)
return ConfigCommit(before, after, changed) return ConfigCommit(before, after, changed)
def resolve_config_env_vars(config: Config) -> Config: def resolve_config_env_vars(config: Config) -> Config:

34
tests/config/test_repository.py
View File

@ -2,7 +2,6 @@ import json
import os import os
import socket import socket
import stat import stat
import threading
from pathlib import Path from pathlib import Path
from unittest.mock import patch from unittest.mock import patch
@ -84,38 +83,15 @@ def test_update_rejects_stale_expected_revision(tmp_path: Path) -> None:
assert repository.load_raw().config.api.port == 9001 assert repository.load_raw().config.api.port == 9001
def test_repositories_for_same_path_serialize_updates(tmp_path: Path) -> None: def test_repositories_for_same_path_read_latest_config(tmp_path: Path) -> None:
path = tmp_path / "config.json" path = tmp_path / "config.json"
first = FileConfigRepository(path) first = FileConfigRepository(path)
second = FileConfigRepository(path) second = FileConfigRepository(path)
first_mutator_entered = threading.Event()
release_first = threading.Event()
second_mutator_entered = threading.Event()
def update_first() -> None: first.update(lambda config: setattr(config.api, "port", 9001))
def mutate(config): second.update(
first_mutator_entered.set() lambda config: setattr(config.agents.defaults, "timezone", "Asia/Shanghai")
assert release_first.wait(timeout=2) )
config.api.port = 9001
first.update(mutate)
def update_second() -> None:
def mutate(config):
second_mutator_entered.set()
config.agents.defaults.timezone = "Asia/Shanghai"
second.update(mutate)
first_thread = threading.Thread(target=update_first)
second_thread = threading.Thread(target=update_second)
first_thread.start()
assert first_mutator_entered.wait(timeout=2)
second_thread.start()
assert not second_mutator_entered.wait(timeout=0.1)
release_first.set()
first_thread.join(timeout=2)
second_thread.join(timeout=2)
config = first.load_raw().config config = first.load_raw().config
assert config.api.port == 9001 assert config.api.port == 9001