diff --git a/agents/DEVELOPING.md b/agents/DEVELOPING.md index 90196797..ad841af8 100644 --- a/agents/DEVELOPING.md +++ b/agents/DEVELOPING.md @@ -95,7 +95,7 @@ just want to *use* the library (decorator, config, backends), see **Storage layout (`storage/`):** - `base.py` — `OperationContext(ABC)` (the new root after PR #604) exposes the `_operation_context(key, *, intent=Intent.WRITE)` context-manager hook — chain via `super()._operation_context(key, intent=intent)`; the `intent: Intent` kw-only param (PR #601) is a `StrEnum` with `WRITE` (the default — takes the exclusive lock) and `READ` (a **no-op** today: the lock mixins short-circuit it and acquire nothing; reserved for a future reader-writer/shared lock, so it must never guard a read-modify-write). `CacheStack._operation_context` enters its non-`stack[0]` members with `READ`, so a transfer into a stack only really locks `stack[0]` where the write lands. `KeyManagement(OperationContext)` adds the abstract `list`/`_evict`/`_contains` plus concrete `evict`/`contains`/`expand`/`shrink`/`_normalize_key` (`expand` raises `KeyError` for prefixes shorter than 4 chars). `StorageBackend(KeyManagement)` adds `put`/`get`. Domain ABCs `ValueStorage`, `CallStorage` (the latter also exposes `transform(func=None)` — re-saves every entry through *func*; entries whose `to_lookup_key()` changes are saved under the new key and the old one is evicted, unchanged ones are re-saved idempotently. `func=None` applies the identity, which is the form to use after a hash-function change. The analogous `Cache.redigest` does its own loop because it must update `values` too). **Bridge mixins** `ValueMixin`/`CallMixin` implement `save`/`load`[/`query`] on top of `put`/`get`. Also: `SaveError`, `AmbiguousDigestError`, `_resolve_prefix` (used by both base and `Sql`). `OperationContext`, `KeyManagement`, `Intent`, `StorageBackend`, `ValueStorage`/`ValueMixin`, `CallStorage`/`CallMixin` are all re-exported from `fleche.storage`. - `file.py` — `FileStorage` base for disk-backed backends (also re-exported from `fleche.storage` for subclassing). Locking is via the `filelock` package: writes use `filelock.FileLock(lock_path, timeout=lock_timeout)` directly; reads go through `_file_read_lock_with_fallback`, which on lock timeout logs a warning and proceeds **without** the lock (a missing/torn file then surfaces as `KeyError`). `root` is resolved (`expanduser`+`absolute`+`resolve`) in `__post_init__`. Subclasses implement `_to_file`/`_from_file`; compression and signing live in `pickle_file.py`, not here. -- `memory.py`, `void.py`, `pickle_file.py` (+`PickleFileBackend.with_pickle`/`with_cloudpickle`/`with_dill`; `compress_all`/`decompress_all` migration helpers; gzip auto-detected by `\x1f\x8b` magic on read), `bagofholding_file.py` (+`rebag(version_validator="none")` re-saves every bag through a chosen validator, useful when older bags would fail strict version checks; `_from_file` opens the bag with `H5Bag(path, _skip_load=True).load(...)` so the file is opened exactly once per read instead of twice — PR #616; optional `prefix_length: int | None = 2` on `BagOfHoldingH5FileBackend` multiplexes keys sharing the first `prefix_length` chars into one `root/{prefix}.h5` file as sibling groups named by the full key, via bagofholding 0.1.12's `file.h5/group` path addressing — single-level, fixed-length split only; the default `2` multi-bags into up to 256 files, `prefix_length=0` restores the one-file-per-key layout, and `prefix_length=None` infers the length from the files already in `root` (empty root → the default `2`) — PR #746 closes #556. The PR #748 add-ons: `refix(n)` copies every entry into a new `prefix_length` layout and returns the re-fixed storage, leaving the receiver invariant (target must be an explicit int, `0` = per-key; fails fast on unreadable entries; resumable — entries already in the target layout are skipped), `__post_init__` raises `ValueError` when the configured `prefix_length` doesn't match the layout of files already in `root` (skippable via the init-only `check_consistency=False`, which requires an explicit `prefix_length` and makes the instance blind to all other layouts' files), and the `consolidate(root, n)` classmethod repairs a mixed-layout root by refixing every other prefix length it finds to `n`), `sql.py` — concrete backends (each exposes `Value*` and/or `Call*` classes; `sql.py` only has `Sql` for calls). `MemoryBackend.put`/`get` deep-copy values, so mutating a stored object after retrieval does not affect later reads. `Sql.query` always pushes name/module/version/code_digest/result and argument filters down to SQL (arguments are compared as `digest(value)` strings via `JOIN`s on the `arguments` table). Metadata filters are pushed down only when *every* filter value is a simple scalar (`str`/`bool`/`int`/`float`) via JSON-extract `as_*` casts; any `None` or complex (e.g. list) value disables metadata pushdown only — name/argument filters still apply at SQL level — and the post-load `meta_matches` check runs on every yielded result regardless. `Sql` does not inherit from `CallMixin` — it implements `CallStorage.save` directly, folding the existing-row check into a private `_persist_call` helper that runs in one transaction (no separate `contains`+`evict` round-trip), with `PerKeyLockMixin` keeping concurrent saves serialised (see `regression/test_sql_concurrent_save.py`). The companion private helper `_fetch_call` materialises a `DigestedCall` from a row plus its argument/metadata child rows; `Sql` does not expose any `put`/`get` methods (it isn't a `StorageBackend` subclass). `_coerce_sqlite_url` accepts a bare path (treated as sqlite, parent dir auto-created), a `sqlite:` URL, or any other SQLAlchemy URL (e.g. `postgresql://`, `mysql+pymysql://`) verbatim. `:memory:` is special-cased. SQLite-only PRAGMAs are enabled via a `connect` event listener (gated on `dialect.name == "sqlite"`): `foreign_keys=ON` always, plus `journal_mode=WAL` for file-backed DBs (skipped for `:memory:`). The schema is three tables: `calls`, `arguments` (one row per arg, `UNIQUE(call_key, name)`, ordered by `position`), `metadata` (one JSON blob per metadata namespace, `UNIQUE(call_key, name)`). `arguments.call_key`/`metadata.call_key` are FKs with `ondelete="CASCADE"`, so `_evict` is a single `session.execute(delete(CallModel).where(...))` (SQLAlchemy DSL, no raw SQL or ORM materialisation) and lets the DB clean up child rows (SQLite needs the `foreign_keys=ON` PRAGMA the dialect listener installs; Postgres/MySQL enforce FK cascades natively). MySQL/MariaDB needs explicit `VARCHAR(255)` for indexable name columns; other dialects get unbounded `TEXT` via `String().with_variant(...)`. `Sql` also accepts `check_same_thread=False` only for sqlite drivers (skipped for Postgres/MySQL where the flag would raise at connect). +- `memory.py`, `void.py`, `pickle_file.py` (+`PickleFileBackend.with_pickle`/`with_cloudpickle`/`with_dill`; `compress_all`/`decompress_all` migration helpers; gzip auto-detected by `\x1f\x8b` magic on read), `bagofholding_file.py` (+`rebag(version_validator="none")` re-saves every bag through a chosen validator, useful when older bags would fail strict version checks; `_from_file` opens the bag with `H5Bag(path, _skip_load=True).load(...)` so the file is opened exactly once per read instead of twice — PR #616; optional `prefix_length: int | None = 2` on `BagOfHoldingH5FileBackend` multiplexes keys sharing the first `prefix_length` chars into one `root/{prefix}.h5` file as sibling groups named by the full key, via bagofholding 0.1.12's `file.h5/group` path addressing — single-level, fixed-length split only; the default `2` multi-bags into up to 256 files, `prefix_length=0` restores the one-file-per-key layout, and `prefix_length=None` infers the length from the files already in `root` (empty root → the default `2`) — PR #746 closes #556. The PR #748 add-ons: `refix(n)` copies every entry into a new `prefix_length` layout and returns the re-fixed storage, leaving the receiver invariant (target must be an explicit int, `0` = per-key; fails fast on unreadable entries; resumable — entries already in the target layout are skipped), `__post_init__` raises `ValueError` when the configured `prefix_length` doesn't match the layout of files already in `root` (skippable via the init-only `check_consistency=False`, which requires an explicit `prefix_length` and makes the instance blind to all other layouts' files), and the `consolidate(root, n)` classmethod repairs a mixed-layout root by refixing every other prefix length it finds to `n`. The #625 multi-bag perf fix adds a **process-wide read-handle cache** (`_BagHandleCache` / module-level `_bag_handles`): `contains`/`list` reuse open read-only `h5py.File` handles instead of paying a full HDF5 open per call, indexed in a `WeakValueDictionary` by absolute path with a strong-ref MRU of `_MAX_OPEN_BAGS` keeping them alive between operations, validated against an `(inode, mtime_ns, size)` stat signature on every acquisition. Handles are opened with `locking=False` so they don't block other processes' writes; the flip side is that HDF5 rejects *any* same-process open whose mode or locking flags differ from a live handle's, so every path that lets H5Bag open the file itself (`_to_file`, `_from_file`, `rebag`) or writes via `h5py.File(..., "a")` (`_evict`) must run inside `_bag_writer(key)`, which closes the cached handle and holds the per-bag in-process lock across the operation; readers hold the same lock for the duration of their handle use (`_bag_reader`). The cache is keyed by path process-wide — not per-instance — precisely because a handle cached by one storage instance must be closable by any other instance addressing the same file. `evict` still pays the write open/close (structural: a durable per-key delete in a shared HDF5 file)), `sql.py` — concrete backends (each exposes `Value*` and/or `Call*` classes; `sql.py` only has `Sql` for calls). `MemoryBackend.put`/`get` deep-copy values, so mutating a stored object after retrieval does not affect later reads. `Sql.query` always pushes name/module/version/code_digest/result and argument filters down to SQL (arguments are compared as `digest(value)` strings via `JOIN`s on the `arguments` table). Metadata filters are pushed down only when *every* filter value is a simple scalar (`str`/`bool`/`int`/`float`) via JSON-extract `as_*` casts; any `None` or complex (e.g. list) value disables metadata pushdown only — name/argument filters still apply at SQL level — and the post-load `meta_matches` check runs on every yielded result regardless. `Sql` does not inherit from `CallMixin` — it implements `CallStorage.save` directly, folding the existing-row check into a private `_persist_call` helper that runs in one transaction (no separate `contains`+`evict` round-trip), with `PerKeyLockMixin` keeping concurrent saves serialised (see `regression/test_sql_concurrent_save.py`). The companion private helper `_fetch_call` materialises a `DigestedCall` from a row plus its argument/metadata child rows; `Sql` does not expose any `put`/`get` methods (it isn't a `StorageBackend` subclass). `_coerce_sqlite_url` accepts a bare path (treated as sqlite, parent dir auto-created), a `sqlite:` URL, or any other SQLAlchemy URL (e.g. `postgresql://`, `mysql+pymysql://`) verbatim. `:memory:` is special-cased. SQLite-only PRAGMAs are enabled via a `connect` event listener (gated on `dialect.name == "sqlite"`): `foreign_keys=ON` always, plus `journal_mode=WAL` for file-backed DBs (skipped for `:memory:`). The schema is three tables: `calls`, `arguments` (one row per arg, `UNIQUE(call_key, name)`, ordered by `position`), `metadata` (one JSON blob per metadata namespace, `UNIQUE(call_key, name)`). `arguments.call_key`/`metadata.call_key` are FKs with `ondelete="CASCADE"`, so `_evict` is a single `session.execute(delete(CallModel).where(...))` (SQLAlchemy DSL, no raw SQL or ORM materialisation) and lets the DB clean up child rows (SQLite needs the `foreign_keys=ON` PRAGMA the dialect listener installs; Postgres/MySQL enforce FK cascades natively). MySQL/MariaDB needs explicit `VARCHAR(255)` for indexable name columns; other dialects get unbounded `TEXT` via `String().with_variant(...)`. `Sql` also accepts `check_same_thread=False` only for sqlite drivers (skipped for Postgres/MySQL where the flag would raise at connect). - `destructuring.py` — `DestructuringMixin` for recursive value splitting + `Digested` ABC with markers `DigestedIterable` (lists/tuples), `DigestedDict`, and `DigestedFields` (shared base reconstructing instances via `object.__new__` + `__setattr__`, bypassing `__init__`/`__post_init__`, so `init=False`/`InitVar`/frozen/slots fields round-trip) with two concrete subclasses `DigestedDataclass` (stdlib dataclasses) and `DigestedAttrs` (`attrs`-decorated classes). All preserve digest equivalence via `__digest__`. `DestructuringMixin` is a `ValueStorage` subclass — operates at the `save`/`load` layer, not `put`/`get`; compose **above** `ValueMixin` in the MRO. `remaining_depth` (default `1`) controls how deep structures are split across keys. `child_digests(key)→set[Digest]` returns the immediate digest references of a stored entry (raw, pre-`mend`); `count_reuses()` tallies how often each key is referenced as a sub-component (useful for GC-style audits). The `HasChildDigests` `runtime_checkable` Protocol declares the `child_digests` shape so `Cache.gc()` can opt-in to transitive walks via plain `isinstance` — any future value storage that exposes `child_digests` satisfies it without explicit registration. NamedTuples are deliberately **not** destructured (`_is_trojan_tuple` guard). `register_destructurer(pred, fn)` (re-exported from `fleche.storage`) appends a new entry to the module-level `_DESTRUCTURERS` list — first match wins, so registering a handler for a brand-new container type is safe; ordering matters if you want to override list/dict/dataclass/attrs. - `thread_safe.py` — `SerializingMixin(OperationContext)` (single `_PicklableRLock`), `PerKeyLockMixin(OperationContext)` (striped locks via a module-level `_per_instance_locks: WeakKeyDictionary`; per-instance `WeakValueDictionary[key, RLock]` — so the storage instance must be **hashable**, which all the frozen-dataclass concrete classes are). Both mixins override `_operation_context(key, *, intent=Intent.WRITE)`; chain via `super()._operation_context(key, intent=intent)`. Both short-circuit `intent=Intent.READ` to a plain `yield` (no lock acquired) — the no-op shared-lock placeholder. Inheriting from `OperationContext` directly (rather than `KeyManagement`) is what lets them attach at *either* the storage layer or the cache layer — `BaseCache(OperationContext)` since PR #622 closed #569. `_PicklableLock`/`_PicklableRLock` survive pickle round-trip with state not preserved — in-process only, NOT inter-process synchronisation. diff --git a/src/fleche/storage/bagofholding_file.py b/src/fleche/storage/bagofholding_file.py index dea6af0d..8cc519bd 100644 --- a/src/fleche/storage/bagofholding_file.py +++ b/src/fleche/storage/bagofholding_file.py @@ -1,3 +1,8 @@ +import contextlib +import os +import threading +import weakref +from collections import OrderedDict from dataclasses import dataclass, field, replace, InitVar from pathlib import Path from typing import Any, Literal, Iterable @@ -56,6 +61,137 @@ def _observed_prefix_lengths(root: Path) -> set[int]: return observed +# Cap on read-only bag-file handles kept open between operations. Handles are +# shared process-wide (keyed by absolute path), so this also bounds the +# process's open-fd contribution regardless of how many storages exist. +_MAX_OPEN_BAGS = 32 + + +def _open_readonly(path: Path) -> "h5py.File": + """Open *path* read-only without OS-level HDF5 file locking. + + Cached handles stay open between operations; with default locking each + would hold a shared HDF5 lock that makes every write from *another + process* fail for as long as the handle lives. Writes are coordinated by + ``filelock`` sidecar locks instead, and files rewritten behind our back + are caught by the stat signature in :class:`_BagHandleCache`. + """ + try: + return h5py.File(path, "r", locking=False) + except (TypeError, ValueError): + # h5py < 3.5 (no ``locking`` kwarg) or HDF5 < 1.12.1 (no support): + # fall back to default locking — correct, just briefly blocks writers. + return h5py.File(path, "r") + + +class _BagHandleCache: + """Process-wide cache of open read-only h5py handles for multi-bag files. + + ``_files`` is a weak-value index of the open handles; ``_recent`` keeps + strong references to the :data:`_MAX_OPEN_BAGS` most recently used ones so + they survive between operations (a weak-only entry would be collected — + and the file closed — the moment the operation using it returns). A + handle evicted from ``_recent`` is *not* closed eagerly: an operation in + another thread may still be reading from it, so only the strong reference + is dropped and the interpreter closes the file once the last user lets go. + + The cache is keyed by absolute path and shared by all storage instances + rather than kept per-instance: HDF5 refuses to open a file for writing + while *any* read handle on it is open in the same process (independent of + OS-level file locking), so a handle cached by one instance must be + closable by every other instance addressing the same file. + + Every access — read or write — to a bag file must happen while holding + that file's :meth:`lock`. Readers hold it for the duration of their use + of the handle, writers across invalidate-open-write-close, so a writer + can never close the cached handle under a reader mid-use, and no cached + handle can be alive during a same-process write open. Staleness from + *other* processes is caught by re-validating a ``(inode, mtime, size)`` + stat signature on every acquisition. + """ + + def __init__(self) -> None: + self._pid = os.getpid() + self._meta_lock = threading.Lock() + # Per-bag locks self-prune like PerKeyLockMixin's: alive while any + # thread holds one (or is inside the with block), recreated otherwise. + self._bag_locks: weakref.WeakValueDictionary[str, threading.RLock] = ( + weakref.WeakValueDictionary() + ) + self._files: weakref.WeakValueDictionary[str, "h5py.File"] = ( + weakref.WeakValueDictionary() + ) + self._recent: OrderedDict[str, "h5py.File"] = OrderedDict() + self._signatures: dict[str, tuple[int, int, int]] = {} + + def lock(self, path: Path) -> threading.RLock: + """The in-process lock guarding all access to the bag file at *path*.""" + with self._meta_lock: + if self._pid != os.getpid(): + # Forked child: inherited handles belong to the parent — drop + # every reference without closing eagerly (they are read-only, + # so the close-on-collect in the child is harmless). + self._pid = os.getpid() + self._bag_locks = weakref.WeakValueDictionary() + self._files = weakref.WeakValueDictionary() + self._recent = OrderedDict() + self._signatures = {} + # Hold a strong reference so the lock is not collected between + # creation and return — WeakValueDictionary only stores a weak ref. + key = str(path) + lock = self._bag_locks.get(key) + if lock is None: + lock = threading.RLock() + self._bag_locks[key] = lock + return lock + + def acquire(self, path: Path) -> "h5py.File | None": + """A validated open read-only handle for *path*, or ``None`` when the + file does not exist. The caller must hold :meth:`lock` for *path*. + + Raises: + OSError: if the file exists but cannot be opened. + """ + key = str(path) + try: + st = os.stat(key) + except OSError: + self.invalidate(path) + return None + signature = (st.st_ino, st.st_mtime_ns, st.st_size) + with self._meta_lock: + f = self._files.get(key) + if f is not None and self._signatures.get(key) != signature: + # Rewritten by another process (or storage instance): safe to + # close because we hold the bag lock, so no reader is mid-use. + f.close() + f = None + if f is None: + f = _open_readonly(path) + with self._meta_lock: + self._files[key] = f + self._signatures[key] = signature + self._recent[key] = f + self._recent.move_to_end(key) + while len(self._recent) > _MAX_OPEN_BAGS: + self._recent.popitem(last=False) + return f + + def invalidate(self, path: Path) -> None: + """Close and drop any cached handle for *path*. The caller must hold + :meth:`lock` for *path*.""" + key = str(path) + with self._meta_lock: + f = self._files.pop(key, None) + self._recent.pop(key, None) + self._signatures.pop(key, None) + if f is not None: + f.close() + + +_bag_handles = _BagHandleCache() + + @dataclass(frozen=True) class BagOfHoldingH5FileBackend(FileStorage): version_validator: VersionValidator | None = None @@ -205,26 +341,65 @@ def consolidate( ) return cls(root, prefix_length=prefix_length, **kwargs) + @contextlib.contextmanager + def _bag_reader(self, file_path: Path): + """Yield a cached read-only handle for *file_path*, or ``None`` when + the file does not exist, holding the in-process bag lock throughout so + no writer can close the handle mid-use. On ``OSError`` — an + unreadable file, or a handle broken by an external rewrite — the + cached handle is dropped before the error propagates, so the next + operation starts from a fresh open.""" + with _bag_handles.lock(file_path): + try: + yield _bag_handles.acquire(file_path) + except OSError: + _bag_handles.invalidate(file_path) + raise + + @contextlib.contextmanager + def _bag_writer(self, key: str): + """Hold *key*'s in-process bag lock across an operation that opens the + bag file itself, closing any cached read handle first. HDF5 refuses a + same-process write open while any read handle is open, and refuses + *any* same-process open whose locking flags differ from an existing + handle's — so this guards not only writes but also :class:`H5Bag` + reads, which use the default flags rather than the cache's + ``locking=False``. No-op in per-key mode, where nothing is cached.""" + if self.prefix_length == 0: + yield + return + file_path = self._bag_file(key) + with _bag_handles.lock(file_path): + _bag_handles.invalidate(file_path) + yield + def _to_file(self, value: Any, path: Path) -> None: - try: - H5Bag.save(value, path) - except (ValueError, TypeError): # h5py choked on something, pass it along - raise SaveError(value) from None + # In multi-bag mode `path` is the composite `{prefix}.h5/{key}`, so + # the key is its final component (in per-key mode _bag_writer ignores it). + with self._bag_writer(path.name): + try: + H5Bag.save(value, path) + except (ValueError, TypeError): # h5py choked on something, pass it along + raise SaveError(value) from None def _from_file(self, path: Path) -> Any: - try: - # _skip_load=True skips the constructor's _load_existing_bag_info() call, - # which would otherwise open and close the file just to read bag metadata - # before load() opens it a second time to read the actual payload. - bag = H5Bag(path, _skip_load=True) - if self.version_validator is not None: - return bag.load(version_validator=self.version_validator) - return bag.load() - except (FileNotFoundError, KeyError): - raise KeyError(path) from None - except OSError as e: - logger.error("Corrupt file present in cache at path %s: %s", path, e, exc_info=True) - raise KeyError(path) from e + # H5Bag opens the file itself, so any cached read handle must be + # closed first and the bag lock held across the load — otherwise the + # flag-mismatched open would fail (and read as a corrupt file). + with self._bag_writer(path.name): + try: + # _skip_load=True skips the constructor's _load_existing_bag_info() call, + # which would otherwise open and close the file just to read bag metadata + # before load() opens it a second time to read the actual payload. + bag = H5Bag(path, _skip_load=True) + if self.version_validator is not None: + return bag.load(version_validator=self.version_validator) + return bag.load() + except (FileNotFoundError, KeyError): + raise KeyError(path) from None + except OSError as e: + logger.error("Corrupt file present in cache at path %s: %s", path, e, exc_info=True) + raise KeyError(path) from e def _bag_file(self, key: str) -> Path: """The HDF5 file backing `key` in multi-bag mode: ``root/{prefix}.h5``.""" @@ -246,12 +421,9 @@ def _lock_path(self, key: str) -> Path: def _contains(self, key: Digest) -> bool: if self.prefix_length == 0: return super()._contains(key) - file_path = self._bag_file(key) - if not file_path.is_file(): - return False try: - with h5py.File(file_path, "r") as f: - return key in f + with self._bag_reader(self._bag_file(key)) as f: + return f is not None and key in f except OSError: return False @@ -263,13 +435,17 @@ def _evict(self, key: Digest) -> None: with filelock.FileLock(lock_path, timeout=self.lock_timeout): if not file_path.is_file(): return - with h5py.File(file_path, "a") as f: - if key in f: - del f[key] - remaining = len(f) - if remaining == 0: - file_path.unlink(missing_ok=True) - lock_path.unlink(missing_ok=True) + # The unlink stays inside _bag_writer so a reader in another + # thread cannot re-open (and re-cache) the file between the + # write-close and its removal. + with self._bag_writer(key): + with h5py.File(file_path, "a") as f: + if key in f: + del f[key] + remaining = len(f) + if remaining == 0: + file_path.unlink(missing_ok=True) + lock_path.unlink(missing_ok=True) def list(self) -> Iterable[Digest]: # Only files of exactly this instance's prefix length are considered, @@ -291,8 +467,9 @@ def list(self) -> Iterable[Digest]: ): continue try: - with h5py.File(p, "r") as f: - keys.extend(Digest(name) for name in f.keys()) + with self._bag_reader(p) as f: + if f is not None: + keys.extend(Digest(name) for name in f.keys()) except OSError as e: logger.error("Corrupt file present in cache at path %s: %s", p, e, exc_info=True) return keys @@ -307,8 +484,11 @@ def rebag(self, version_validator: VersionValidator = "none") -> None: path = self._path(key) with self._operation_context(key): try: - value = H5Bag(path, _skip_load=True).load(version_validator=version_validator) - H5Bag.save(value, path) + # both the load and the save open the file with H5Bag's + # own (default) locking flags — see _bag_writer + with self._bag_writer(key): + value = H5Bag(path, _skip_load=True).load(version_validator=version_validator) + H5Bag.save(value, path) except OSError as e: logger.warning("Failed to rebag %s: %s", key, e) diff --git a/tests/unit/storage/test_bagofholding_file.py b/tests/unit/storage/test_bagofholding_file.py index 11aaf5fe..29ff67f2 100644 --- a/tests/unit/storage/test_bagofholding_file.py +++ b/tests/unit/storage/test_bagofholding_file.py @@ -568,6 +568,156 @@ def test_init_check_ignores_unrelated_files(tmp_path): BagOfHoldingH5FileBackend(tmp_path, prefix_length=2) +def _count_backend_h5_opens(monkeypatch): + """Count h5py.File opens made by the backend itself (H5Bag's internal + opens go through bagofholding's own import and are not counted).""" + import fleche.storage.bagofholding_file as boh_mod + + opens = [] + real_file = boh_mod.h5py.File + + def counting_file(*args, **kwargs): + opens.append(args) + return real_file(*args, **kwargs) + + monkeypatch.setattr(boh_mod.h5py, "File", counting_file) + return opens + + +def test_multi_bag_contains_reuses_cached_handle(tmp_path, monkeypatch): + pytest.importorskip("bagofholding") + s = BagOfHoldingH5FileBackend(tmp_path, prefix_length=2) + key = _digest_like("a") + s.put("value", key) + + opens = _count_backend_h5_opens(monkeypatch) + for _ in range(3): + assert s.contains(key) + assert len(opens) == 1 + + +def test_multi_bag_list_reuses_cached_handle(tmp_path, monkeypatch): + pytest.importorskip("bagofholding") + s = BagOfHoldingH5FileBackend(tmp_path, prefix_length=2) + key = _digest_like("a") + s.put("value", key) + + opens = _count_backend_h5_opens(monkeypatch) + assert s.contains(key) + assert set(s.list()) == {key} + assert set(s.list()) == {key} + assert len(opens) == 1 + + +def test_multi_bag_get_with_warm_handle_cache(tmp_path): + """H5Bag opens files with HDF5's default locking flags, which conflict + with the differently-flagged cached read handle contains() leaves behind + — get() must close it first instead of failing the load.""" + pytest.importorskip("bagofholding") + s = BagOfHoldingH5FileBackend(tmp_path, prefix_length=2) + key = _digest_like("a") + s.put("value", key) + + assert s.contains(key) + assert s.get(key) == "value" + + +def test_multi_bag_contains_sees_sibling_writes(tmp_path): + pytest.importorskip("bagofholding") + s = BagOfHoldingH5FileBackend(tmp_path, prefix_length=2) + key1 = Digest("ab" + "1" * 62) + key2 = Digest("ab" + "2" * 62) + + s.put("first", key1) + assert s.contains(key1) + assert not s.contains(key2) + s.put("second", key2) # rewrites the bag the cached handle points at + + assert s.contains(key2) + + +def test_multi_bag_evict_with_warm_handle_cache(tmp_path): + pytest.importorskip("bagofholding") + s = BagOfHoldingH5FileBackend(tmp_path, prefix_length=2) + key1 = Digest("ab" + "1" * 62) + key2 = Digest("ab" + "2" * 62) + s.put("first", key1) + s.put("second", key2) + + assert s.contains(key1) # warm the handle cache + s.evict(key1) + assert not s.contains(key1) + assert s.contains(key2) + s.evict(key2) + assert not (tmp_path / "ab.h5").exists() # no cached handle kept it alive + + +def test_multi_bag_write_from_second_instance_with_cached_reader(tmp_path): + """A cached read handle must not block same-process writes from another + (non-equal) storage instance on the same root, and the write must be + visible through the first instance afterwards.""" + pytest.importorskip("bagofholding") + a = BagOfHoldingH5FileBackend(tmp_path, prefix_length=2) + b = BagOfHoldingH5FileBackend(tmp_path, prefix_length=2, version_validator="none") + key1 = Digest("ab" + "1" * 62) + key2 = Digest("ab" + "2" * 62) + + a.put("first", key1) + assert a.contains(key1) + b.put("second", key2) + + assert a.contains(key2) + assert b.get(key2) == "second" + + +def test_multi_bag_contains_sees_external_writes(tmp_path): + """Writers in other processes must neither be blocked by a cached read + handle (HDF5 file locking is disabled on cached handles) nor have their + writes masked by it (the stat signature forces a reopen).""" + pytest.importorskip("bagofholding") + import subprocess + import sys + + s = BagOfHoldingH5FileBackend(tmp_path, prefix_length=2) + key1 = Digest("ab" + "1" * 62) + key2 = Digest("ab" + "2" * 62) + s.put("first", key1) + assert s.contains(key1) # warm the handle cache + assert not s.contains(key2) + + code = ( + "import sys, h5py\n" + "with h5py.File(sys.argv[1], 'a') as f:\n" + " f.create_group(sys.argv[2])\n" + ) + subprocess.run( + [sys.executable, "-c", code, str(tmp_path / "ab.h5"), str(key2)], + check=True, + capture_output=True, + ) + + assert s.contains(key2) + + +def test_open_handle_cache_is_bounded(tmp_path): + pytest.importorskip("bagofholding") + import fleche.storage.bagofholding_file as boh_mod + + s = BagOfHoldingH5FileBackend(tmp_path, prefix_length=2) + hexes = "0123456789abcdef" + prefixes = [a + b for a in hexes for b in hexes] + keys = [ + Digest(prefix.ljust(64, "0")) + for prefix in prefixes[: boh_mod._MAX_OPEN_BAGS + 4] + ] + for i, key in enumerate(keys): + s.put(str(i), key) + assert s.contains(key) + + assert len(boh_mod._bag_handles._recent) <= boh_mod._MAX_OPEN_BAGS + assert all(s.contains(key) for key in keys) + + def test_multi_bag_rebag(tmp_path): pytest.importorskip("bagofholding") s = BagOfHoldingH5FileBackend(tmp_path, prefix_length=2)