diff --git a/exca/cachedict/__init__.py b/exca/cachedict/__init__.py index 47d330d9..0c9804bf 100644 --- a/exca/cachedict/__init__.py +++ b/exca/cachedict/__init__.py @@ -6,6 +6,7 @@ from .core import METADATA_TAG as METADATA_TAG from .core import CacheDict as CacheDict +from .core import observe_access as observe_access CacheDictWriter = CacheDict # deprecated: use CacheDict directly from exca.dumperloader import MEMMAP_ARRAY_FILE_MAX_CACHE as MEMMAP_ARRAY_FILE_MAX_CACHE diff --git a/exca/cachedict/core.py b/exca/cachedict/core.py index 6c8274ff..c9fc2626 100644 --- a/exca/cachedict/core.py +++ b/exca/cachedict/core.py @@ -30,6 +30,49 @@ logger = logging.getLogger(__name__) METADATA_TAG = "metadata=" +_access_observers: list[tp.Callable[[Path], None]] = [] +_observers_lock = threading.Lock() + + +@contextlib.contextmanager +def observe_access( + callback: tp.Callable[[Path], None], +) -> tp.Iterator[tp.Callable[[Path], None]]: + """Call *callback* with the folder of every ``CacheDict`` touched in scope. + + For the duration of the ``with`` block, *callback* receives the ``folder`` + of each :class:`CacheDict` as it is constructed or accessed (``keys``, + ``__getitem__``, ``__setitem__``, ``__contains__``). Observation is + process-wide and nested scopes each get their own notifications. + + Callbacks must be thread-safe (they may be invoked from cache reader + threads) and must not raise; exceptions are caught and logged at debug + level. Yields *callback* for ``with observe_access(cb):`` usage. + """ + with _observers_lock: + _access_observers.append(callback) + try: + yield callback + finally: + with _observers_lock: + try: + _access_observers.remove(callback) + except ValueError: + pass + + +def _notify_access(folder: Path | None) -> None: + """Notify active :func:`observe_access` callbacks of a touched *folder*.""" + if folder is None or not _access_observers: + return + with _observers_lock: + observers = list(_access_observers) + for callback in observers: + try: + callback(folder) + except Exception: # observation must never break caching + logger.debug("cache-access observer failed", exc_info=True) + @dataclasses.dataclass class DumpInfo: @@ -144,6 +187,7 @@ def __init__( if self.folder is not None: self._dumper = DumpContext(self.folder, permissions=self.permissions) self._local = threading.local() # per-thread write context, see _write_ctx + _notify_access(self.folder) def __repr__(self) -> str: name = self.__class__.__name__ @@ -181,6 +225,7 @@ def __len__(self) -> int: def keys(self) -> tp.Iterator[str]: """Returns the keys in the dictionary (triggers a cache folder reading if folder is not None)""" + _notify_access(self.folder) self._read_info_files() keys = set(self._ram_data) | set(self._key_info) return iter(keys) @@ -271,6 +316,7 @@ def items(self) -> tp.Iterator[tuple[str, X]]: yield key, self[key] def __getitem__(self, key: str) -> X: + _notify_access(self.folder) if self._keep_in_ram: if key in self._ram_data or self.folder is None: return self._ram_data[key] @@ -285,6 +331,37 @@ def __getitem__(self, key: str) -> X: self._ram_data[key] = loaded return loaded # type: ignore + def filepath(self, key: str) -> Path: + """Absolute path of the file (or directory) backing *key* on disk. + + Indexing returns the loaded value; this returns the path it was dumped + to instead (parquet/npy/pickle dumps, directory entries, ...). Triggers + a cache-folder read if *key* has not been loaded yet. + + Raises + ------ + RuntimeError + if this cache is RAM-only (``folder is None``). + KeyError + if *key* is not present in the cache. + ValueError + if *key*'s entry is not backed by a single file/directory (its + info record carries no ``filename``, e.g. an inline JSON value). + """ + if self.folder is None: + raise RuntimeError("filepath requires a folder-backed CacheDict") + if key not in self._key_info: + _ = self.keys() # reload keys from disk + if key not in self._key_info: + raise KeyError(key) + filename = self._key_info[key].content.get("filename") + if filename is None: + raise ValueError( + f"Entry {key!r} is not backed by a single file " + "(no 'filename' in its cache info)" + ) + return (self.folder / filename).resolve() + # Thread-local write context: each thread gets its own DumpContext # (and thus its own JSONL file), enabling concurrent writers. @property @@ -325,6 +402,7 @@ def writer(self) -> tp.Iterator["CacheDict[X]"]: yield self def __setitem__(self, key: str, value: X) -> None: + _notify_access(self.folder) if not isinstance(key, str): raise TypeError(f"Non-string keys are not allowed (got {key!r})") if self.folder is not None and self._write_ctx is None: @@ -368,6 +446,7 @@ def __delitem__(self, key: str) -> None: self._dumper.delete(info.content) def __contains__(self, key: str) -> bool: + _notify_access(self.folder) # in-memory cache if key in self._ram_data: return True diff --git a/exca/cachedict/test_cachedict.py b/exca/cachedict/test_cachedict.py index 2d0c23f7..f270968e 100644 --- a/exca/cachedict/test_cachedict.py +++ b/exca/cachedict/test_cachedict.py @@ -391,3 +391,72 @@ def test_orphaned_cleanup_file_deleted_concurrently(tmp_path: Path) -> None: keys = list(cache.keys()) assert keys == [] assert reader._fp.name not in cache._jsonl_readers + + +def test_filepath(tmp_path: Path) -> None: + cache: cd.CacheDict[np.ndarray] = cd.CacheDict(folder=tmp_path, keep_in_ram=False) + with cache.write(): + cache["arr"] = np.array([1, 2, 3]) + fp = cache.filepath("arr") + assert fp.is_absolute() + assert fp.exists() + assert tmp_path.resolve() in fp.parents + # a fresh view (empty _key_info) resolves by reloading keys from disk + cache2: cd.CacheDict[np.ndarray] = cd.CacheDict(folder=tmp_path) + assert cache2.filepath("arr") == fp + with pytest.raises(KeyError): + cache2.filepath("missing") + + +def test_filepath_ram_only_raises() -> None: + cache: cd.CacheDict[int] = cd.CacheDict(folder=None, keep_in_ram=True) + with pytest.raises(RuntimeError): + cache.filepath("x") + + +def test_filepath_inline_value_raises(tmp_path: Path) -> None: + cache: cd.CacheDict[tp.Any] = cd.CacheDict(folder=tmp_path, keep_in_ram=False) + # simulate an entry whose info record carries no 'filename' (inline value) + cache._key_info["x"] = cd.DumpInfo( + jsonl=tmp_path / "w-info.jsonl", byte_range=(0, 0), content={"#type": "Json"} + ) + with pytest.raises(ValueError, match="not backed by a single file"): + cache.filepath("x") + + +def test_observe_access(tmp_path: Path) -> None: + folder = tmp_path / "sub" + seen: list[Path] = [] + with cd.observe_access(seen.append): + cache: cd.CacheDict[np.ndarray] = cd.CacheDict(folder=folder) # __init__ + with cache.write(): + cache["a"] = np.array([1]) # __setitem__ + assert "a" in cache # __contains__ + assert set(cache.keys()) == {"a"} # keys + _ = cache["a"] # __getitem__ + # every notification is for this cache's folder, and construction was seen + assert seen # got at least the __init__ notification + assert set(seen) == {folder} + # a RAM-only cache (folder is None) never notifies + seen.clear() + with cd.observe_access(seen.append): + cd.CacheDict(folder=None, keep_in_ram=True) + assert seen == [] + # after every scope exits, the registry is empty again + assert cd._access_observers == [] + + +def test_observe_access_nested_and_error_safe(tmp_path: Path) -> None: + outer: list[Path] = [] + inner: list[Path] = [] + + def boom(_folder: Path) -> None: + raise RuntimeError("observer errors must never break caching") + + with cd.observe_access(outer.append): + with cd.observe_access(boom): # raising observer is swallowed + with cd.observe_access(inner.append): + cd.CacheDict(folder=tmp_path / "x") + assert (tmp_path / "x") in outer + assert (tmp_path / "x") in inner + assert cd._access_observers == []