From 0b4cba59807ec4fe890191ea303c4336358749e4 Mon Sep 17 00:00:00 2001 From: Derrick Chambers Date: Sat, 25 Jul 2026 13:28:58 +0200 Subject: [PATCH 1/3] ENH: synchronize registries, units, and catalogs for free-threading Replaces the heavyweight synchronization in #763 with one lock per piece of shared mutable state: - _FiberIOManager gets one instance RLock held across a whole load_plugins() call (including the loader), so no caller can see a multi-version format half registered. Cached lookups become invalidatable snapshots (frozenset/tuple) and copy/pickle drops and recreates the lock. - Namespace registration and lazy attachment each get one module/class level RLock; plugin loading and third-party namespace construction stay off the registry lock, and a double check lets a second thread reuse the attached instance. - units.py gets one _UNIT_LOCK around every helper that touches the mutable pint registry; the registry is built exactly once. - PatchCatalog serializes its revision counter and the caches keyed on it; file reads, patch resolution, and directory syncs stay outside the lock. - Shared data becomes caller owned: read-only cached coord arrays, a FrozenDict for coord_shapes, and a copied frame from Spool.get_contents(). Adds _locked and _reinit_after_fork helpers to utils.misc. --- dascore/core/coordmanager.py | 6 +- dascore/core/coords.py | 6 +- dascore/core/patch.py | 4 +- dascore/core/spool.py | 20 +- dascore/io/core.py | 137 ++++++++++---- dascore/io/index/catalog.py | 230 +++++++++++++---------- dascore/units.py | 91 +++++---- dascore/utils/misc.py | 41 ++++ dascore/utils/namespace.py | 79 +++++--- tests/test_core/test_coord_segmented.py | 5 + tests/test_core/test_coordmanager.py | 7 + tests/test_core/test_coords.py | 6 + tests/test_core/test_spool.py | 6 + tests/test_io/test_index/test_catalog.py | 47 +++++ tests/test_io/test_io_core.py | 99 +++++++++- tests/test_units.py | 42 +++++ tests/test_utils/test_misc.py | 38 ++++ tests/test_utils/test_namespace.py | 49 +++++ 18 files changed, 711 insertions(+), 202 deletions(-) diff --git a/dascore/core/coordmanager.py b/dascore/core/coordmanager.py index 7f12fbbeb..6a4c303d0 100644 --- a/dascore/core/coordmanager.py +++ b/dascore/core/coordmanager.py @@ -969,9 +969,9 @@ def decimate(self, **kwargs) -> tuple[Self, tuple[slice, ...]]: @property @cached_method - def coord_shapes(self) -> dict[str, tuple[int, ...]]: - """Get a dict of {coord_name: shape}.""" - return {i: v.shape for i, v in self.coord_map.items()} + def coord_shapes(self) -> FrozenDict[str, tuple[int, ...]]: + """Get an immutable mapping of {coord_name: shape}.""" + return FrozenDict({i: v.shape for i, v in self.coord_map.items()}) @property @cached_method diff --git a/dascore/core/coords.py b/dascore/core/coords.py index 123fe7cad..a425fe368 100644 --- a/dascore/core/coords.py +++ b/dascore/core/coords.py @@ -1501,7 +1501,8 @@ def update_limits(self, min=None, max=None, step=None, **kwargs) -> Self: def values(self) -> ArrayLike: """Return the values of the coordinate as an array.""" if len(self) == 1: - return np.asarray([self.start]) + # Cached, so it must be read-only like the other branch. + return array(np.asarray([self.start])) # note: linspace works better for floats that might have slightly # uneven spacing. It ensures the length of the output array is robust # to small deviations in spacing. However, this doesnt work for datetimes. @@ -2014,7 +2015,8 @@ def segment_count(self) -> int: def _segment_offsets(self) -> np.ndarray: """Return the starting sample index of each segment.""" lens = [len(x) for x in self.segments] - return np.cumsum([0, *lens[:-1]]) + # Cached and shared by callers, so hand back a read-only array. + return array(np.cumsum([0, *lens[:-1]])) @property @cached_method diff --git a/dascore/core/patch.py b/dascore/core/patch.py index 27ae737bc..35552c07a 100644 --- a/dascore/core/patch.py +++ b/dascore/core/patch.py @@ -239,8 +239,8 @@ def ndim(self) -> int: return len(self.coords.dims) @property - def coord_shapes(self) -> dict[str, tuple[int, ...]]: - """Return a dict of {coordinate: (shape, ...)}.""" + def coord_shapes(self) -> Mapping[str, tuple[int, ...]]: + """Return an immutable mapping of {coordinate: (shape, ...)}.""" return self.coords.coord_shapes @property diff --git a/dascore/core/spool.py b/dascore/core/spool.py index 742fdf1e5..1eff802dc 100644 --- a/dascore/core/spool.py +++ b/dascore/core/spool.py @@ -48,6 +48,18 @@ T = TypeVar("T") +def _copy_public_dataframe(frame: pd.DataFrame) -> pd.DataFrame: + """ + Return a caller-owned view of an internally cached dataframe. + + With copy-on-write (always on in pandas 3) a shallow copy already + detaches on the first write; without it the blocks must be copied. + """ + if int(pd.__version__.split(".", maxsplit=1)[0]) >= 3: + return frame.copy(deep=False) + return frame.copy(deep=not pd.options.mode.copy_on_write) + + class BaseSpool(NamespaceOwner, abc.ABC): """Spool Abstract Base Class (ABC) for defining Spool interface.""" @@ -260,6 +272,12 @@ def get_contents(self) -> pd.DataFrame: """ Get a dataframe of the spool contents. + Notes + ----- + Each call returns a caller-owned dataframe; mutating it never + changes the spool. Use ``frame.copy(deep=True)`` when an eager + block copy is needed. + Examples -------- >>> import dascore as dc @@ -479,7 +497,7 @@ def _df(self) -> pd.DataFrame: @compose_docstring(doc=BaseSpool.get_contents.__doc__) def get_contents(self) -> pd.DataFrame: """{doc}.""" - return self._df + return _copy_public_dataframe(self._df) def __len__(self): # counting pushes to SQL (or the cold live registry); the flat diff --git a/dascore/io/core.py b/dascore/io/core.py index a551973d3..fafcd139c 100644 --- a/dascore/io/core.py +++ b/dascore/io/core.py @@ -9,9 +9,10 @@ import warnings from collections import defaultdict from collections.abc import Generator, Mapping -from functools import cache, cached_property, wraps +from functools import cached_property, wraps from numbers import Integral from pathlib import Path +from threading import RLock from typing import Any, Literal, NotRequired, TypedDict, get_type_hints import numpy as np @@ -43,7 +44,14 @@ ) from dascore.utils.io import IOResourceManager, get_handle_from_resource from dascore.utils.mapping import FrozenDict -from dascore.utils.misc import _iter_filesystem, cached_method, iterate, warn_or_raise +from dascore.utils.misc import ( + _iter_filesystem, + _locked, + _reinit_after_fork, + cached_method, + iterate, + warn_or_raise, +) from dascore.utils.paths import coerce_to_local_path, coerce_to_upath, is_local_path from dascore.utils.plugins import get_entry_point_loaders from dascore.utils.progress import track @@ -376,13 +384,33 @@ class _FiberIOManager: def __init__(self, entry_point: str): self._entry_point = entry_point + # One lock guards all mutable state below; it is held for the whole + # of load_plugins so no caller can observe a half-loaded format. + self._lock = RLock() self._loaded_eps: set[str] = set() + # Formats whose load attempt finished, successfully or not. The + # outcome lives in _format_version/_failed_formats. + self._loaded_formats: set[str] = set() self._failed_formats: set[str] = set() self._format_version = defaultdict(dict) self._extension_list = defaultdict(list) # This is a dict of {input_type: (fiberio_name, version)} self._fiber_io_by_input_type = defaultdict(set) self._fiber_io_name_ver = set() + # Snapshots derived from the registry; cleared when it changes. + self._lookup_cache: dict[tuple, frozenset | tuple] = {} + + def __getstate__(self) -> dict: + """Return copy/pickle state without the process-local lock.""" + with self._lock: + state = dict(self.__dict__) + state.pop("_lock", None) + return state + + def __setstate__(self, state: dict) -> None: + """Restore state with a fresh process-local lock.""" + self.__dict__.update(state) + self._lock = RLock() @cached_property def _eps(self): @@ -393,38 +421,46 @@ def _eps(self): return pd.Series(get_entry_point_loaders("dascore.fiber_io")) @cached_property - def known_formats(self): + @_locked("_lock") + def known_formats(self) -> frozenset[str]: """Return names of known formats.""" formats = [name.split("__", maxsplit=1)[0] for name in self._eps.index] - return set(formats) | set(self._format_version) + return frozenset(formats) | frozenset(self._format_version) @property - def unloaded_formats(self): - """Return names of known formats.""" + @_locked("_lock") + def unloaded_formats(self) -> list[str]: + """Return names of known formats which have not been loaded.""" loaded_or_failed = set(self._format_version) | self._failed_formats return sorted(self.known_formats - loaded_or_failed) - @cached_method - def _get_fiber_io_by_input_type(self, input_type) -> set[FiberIO]: + @_locked("_lock") + def _get_fiber_io_by_input_type(self, input_type) -> frozenset[FiberIO]: """Get a set of FiberIO instances that meet input type.""" - if input_type not in self._fiber_io_by_input_type: - out = set() - for input_set in self._fiber_io_by_input_type.values(): - out |= input_set - else: - out = self._fiber_io_by_input_type[input_type] - return out + key = ("input_type", input_type) + if (cached := self._lookup_cache.get(key)) is None: + if input_type not in self._fiber_io_by_input_type: + out = set() + for input_set in self._fiber_io_by_input_type.values(): + out |= input_set + else: + out = self._fiber_io_by_input_type[input_type] + cached = self._lookup_cache[key] = frozenset(out) + return cached - @cache - def _get_prioritized_list(self, input_type="file"): + @_locked("_lock") + def _get_prioritized_list(self, input_type="file") -> tuple[FiberIO, ...]: """Yield a prioritized list of fiber_ios.""" # must load all plugins before getting list self.load_plugins() + key = ("prioritized", input_type) + if (cached := self._lookup_cache.get(key)) is not None: + return cached priority_fiber_ios = [] second_class_fiber_ios = [] for format_name in self.known_formats: - unsorted = self._format_version[format_name] - if not unsorted: + # Use get; indexing the defaultdict would register empty formats. + if not (unsorted := self._format_version.get(format_name)): continue keys = sorted(unsorted, reverse=True) fiber_ios = [unsorted[key] for key in keys] @@ -436,28 +472,40 @@ def _get_prioritized_list(self, input_type="file"): valid_fiberio_by_type = self._get_fiber_io_by_input_type(input_type) out = tuple(x for x in maybe_ios if x in valid_fiberio_by_type) # And return fiberIOs that much the input type. + self._lookup_cache[key] = out return out - @cached_method def load_plugins(self, format: str | None = None): """Load plugin for specific format or ensure all formats are loaded.""" - if format is not None and format in self._format_version: - return # already loaded - if not (unloaded := self.unloaded_formats): + # A format only lands in _loaded_formats once every one of its entry + # points is registered, so this fast path (and the lock below) keep + # multi-version formats from being seen half loaded. + if format is not None and format in self._loaded_formats: + return + with self._lock: + if format is not None and format in self._format_version: + self._loaded_formats.add(format) + return # already loaded + if not (unloaded := self.unloaded_formats): + return + formats = {format} if format is not None else unloaded + # Load one, or all, formats. Plugin imports deliberately run + # under the lock: tracking in-flight formats instead would need + # a claim/wait graph for a step which happens once per format. + # The cost is that a thread importing a module which defines a + # FiberIO waits here until any in-progress load finishes. + for form in formats: + entries = [name for name in self._eps.index if name.startswith(form)] + for name, loader in self._eps.loc[entries].items(): + fiberio = self._load_entry_point(name, loader) + if fiberio is not None: + self.register_fiberio(fiberio) + if form not in self._format_version: + self._failed_formats.add(form) + self._loaded_formats.add(form) + # The selected format(s) should now be loaded + assert set(formats).isdisjoint(self.unloaded_formats) return - formats = {format} if format is not None else unloaded - # Load one, or all, formats - for form in formats: - entries = [name for name in self._eps.index if name.startswith(form)] - for name, loader in self._eps.loc[entries].items(): - fiberio = self._load_entry_point(name, loader) - if fiberio is not None: - self.register_fiberio(fiberio) - if form not in self._format_version: - self._failed_formats.add(form) - # The selected format(s) should now be loaded - assert set(formats).isdisjoint(self.unloaded_formats) - return def _load_entry_point(self, name: str, loader) -> FiberIO | None: """Load one FiberIO entry point, skipping broken registrations.""" @@ -474,6 +522,7 @@ def _load_entry_point(self, name: str, loader) -> FiberIO | None: warnings.warn(msg, UserWarning, stacklevel=2) return None + @_locked("_lock") def register_fiberio(self, fiberio: FiberIO): """Register a new fiber IO to manage.""" forma, ver = fiberio.name.upper(), fiberio.version @@ -486,6 +535,8 @@ def register_fiberio(self, fiberio: FiberIO): self._format_version[forma][ver] = fiberio self._fiber_io_by_input_type[fiberio.input_type].add(fiberio) self._fiber_io_name_ver.add(id_tuple) + # Snapshots derived from the registry are now stale. + self._lookup_cache.clear() @cached_method def get_fiberio( @@ -574,7 +625,9 @@ def _yield_format_version(self, format, version): assert isinstance(format, str), "Only works once format is known." format = format.upper() self.load_plugins(format) - fiber_ios = self._format_version.get(format, None) + with self._lock: + # Snapshot; the generator must not read shared state while paused. + fiber_ios = dict(self._format_version.get(format, {})) # no format found if not fiber_ios: format_list = list(self.known_formats) @@ -601,7 +654,9 @@ def _yield_extensions(self, extension, input_type=None): has_yielded = set() self.load_plugins() potential_fiberios = self._get_fiber_io_by_input_type(input_type) - for fiber_io in self._extension_list[extension]: + with self._lock: + extension_fiberios = tuple(self._extension_list[extension]) + for fiber_io in extension_fiberios: if fiber_io in potential_fiberios: yield fiber_io has_yielded.add(fiber_io) @@ -931,6 +986,12 @@ def __init_subclass__(cls, **kwargs): setattr(cls, name, method_wrapped) +@_reinit_after_fork +def _reinit_manager_lock(): + """Install a fresh lock on the FiberIO manager; see _reinit_after_fork.""" + FiberIO.manager._lock = RLock() + + def read( path: path_types | IOResourceManager, file_format: str | None = None, diff --git a/dascore/io/index/catalog.py b/dascore/io/index/catalog.py index 7d4ed8abe..6f026385a 100644 --- a/dascore/io/index/catalog.py +++ b/dascore/io/index/catalog.py @@ -19,8 +19,9 @@ import abc import json from collections.abc import Mapping, Sequence -from dataclasses import dataclass, replace +from dataclasses import dataclass, field, replace from pathlib import Path +from threading import RLock import numpy as np import pandas as pd @@ -417,9 +418,21 @@ def _merge_source_records(existing, new): @dataclass class _CatalogRevision: - """Shared mutation revision for live catalog views.""" + """Shared mutation revision, and its lock, for live catalog views.""" value: int = 0 + # A root and all its views share this; it guards the revision counter + # and every cache keyed on it. + lock: RLock = field(default_factory=RLock, repr=False, compare=False) + + def __getstate__(self) -> dict: + """Pickle the revision without its process-local lock.""" + return {"value": self.value} + + def __setstate__(self, state: dict) -> None: + """Restore the revision with a fresh process-local lock.""" + self.value = state["value"] + self.lock = RLock() # sentinel: _view keeps the current order/ids spec unless told otherwise @@ -578,22 +591,25 @@ def backend(self): Every metadata operation funnels through here, so this is also where a brand-new directory index gets its one automatic update. - """ - if self._backend is None: - if self._syncer is not None: - # Directory catalogs re-adopt the (unpickled) syncer's - # backend; both must keep sharing one connection. - self._backend = self._syncer._backend - else: - self._backend = get_backend(":memory:") - if self._rebuild_records: - self._backend.write_sources(list(self._rebuild_records)) - self._rebuild_records = () - elif registry := getattr(self.resolver, "_registry", None): - self._backend.write_sources(_live_records(registry)) - if self._syncer is not None and self._syncer.ensure_updated(): - self._invalidate() - return self._backend + Bootstrapping runs under the revision lock so concurrent first + use cannot build (or ingest into) two backends. + """ + with self._revision.lock: + if self._backend is None: + if self._syncer is not None: + # Directory catalogs re-adopt the (unpickled) syncer's + # backend; both must keep sharing one connection. + self._backend = self._syncer._backend + else: + self._backend = get_backend(":memory:") + if self._rebuild_records: + self._backend.write_sources(list(self._rebuild_records)) + self._rebuild_records = () + elif registry := getattr(self.resolver, "_registry", None): + self._backend.write_sources(_live_records(registry)) + if self._syncer is not None and self._syncer.ensure_updated(): + self._invalidate() + return self._backend def __getstate__(self) -> dict: """ @@ -719,11 +735,12 @@ def restrict(self, indices) -> PatchCatalog: return self._view(self._queries, self._residuals, ids=deduped) def _invalidate(self) -> None: - self._revision.value += 1 - self._df_cache = None - self._df_cache_revision = -1 - self._live_cache = None - self._live_cache_revision = -1 + with self._revision.lock: + self._revision.value += 1 + self._df_cache = None + self._df_cache_revision = -1 + self._live_cache = None + self._live_cache_revision = -1 def _cold_live_values(self) -> tuple | None: """ @@ -733,6 +750,8 @@ def _cold_live_values(self) -> tuple | None: This keeps len/iteration/indexing on freshly-built patch-list spools allocation-free: no ingest, no SQL, no flat relation. + + Callers must hold the revision lock. """ cold = ( self._backend is None @@ -834,64 +853,72 @@ def to_df(self) -> pd.DataFrame: hidden or renamed private so chunk merge-compatibility (which compares all non-private columns) is not spuriously blocked. """ - if self._df_cache is None or self._df_cache_revision != self._revision.value: - df = self.backend.query( - list(self._queries) or None, - order_by=self._effective_order, - patch_ids=self._ids, - ) - if self._ids is not None and self._order is None: - # id membership presents in its own (window/array) order - position = {pid: i for i, pid in enumerate(self._ids)} - df = df.sort_values( - "patch_id", key=lambda s: s.map(position), kind="stable" - ).reset_index(drop=True) - df = df.drop(columns=list(SPOOL_HIDDEN_COLUMNS), errors="ignore").rename( - columns={"patch_id": "_patch_id"} - ) - # SQL identifies overlapping source patches. Expose the selected - # envelopes, matching spool.get_contents() and the exact trim - # applied when each patch is materialized. Each pass copies the - # frame, so disjoint-name range sets collapse into one pass. - range_dicts = [ - ranges - for query in self._queries - if ( - ranges := { - name: _envelope_range(value) - for name, value in query.coords.items() - if is_range(value) - } + with self._revision.lock: + if ( + self._df_cache is None + or self._df_cache_revision != self._revision.value + ): + df = self.backend.query( + list(self._queries) or None, + order_by=self._effective_order, + patch_ids=self._ids, ) - ] - names = [name for ranges in range_dicts for name in ranges] - if range_dicts and len(set(names)) == len(names): - range_dicts = [{k: v for d in range_dicts for k, v in d.items()}] - for ranges in range_dicts: - df = adjust_segments(df, ignore_bad_kwargs=True, **ranges) - self._df_cache = df - self._df_cache_revision = self._revision.value - return self._df_cache + if self._ids is not None and self._order is None: + # id membership presents in its own (window/array) order + position = {pid: i for i, pid in enumerate(self._ids)} + df = df.sort_values( + "patch_id", key=lambda s: s.map(position), kind="stable" + ).reset_index(drop=True) + df = df.drop( + columns=list(SPOOL_HIDDEN_COLUMNS), errors="ignore" + ).rename(columns={"patch_id": "_patch_id"}) + # SQL identifies overlapping source patches. Expose the selected + # envelopes, matching spool.get_contents() and the exact trim + # applied when each patch is materialized. Each pass copies the + # frame, so disjoint-name range sets collapse into one pass. + range_dicts = [ + ranges + for query in self._queries + if ( + ranges := { + name: _envelope_range(value) + for name, value in query.coords.items() + if is_range(value) + } + ) + ] + names = [name for ranges in range_dicts for name in ranges] + if range_dicts and len(set(names)) == len(names): + range_dicts = [{k: v for d in range_dicts for k, v in d.items()}] + for ranges in range_dicts: + df = adjust_segments(df, ignore_bad_kwargs=True, **ranges) + self._df_cache = df + self._df_cache_revision = self._revision.value + return self._df_cache def __len__(self) -> int: - if (live := self._cold_live_values()) is not None: - return len(live) - # Count in SQL when the relation is not already realized: coord - # range residuals only drop patches the SQL candidacy already - # excludes and samples/relative residuals never drop patches, so - # the count matches len(to_df()) without projecting or pivoting. - if ( - self._df_cache is not None - and self._df_cache_revision == self._revision.value - ): - return len(self._df_cache) - return self.backend.count(list(self._queries) or None, patch_ids=self._ids) + with self._revision.lock: + if (live := self._cold_live_values()) is not None: + return len(live) + # Count in SQL when the relation is not already realized: coord + # range residuals only drop patches the SQL candidacy already + # excludes and samples/relative residuals never drop patches, so + # the count matches len(to_df()) without projecting or pivoting. + if ( + self._df_cache is not None + and self._df_cache_revision == self._revision.value + ): + return len(self._df_cache) + return self.backend.count(list(self._queries) or None, patch_ids=self._ids) def get_patch(self, index: int) -> dc.Patch: """Materialize one patch: resolve, then exact two-stage trim.""" - if (live := self._cold_live_values()) is not None: - return live[index] - row = self.to_df().iloc[index].to_dict() + with self._revision.lock: + if (live := self._cold_live_values()) is not None: + return live[index] + row = self.to_df().iloc[index].to_dict() + # Reading (and trimming) the patch happens outside the lock; only + # the row it starts from must come from a consistent relation. return self.resolve_row(row) def resolve_row(self, row: Mapping, extra_trim: Mapping | None = None) -> dc.Patch: @@ -929,21 +956,24 @@ def __iter__(self): def add(self, patches: Sequence[dc.Patch] | dc.Patch) -> PatchCatalog: """Add live patches to the catalog.""" - self._require_root("add") - if not isinstance(self.resolver, LiveResolver): - msg = "add() currently supports in-memory catalogs only." - raise NotImplementedError(msg) - patches = [patches] if isinstance(patches, dc.Patch) else list(patches) - additions = {_patch_path(x): x for x in patches} - self.resolver._registry.update(additions) - # Re-adding a patch replaces its row (same identity), so this - # stays idempotent. - self.backend.write_sources(_live_records(additions)) - self._invalidate() - return self + with self._revision.lock: + self._require_root("add") + if not isinstance(self.resolver, LiveResolver): + msg = "add() currently supports in-memory catalogs only." + raise NotImplementedError(msg) + patches = [patches] if isinstance(patches, dc.Patch) else list(patches) + additions = {_patch_path(x): x for x in patches} + self.resolver._registry.update(additions) + # Re-adding a patch replaces its row (same identity), so this + # stays idempotent. + self.backend.write_sources(_live_records(additions)) + self._invalidate() + return self def update(self, progress: PROGRESS_LEVELS = "standard") -> PatchCatalog: """Sync a directory-backed catalog with the filesystem.""" + # The scan itself is long-running file IO; it manages its own + # state, so only the cache invalidation needs the revision lock. if self._syncer is not None: self._syncer.update(progress=progress) self._invalidate() @@ -951,16 +981,17 @@ def update(self, progress: PROGRESS_LEVELS = "standard") -> PatchCatalog: def remove(self, source_paths: Sequence[str], base_uri: str = "") -> PatchCatalog: """Remove sources (and their patches) from the catalog.""" - self._require_root("remove") - source_paths = list(source_paths) - self.backend.delete_sources(source_paths, base_uri=base_uri) - # The live registry is the store for in-memory patches; it must - # stay in step with the backend rows (pickling rebuilds from it). - registry = self.resolver.live_entries() if self.resolver else {} - for path in source_paths: - registry.pop(path, None) - self._invalidate() - return self + with self._revision.lock: + self._require_root("remove") + source_paths = list(source_paths) + self.backend.delete_sources(source_paths, base_uri=base_uri) + # The live registry is the store for in-memory patches; it must + # stay in step with the backend rows (pickling rebuilds from it). + registry = self.resolver.live_entries() if self.resolver else {} + for path in source_paths: + registry.pop(path, None) + self._invalidate() + return self # --- introspection ------------------------------------------------------- @@ -982,5 +1013,6 @@ def get_metadata(self) -> dict: def close(self) -> None: """Close the backend (root and all views share it).""" - if self._backend is not None: - self._backend.close() + with self._revision.lock: + if self._backend is not None: + self._backend.close() diff --git a/dascore/units.py b/dascore/units.py index 8ce1e35f7..696fd8a8e 100644 --- a/dascore/units.py +++ b/dascore/units.py @@ -5,6 +5,7 @@ import shutil from collections.abc import Sequence from functools import cache +from threading import RLock from typing import Any, TypeVar import numpy as np @@ -16,7 +17,7 @@ import dascore as dc from dascore.compat import is_array from dascore.exceptions import UnitError -from dascore.utils.misc import iterate, unbyte +from dascore.utils.misc import _reinit_after_fork, iterate, unbyte from dascore.utils.time import dtype_time_like, is_datetime64, is_timedelta64, to_float str_or_none = TypeVar("str_or_none", None, str) @@ -33,26 +34,45 @@ def _get_unit_registry(): return pint.UnitRegistry(cache_folder=":auto:") -@cache +# The pint registry is mutable (it memoizes parsed units and conversions) +# so every helper which touches it runs under this lock. Each such helper +# is also cached, meaning the lock is only taken on a cache miss. +_UNIT_LOCK = RLock() +_UNIT_REGISTRY: pint.UnitRegistry | None = None + + +@_reinit_after_fork +def _reinit_unit_lock(): + """Install a fresh unit lock; see _reinit_after_fork.""" + global _UNIT_LOCK + _UNIT_LOCK = RLock() + + def get_registry(): - """Get the pint unit registry.""" - ureg = _get_unit_registry() - # a few custom defs, we may need our own unit registry if this - # gets too long. - ureg.define("PI=pi") - ureg.define("RADIANS=radians") - ureg.define("Radians=radians") - ureg.define("Radian=radians") - # define strain - ureg.define("strain=[]=ϵ") - # allow multiplication with offset units. - ureg.autoconvert_offset_to_baseunit = True - # set the shortest display for units. - # .formatter was added in new versions of pint; this makes it work with both - formatter = getattr(ureg, "formatter", ureg) - formatter.default_format = "~" - pint.set_application_registry(ureg) - return ureg + """Get the pint unit registry, creating it exactly once.""" + global _UNIT_REGISTRY + with _UNIT_LOCK: + if _UNIT_REGISTRY is None: + ureg = _get_unit_registry() + # a few custom defs, we may need our own unit registry if this + # gets too long. + ureg.define("PI=pi") + ureg.define("RADIANS=radians") + ureg.define("Radians=radians") + ureg.define("Radian=radians") + # define strain + ureg.define("strain=[]=ϵ") + # allow multiplication with offset units. + ureg.autoconvert_offset_to_baseunit = True + # set the shortest display for units. + # .formatter was added in new versions of pint; this makes it + # work with both + formatter = getattr(ureg, "formatter", ureg) + formatter.default_format = "~" + pint.set_application_registry(ureg) + # Publish only once fully defined. + _UNIT_REGISTRY = ureg + return _UNIT_REGISTRY @cache @@ -80,16 +100,18 @@ def get_unit(value) -> Unit: if isinstance(value, Quantity): assert value.magnitude == 1.0 value = value.units - return get_registry().Unit(value) + with _UNIT_LOCK: + return get_registry().Unit(value) @cache def _str_to_quant(qunat_str): """Get quantity from a string; cache output.""" - if isinstance(qunat_str, Unit): - qunat_str = str(qunat_str) # ensure unit is converted to quantity - ureg = get_registry() - return ureg.Quantity(qunat_str) + with _UNIT_LOCK: + if isinstance(qunat_str, Unit): + qunat_str = str(qunat_str) # ensure unit is converted to quantity + ureg = get_registry() + return ureg.Quantity(qunat_str) def get_quantity(value: str_or_none) -> Quantity | None: @@ -135,12 +157,13 @@ def get_factor_and_unit( @cache def _get_conversion_factors(from_quant, to_quant) -> tuple[float, float, float]: """Get multiplicative and additive conversion factors.""" - add_mag = (0 * from_quant).to(0 * to_quant).magnitude - # need to convert from and to units to deltas for proper conversion. - from_delta = (1 * from_quant.units) - (from_quant.units * 0) - to_delta = (1 * to_quant.units) - (to_quant.units * 0) - mult_mag1 = from_delta.to(to_delta).magnitude - return mult_mag1 * from_quant.magnitude, add_mag, 1 / to_quant.magnitude + with _UNIT_LOCK: + add_mag = (0 * from_quant).to(0 * to_quant).magnitude + # need to convert from and to units to deltas for proper conversion. + from_delta = (1 * from_quant.units) - (from_quant.units * 0) + to_delta = (1 * to_quant.units) - (to_quant.units * 0) + mult_mag1 = from_delta.to(to_delta).magnitude + return mult_mag1 * from_quant.magnitude, add_mag, 1 / to_quant.magnitude def convert_units( @@ -217,7 +240,8 @@ def _unit_to_str(unit: Unit) -> str: Unit equality/hashing is exact (e.g. m != cm) so, unlike Quantity, Unit is safe to use as a cache key. """ - return str(unit) + with _UNIT_LOCK: + return str(unit) def get_quantity_str(quant_value: str | Quantity | None) -> str | None: @@ -255,7 +279,8 @@ def get_quantity_str(quant_value: str | Quantity | None) -> str | None: def _validate_quantity_str(quant_str: str) -> None: """Raise a UnitError if the string doesn't specify a valid quantity.""" try: - get_quantity(quant_str) + with _UNIT_LOCK: + get_quantity(quant_str) except UndefinedUnitError as e: msg = f"DASCore failed to parse the following unit/quantity: {quant_str}" raise UnitError(msg) from e diff --git a/dascore/utils/misc.py b/dascore/utils/misc.py index 306f007a6..711285ba9 100644 --- a/dascore/utils/misc.py +++ b/dascore/utils/misc.py @@ -630,6 +630,47 @@ def _matches_prefix_suffix(input_str, suffixes, prefixes=None): return bool(re.match(regex, input_str)) +def _locked(lock_name: str): + """ + Run the decorated method while holding one of its owner's locks. + + Parameters + ---------- + lock_name + Name of the attribute (on self or cls) holding the lock, eg "_lock". + It is looked up on each call so the owner can swap in a fresh lock + (eg after a fork). + + Notes + ----- + Not for generator functions; the lock would be held across the + consumer's iteration rather than the function body. + """ + + def _decorator(func): + @functools.wraps(func) + def _wrapper(self, *args, **kwargs): + with getattr(self, lock_name): + return func(self, *args, **kwargs) + + return _wrapper + + return _decorator + + +def _reinit_after_fork(func): + """ + Register a callable to run in the child process after a fork. + + A fork can copy a lock while another thread holds it. That thread does + not exist in the child, so the inherited copy would never be released. + Hooks registered here install fresh locks in the child. + """ + if hasattr(os, "register_at_fork"): # not available on windows + os.register_at_fork(after_in_child=func) + return func + + def cached_method(func): """ Cache decorated method. diff --git a/dascore/utils/namespace.py b/dascore/utils/namespace.py index 8ed647919..d5a747fb5 100644 --- a/dascore/utils/namespace.py +++ b/dascore/utils/namespace.py @@ -6,19 +6,28 @@ import warnings from collections import defaultdict from pathlib import Path +from threading import RLock from typing import ClassVar import pandas as pd from dascore.exceptions import DASCorePluginError from dascore.utils.mapping import FrozenDict +from dascore.utils.misc import _locked, _reinit_after_fork from dascore.utils.plugins import maybe_load_entry_point _PLUGIN_REGISTRY_DIR = Path(__file__).parent.parent / "plugin_registry" +# Serializes attaching a lazily loaded namespace to its host object. One +# global lock (rather than one per host) is enough because attachment only +# constructs a small wrapper around the host. +_ATTACHMENT_LOCK = RLock() + @functools.cache -def _load_plugin_registry(entry_point_group: str | None) -> dict[str, tuple[str, str]]: +def _load_plugin_registry( + entry_point_group: str | None, +) -> FrozenDict[str, tuple[str, str]]: """Load plugin registry CSV for the given entry point group. Parameters @@ -31,14 +40,14 @@ def _load_plugin_registry(entry_point_group: str | None) -> dict[str, tuple[str, A mapping of namespace name to (package_name, package_url). """ if entry_point_group is None: - return {} + return FrozenDict() # "dascore.patch_namespace" -> "patch" stem = entry_point_group.split(".")[-1].replace("_namespace", "") csv_path = _PLUGIN_REGISTRY_DIR / f"{stem}.csv" if not csv_path.exists(): - return {} + return FrozenDict() df = pd.read_csv(csv_path) - return dict(zip(df["namespace"], zip(df["package_name"], df["package_url"]))) + return FrozenDict(zip(df["namespace"], zip(df["package_name"], df["package_url"]))) def _pass_to_host_method(func): @@ -80,6 +89,8 @@ class _MethodNameSpace(metaclass=_NameSpaceMeta): _registry: ClassVar[defaultdict[str | None, dict[str, type[_MethodNameSpace]]]] = ( defaultdict(dict) ) + # Guards every read and write of _registry. + _registry_lock: ClassVar[RLock] = RLock() def __init__(self, obj): self._obj = obj @@ -92,17 +103,32 @@ def __init_subclass__(cls, **kwargs): setattr(cls, key, val) # Register all subclasses. if cls.name is not None: - registry = cls._registry[cls.entry_point_group] - existing = registry.get(cls.name) - if existing is not None and existing is not cls: - msg = ( - f"Namespace collision for group {cls.entry_point_group!r} and " - f"name {cls.name!r}: replacing " - f"{existing.__module__}.{existing.__name__} with " - f"{cls.__module__}.{cls.__name__}." - ) - warnings.warn(msg, UserWarning, stacklevel=2) - registry[cls.name] = cls + with cls._registry_lock: + registry = cls._registry[cls.entry_point_group] + existing = registry.get(cls.name) + if existing is not None and existing is not cls: + msg = ( + f"Namespace collision for group {cls.entry_point_group!r} and " + f"name {cls.name!r}: replacing " + f"{existing.__module__}.{existing.__name__} with " + f"{cls.__module__}.{cls.__name__}." + ) + warnings.warn(msg, UserWarning, stacklevel=2) + registry[cls.name] = cls + + @classmethod + @_locked("_registry_lock") + def _get_namespace_type(cls, group: str | None, name: str): + """Return the namespace class registered to a group, or None.""" + return cls._registry.get(group, {}).get(name) + + +@_reinit_after_fork +def _reinit_namespace_locks(): + """Install fresh namespace locks; see _reinit_after_fork.""" + global _ATTACHMENT_LOCK + _ATTACHMENT_LOCK = RLock() + _MethodNameSpace._registry_lock = RLock() class PatchNameSpace(_MethodNameSpace): @@ -128,19 +154,26 @@ class NamespaceOwner: @classmethod def get_registered_namespaces(cls): """Return registered method namespaces on the class.""" - registry = _MethodNameSpace._registry.get(cls._namespace_entry_point_group, {}) - return FrozenDict(registry) + with _MethodNameSpace._registry_lock: + registry = _MethodNameSpace._registry.get( + cls._namespace_entry_point_group, {} + ) + return FrozenDict(registry) def __getattr__(self, item): """Try loading a lazily registered namespace before failing.""" - # Unknown attribute; try loading the namespaces. - maybe_load_entry_point(self._namespace_entry_point_group, item) + # Unknown attribute; try loading the namespaces. Plugin imports must + # not run while a lock is held. + group = self._namespace_entry_point_group + maybe_load_entry_point(group, item) # Once loaded the registry should be populated. - registry = _MethodNameSpace._registry.get(self._namespace_entry_point_group, {}) - if item in registry: - instance = registry[item](self) - self.__dict__[item] = instance + if namespace_type := _MethodNameSpace._get_namespace_type(group, item): + with _ATTACHMENT_LOCK: + # Another thread may have attached this namespace first; reuse + # its instance so all callers share one namespace object. + if (instance := self.__dict__.get(item)) is None: + instance = self.__dict__[item] = namespace_type(self) return instance # Check plugin registry for a known third-party package that provides this. diff --git a/tests/test_core/test_coord_segmented.py b/tests/test_core/test_coord_segmented.py index 7b68c389e..839a5a3b9 100644 --- a/tests/test_core/test_coord_segmented.py +++ b/tests/test_core/test_coord_segmented.py @@ -230,6 +230,11 @@ def test_not_evenly_sampled(self, float_gap_coord): """Segmented coords are never evenly sampled.""" assert not float_gap_coord.evenly_sampled + def test_cached_arrays_read_only(self, float_gap_coord): + """Cached arrays are shared between callers, so must be read-only.""" + assert not float_gap_coord._segment_offsets().flags.writeable + assert not float_gap_coord.values.flags.writeable + def test_sorted_flags(self, float_gap_coord, reverse_gap_coord): """Sort direction is reported correctly.""" assert float_gap_coord.sorted diff --git a/tests/test_core/test_coordmanager.py b/tests/test_core/test_coordmanager.py index f6a1e1a2f..389eed7a7 100644 --- a/tests/test_core/test_coordmanager.py +++ b/tests/test_core/test_coordmanager.py @@ -85,6 +85,13 @@ def test_to_dict(self, coord_manager): for key in set(expected): assert np.all(expected[key] == np.array(c_dict[key])) + def test_coord_shapes_immutable(self, coord_manager): + """coord_shapes is cached and shared, so it must be immutable.""" + shapes = coord_manager.coord_shapes + assert set(shapes) == set(coord_manager.coord_map) + with pytest.raises(TypeError): + shapes["time"] = (1,) + def test_membership(self, coord_manager): """Coord membership should work for coord names.""" coords = list(coord_manager.coord_map) diff --git a/tests/test_core/test_coords.py b/tests/test_core/test_coords.py index 0db3e35b4..c526614db 100644 --- a/tests/test_core/test_coords.py +++ b/tests/test_core/test_coords.py @@ -1156,6 +1156,12 @@ def test_values(self, evenly_sampled_coord): vals = evenly_sampled_coord.values assert len(vals) == len(evenly_sampled_coord) + def test_values_are_read_only(self, evenly_sampled_coord): + """Cached values are shared, so they must not be writable.""" + coords = [evenly_sampled_coord, get_coord(start=0, stop=1, step=1)] + for coord in coords: + assert not coord.values.flags.writeable + def test_set_units(self, evenly_sampled_coord): """Ensure units can be set.""" out = evenly_sampled_coord.set_units("m") diff --git a/tests/test_core/test_spool.py b/tests/test_core/test_spool.py index 4c2c4e57b..f3c79982a 100644 --- a/tests/test_core/test_spool.py +++ b/tests/test_core/test_spool.py @@ -439,6 +439,12 @@ def test_filter(self, random_spool): assert len(sub) == (len(full_df) - 1) assert (sub["time_min"] < new_max).all() + def test_contents_are_caller_owned(self, random_spool): + """Mutating the returned dataframe must not change the spool.""" + df = random_spool.get_contents() + df["tag"] = "modified" + assert (random_spool.get_contents()["tag"] != "modified").all() + class TestSelect: """Tests for selecting/trimming spools.""" diff --git a/tests/test_io/test_index/test_catalog.py b/tests/test_io/test_index/test_catalog.py index 1ce795fec..daba877a7 100644 --- a/tests/test_io/test_index/test_catalog.py +++ b/tests/test_io/test_index/test_catalog.py @@ -2,6 +2,9 @@ from __future__ import annotations +import pickle +import threading + import numpy as np import pytest @@ -306,3 +309,47 @@ def test_view_pickles_membership_only(self): assert loaded[0].attrs["tag"] == "0" # and the root spool's registry is untouched assert len(spool._catalog.resolver.live_entries()) == 5 + + +class TestCatalogConcurrency: + """Catalog caches must stay coherent under concurrent readers.""" + + def _run(self, func, count=4): + """Run func(index) in count threads, all released together.""" + barrier = threading.Barrier(count) + results = [None] * count + + def worker(index): + barrier.wait() + results[index] = func(index) + + threads = [threading.Thread(target=worker, args=(i,)) for i in range(count)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + return results + + def test_concurrent_realization(self, live_catalog, patches): + """Racing first realizations build one backend and one relation.""" + results = self._run(lambda _: len(live_catalog.to_df())) + assert set(results) == {len(patches)} + # one cached relation, not one per thread + assert live_catalog.to_df() is live_catalog.to_df() + + def test_concurrent_reads(self, live_catalog, patches): + """Mixed len/patch access from several threads agrees.""" + + def read(index): + """Read the catalog a few different ways.""" + return (len(live_catalog), live_catalog.get_patch(index).shape) + + results = self._run(lambda index: read(index), len(patches)) + assert {x[0] for x in results} == {len(patches)} + assert [x[1] for x in results] == [x.shape for x in patches] + + def test_pickled_revision_gets_new_lock(self, live_catalog): + """Unpickling a catalog installs a fresh revision lock.""" + rebuilt = pickle.loads(pickle.dumps(live_catalog)) + assert rebuilt._revision.lock is not live_catalog._revision.lock + assert len(rebuilt) == len(live_catalog) diff --git a/tests/test_io/test_io_core.py b/tests/test_io/test_io_core.py index b892ee3d3..5ac4605e3 100644 --- a/tests/test_io/test_io_core.py +++ b/tests/test_io/test_io_core.py @@ -4,6 +4,7 @@ import copy import io +import threading from pathlib import Path from typing import TypeVar @@ -26,6 +27,7 @@ ) from dascore.io.core import ( FiberIO, + _FiberIOManager, _get_reloadable_source_path, _make_scan_payload, _scan_result_to_summary, @@ -423,7 +425,7 @@ def test_known_formats_empty_entry_points(self, format_manager): format_manager.__dict__.pop("_eps", None) format_manager.__dict__.pop("known_formats", None) format_manager._eps = pd.Series(dtype=object) - assert isinstance(format_manager.known_formats, set) + assert isinstance(format_manager.known_formats, frozenset) def test_load_plugins_empty_entry_points(self, format_manager): """Loading plugins should no-op when no entry points are present.""" @@ -508,6 +510,101 @@ def test_other_formats_still_usable(self, broken_ep_manager): assert len(list(broken_ep_manager.yield_fiberio())) +class TestFormatManagerConcurrency: + """Concurrent plugin loading must never expose a partial registry.""" + + def _make_manager(self, eps): + """Return a manager whose entry points are the provided loaders.""" + manager = _FiberIOManager("dascore.fiber_io") + manager.__dict__["_eps"] = pd.Series(eps) + return manager + + def _run(self, func, count): + """Run func(index) in count threads, all released together.""" + barrier = threading.Barrier(count) + results = [None] * count + + def worker(index): + barrier.wait() + results[index] = func(index) + + threads = [threading.Thread(target=worker, args=(i,)) for i in range(count)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + return results + + def test_multi_version_format_never_partial(self): + """Every thread sees all versions, even mid-load.""" + entered, release, calls = threading.Event(), threading.Event(), [] + + def slow_loader(): + """Stall inside the first version's loader.""" + calls.append("v1") + entered.set() + release.wait() + return _FiberFormatTestV1 + + manager = self._make_manager( + { + "_TESTFORMATTER__V1": slow_loader, + "_TESTFORMATTER__V2": lambda: _FiberFormatTestV2, + } + ) + # Free the stalled loader once it has registered nothing but v1; + # the other threads are queued behind the manager lock by then. + releaser = threading.Thread(target=lambda: (entered.wait(), release.set())) + releaser.start() + results = self._run( + lambda _: tuple( + x.version for x in manager.yield_fiberio(format="_TestFormatter") + ), + 4, + ) + releaser.join() + # Newest version first, and the loader ran exactly once. + assert all(x == ("2", "1") for x in results) + assert calls == ["v1"] + + def test_concurrent_full_load_runs_each_loader_once(self): + """Loading all formats from several threads loads each entry point once.""" + calls = [] + + def make_loader(fiber_io): + """Return a loader which records that it ran.""" + + def loader(): + calls.append(fiber_io.version) + return fiber_io + + return loader + + manager = self._make_manager( + { + "_TESTFORMATTER__V1": make_loader(_FiberFormatTestV1), + "_TESTFORMATTER__V2": make_loader(_FiberFormatTestV2), + } + ) + results = self._run(lambda _: len(list(manager.yield_fiberio())), 4) + assert set(results) == {2} + assert sorted(calls) == ["1", "2"] + + def test_snapshots_are_immutable(self): + """Cached lookups hand back immutable snapshots.""" + manager = self._make_manager({"_TESTFORMATTER__V1": lambda: _FiberFormatTestV1}) + assert isinstance(manager.known_formats, frozenset) + assert isinstance(manager._get_prioritized_list(), tuple) + assert isinstance(manager._get_fiber_io_by_input_type("file"), frozenset) + + def test_copy_gets_own_lock(self): + """A copied manager must not share the original's lock.""" + manager = self._make_manager({"_TESTFORMATTER__V1": lambda: _FiberFormatTestV1}) + copied = copy.deepcopy(manager) + assert copied._lock is not manager._lock + assert list(copied.yield_fiberio(format="_TestFormatter")) + + class TestFormatter: """Tests for adding file supports through Formatter.""" diff --git a/tests/test_units.py b/tests/test_units.py index e91f1f454..825bf18fa 100644 --- a/tests/test_units.py +++ b/tests/test_units.py @@ -2,10 +2,14 @@ from __future__ import annotations +import threading + import numpy as np +import pint import pytest import dascore as dc +import dascore.units as units_module from dascore.exceptions import UnitError from dascore.units import ( Quantity, @@ -465,3 +469,41 @@ def test_fractional_percentage(self): result = maybe_convert_percent_to_fraction(get_quantity("12.5%")) assert len(result) == 1 assert np.isclose(result[0], 0.125) + + +class TestUnitConcurrency: + """The pint registry must initialize once and parse safely in threads.""" + + def _run(self, func, count=4): + """Run func(index) in count threads, all released together.""" + barrier = threading.Barrier(count) + results = [None] * count + + def worker(index): + barrier.wait() + results[index] = func(index) + + threads = [threading.Thread(target=worker, args=(i,)) for i in range(count)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + return results + + def test_registry_created_once(self, monkeypatch): + """Racing threads all get the same registry instance.""" + # The registry is process-wide; restore it so quantities created by + # other tests keep belonging to the active registry. + monkeypatch.setattr(units_module, "_UNIT_REGISTRY", None) + original = pint.get_application_registry().get() + try: + results = self._run(lambda _: units_module.get_registry()) + finally: + pint.set_application_registry(original) + assert len({id(x) for x in results}) == 1 + + def test_concurrent_parsing(self): + """Parsing distinct quantities in threads returns correct values.""" + strings = ["m/s", "1/s", "furlong/fortnight", "strain"] + results = self._run(lambda index: get_quantity(strings[index])) + assert results == [get_quantity(x) for x in strings] diff --git a/tests/test_utils/test_misc.py b/tests/test_utils/test_misc.py index 8bc23f04b..b221b1ce9 100644 --- a/tests/test_utils/test_misc.py +++ b/tests/test_utils/test_misc.py @@ -3,10 +3,12 @@ from __future__ import annotations import os +import threading import time import warnings from io import BytesIO from pathlib import Path +from threading import Lock import numpy as np import pandas as pd @@ -16,6 +18,7 @@ from dascore.exceptions import MissingOptionalDependencyError from dascore.utils.misc import ( _iter_filesystem, + _locked, _spool_map, all_diffs_close_enough, cached_method, @@ -479,6 +482,41 @@ def test_data_with_outliers(self): assert np.allclose(result, [expected_lower, expected_upper]) +class TestLocked: + """Ensure the _locked decorator runs the body holding the owner's lock.""" + + class _Counter: + """A class whose increments run under a non-reentrant lock.""" + + def __init__(self): + self._lock = Lock() + self.value = 0 + + @_locked("_lock") + def increment(self): + """Increment; the decorator must hold _lock for this call.""" + # A non-reentrant lock cannot be re-acquired while it is held. + assert not self._lock.acquire(blocking=False) + self.value += 1 + + def test_every_call_is_locked(self): + """Concurrent callers each run the body with the lock held.""" + counter = self._Counter() + threads = [threading.Thread(target=counter.increment) for _ in range(8)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + assert counter.value == 8 + + def test_lock_looked_up_per_call(self): + """A swapped-in lock is the one used by later calls.""" + counter = self._Counter() + counter._lock = Lock() + counter.increment() + assert counter.value == 1 + + class TestCachedMethod: """Ensure cached methods caches method calls (duh).""" diff --git a/tests/test_utils/test_namespace.py b/tests/test_utils/test_namespace.py index 381cd3c86..a843c67b5 100644 --- a/tests/test_utils/test_namespace.py +++ b/tests/test_utils/test_namespace.py @@ -2,6 +2,7 @@ from __future__ import annotations +import threading from typing import ClassVar import pandas as pd @@ -258,3 +259,51 @@ def test_getattr_unknown_attr_raises_default_error(self, monkeypatch, tmp_path): AttributeError, match="ParentClass has no attribute 'totally_unknown'" ): inst.totally_unknown + + +class TestNamespaceConcurrency: + """Lazy namespace attachment must be safe under concurrent first use.""" + + def _run(self, func, count=4): + """Run func(index) in count threads, all released together.""" + barrier = threading.Barrier(count) + results = [None] * count + + def worker(index): + barrier.wait() + results[index] = func(index) + + threads = [threading.Thread(target=worker, args=(i,)) for i in range(count)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + return results + + def test_one_instance_per_host(self): + """Concurrent first access on one object returns a single namespace.""" + inst = ParentClass() + results = self._run(lambda _: inst.bob) + assert len({id(x) for x in results}) == 1 + assert results[0] is inst.bob + + def test_distinct_hosts_get_own_namespace(self): + """Concurrent first access on distinct objects binds each host.""" + instances = [ParentClass() for _ in range(4)] + results = self._run(lambda index: instances[index].bob) + assert [x.return_self() for x in results] == instances + + def test_concurrent_registration_and_lookup(self): + """Registering namespaces while reading the registry stays consistent.""" + + class ConcurrentBase(_MethodNameSpace): + entry_point_group = "dascore.concurrent_test" + + def register(index): + """Define a new namespace and read the registry back.""" + type(f"Namespace{index}", (ConcurrentBase,), {"name": f"ns_{index}"}) + return set(_MethodNameSpace._registry["dascore.concurrent_test"]) + + self._run(register) + registered = _MethodNameSpace._registry["dascore.concurrent_test"] + assert set(registered) == {f"ns_{i}" for i in range(4)} From b954aad3929d0bb9cf9c1d1412d6b6db7875dc07 Mon Sep 17 00:00:00 2001 From: Derrick Chambers Date: Sat, 25 Jul 2026 14:47:12 +0200 Subject: [PATCH 2/3] Refactor caches and iteration, address review Refactors: - Collapse the duplicated revision-stamp caches in PatchCatalog into a small _RevisionCache (value plus the revision it was built at). - PatchCatalog.__iter__ is now the single iteration implementation: it snapshots the relation under one lock acquisition, resolves patches outside the lock, and owns the #583 skip warning. Spool.__iter__ delegates to it instead of repeating the loop with per-patch locking. - _FiberIOManager registries are plain dicts, so a missing-key read can no longer register an empty entry. - load_plugins expresses its bookkeeping as one set expression, and regains a memoized no-op for the already-loaded case (its repeat call is on the get_format path). Review (#779): - get_contents: only the literal True enables pandas copy-on-write, so the "warn" setting takes the deep copy again. - _load_plugin_registry zips strictly. - The concurrent registry test snapshots under the registry lock. - The four copies of the thread harness become one run_in_threads fixture, with barrier and join timeouts so a deadlock fails instead of hanging. - Document why the one-time directory scan stays under the revision lock (it serializes the build, and readers have nothing to read until it finishes) while a later update() does not. Adds tests for the fork handlers and both dataframe-copy branches, which were the only uncovered lines in the patch. --- dascore/core/spool.py | 16 +-- dascore/io/core.py | 71 +++++----- dascore/io/index/catalog.py | 167 ++++++++++++++--------- dascore/utils/namespace.py | 8 +- tests/conftest.py | 30 ++++ tests/test_core/test_spool.py | 21 ++- tests/test_io/test_index/test_catalog.py | 27 +--- tests/test_io/test_io_core.py | 43 +++--- tests/test_units.py | 40 +++--- tests/test_utils/test_namespace.py | 48 +++---- 10 files changed, 275 insertions(+), 196 deletions(-) diff --git a/dascore/core/spool.py b/dascore/core/spool.py index 1eff802dc..07529b56b 100644 --- a/dascore/core/spool.py +++ b/dascore/core/spool.py @@ -3,7 +3,6 @@ from __future__ import annotations import abc -import warnings from collections.abc import Callable, Generator, Sequence from functools import singledispatch from pathlib import Path @@ -54,10 +53,12 @@ def _copy_public_dataframe(frame: pd.DataFrame) -> pd.DataFrame: With copy-on-write (always on in pandas 3) a shallow copy already detaches on the first write; without it the blocks must be copied. + Only the literal True enables isolation: the "warn" setting keeps the + old (sharing) semantics, so it still needs a deep copy. """ if int(pd.__version__.split(".", maxsplit=1)[0]) >= 3: return frame.copy(deep=False) - return frame.copy(deep=not pd.options.mode.copy_on_write) + return frame.copy(deep=pd.options.mode.copy_on_write is not True) class BaseSpool(NamespaceOwner, abc.ABC): @@ -532,14 +533,9 @@ def __getitem__(self, item) -> PatchType | BaseSpool: raise IndexError(msg) from None def __iter__(self): - for ind in range(len(self._catalog)): - try: - yield self._catalog.get_patch(ind) - except MissingPatchError as e: - # The patch couldn't be produced, usually because a - # coordinate mismatch trimmed it to nothing (see #583). - msg = f"Skipping patch at index {ind} (see #583): {e}" - warnings.warn(msg, UserWarning, stacklevel=2) + # The catalog snapshots the relation once and skips patches which + # cannot be resolved (see #583). + yield from self._catalog # --- selection and presentation specs ------------------------------- diff --git a/dascore/io/core.py b/dascore/io/core.py index fafcd139c..57393c1f3 100644 --- a/dascore/io/core.py +++ b/dascore/io/core.py @@ -391,11 +391,16 @@ def __init__(self, entry_point: str): # Formats whose load attempt finished, successfully or not. The # outcome lives in _format_version/_failed_formats. self._loaded_formats: set[str] = set() + # True once no format is left to load; keeps the (hot) repeat call + # to load_plugins() off the lock entirely. + self._all_loaded = False self._failed_formats: set[str] = set() - self._format_version = defaultdict(dict) - self._extension_list = defaultdict(list) - # This is a dict of {input_type: (fiberio_name, version)} - self._fiber_io_by_input_type = defaultdict(set) + # Plain dicts, not defaultdicts: these are shared state, and a + # missing-key read must not register anything. + self._format_version: dict[str, dict[str, FiberIO]] = {} + self._extension_list: dict[str, list[FiberIO]] = {} + # This is a dict of {input_type: {fiberio, ...}} + self._fiber_io_by_input_type: dict[str, set[FiberIO]] = {} self._fiber_io_name_ver = set() # Snapshots derived from the registry; cleared when it changes. self._lookup_cache: dict[tuple, frozenset | tuple] = {} @@ -439,27 +444,24 @@ def _get_fiber_io_by_input_type(self, input_type) -> frozenset[FiberIO]: """Get a set of FiberIO instances that meet input type.""" key = ("input_type", input_type) if (cached := self._lookup_cache.get(key)) is None: - if input_type not in self._fiber_io_by_input_type: + if (out := self._fiber_io_by_input_type.get(input_type)) is None: out = set() for input_set in self._fiber_io_by_input_type.values(): out |= input_set - else: - out = self._fiber_io_by_input_type[input_type] cached = self._lookup_cache[key] = frozenset(out) return cached @_locked("_lock") def _get_prioritized_list(self, input_type="file") -> tuple[FiberIO, ...]: """Yield a prioritized list of fiber_ios.""" - # must load all plugins before getting list - self.load_plugins() key = ("prioritized", input_type) if (cached := self._lookup_cache.get(key)) is not None: return cached + # must load all plugins before getting list + self.load_plugins() priority_fiber_ios = [] second_class_fiber_ios = [] for format_name in self.known_formats: - # Use get; indexing the defaultdict would register empty formats. if not (unsorted := self._format_version.get(format_name)): continue keys = sorted(unsorted, reverse=True) @@ -477,24 +479,29 @@ def _get_prioritized_list(self, input_type="file") -> tuple[FiberIO, ...]: def load_plugins(self, format: str | None = None): """Load plugin for specific format or ensure all formats are loaded.""" - # A format only lands in _loaded_formats once every one of its entry - # points is registered, so this fast path (and the lock below) keep - # multi-version formats from being seen half loaded. - if format is not None and format in self._loaded_formats: + # A format only lands in _loaded_formats (or _all_loaded) once every + # one of its entry points is registered, so these fast paths (and + # the lock below) keep multi-version formats from being seen half + # loaded. + if self._all_loaded or (format is not None and format in self._loaded_formats): return with self._lock: - if format is not None and format in self._format_version: - self._loaded_formats.add(format) - return # already loaded - if not (unloaded := self.unloaded_formats): - return - formats = {format} if format is not None else unloaded - # Load one, or all, formats. Plugin imports deliberately run - # under the lock: tracking in-flight formats instead would need - # a claim/wait graph for a step which happens once per format. - # The cost is that a thread importing a module which defines a - # FiberIO waits here until any in-progress load finishes. - for form in formats: + # Anything already registered (directly, or by another thread + # while this one waited) is not pending; it only needs stamping. + pending = set(self.unloaded_formats) + formats = {format} if format is not None else pending + self._loaded_formats |= formats + # known_formats is fixed once computed, so what stays pending + # here stays pending until it is loaded. + self._all_loaded = not (pending - formats) + if not (todo := formats & pending): + return # nothing left to load; already registered or failed + # Plugin imports deliberately run under the lock: tracking + # in-flight formats instead would need a claim/wait graph for a + # step which happens once per format. The cost is that a thread + # importing a module which defines a FiberIO waits here until + # any in-progress load finishes. + for form in todo: entries = [name for name in self._eps.index if name.startswith(form)] for name, loader in self._eps.loc[entries].items(): fiberio = self._load_entry_point(name, loader) @@ -502,10 +509,8 @@ def load_plugins(self, format: str | None = None): self.register_fiberio(fiberio) if form not in self._format_version: self._failed_formats.add(form) - self._loaded_formats.add(form) # The selected format(s) should now be loaded - assert set(formats).isdisjoint(self.unloaded_formats) - return + assert formats.isdisjoint(self.unloaded_formats) def _load_entry_point(self, name: str, loader) -> FiberIO | None: """Load one FiberIO entry point, skipping broken registrations.""" @@ -531,9 +536,9 @@ def register_fiberio(self, fiberio: FiberIO): return self._loaded_eps.add(fiberio.name) for ext in iter(fiberio.preferred_extensions): - self._extension_list[ext].append(fiberio) - self._format_version[forma][ver] = fiberio - self._fiber_io_by_input_type[fiberio.input_type].add(fiberio) + self._extension_list.setdefault(ext, []).append(fiberio) + self._format_version.setdefault(forma, {})[ver] = fiberio + self._fiber_io_by_input_type.setdefault(fiberio.input_type, set()).add(fiberio) self._fiber_io_name_ver.add(id_tuple) # Snapshots derived from the registry are now stale. self._lookup_cache.clear() @@ -655,7 +660,7 @@ def _yield_extensions(self, extension, input_type=None): self.load_plugins() potential_fiberios = self._get_fiber_io_by_input_type(input_type) with self._lock: - extension_fiberios = tuple(self._extension_list[extension]) + extension_fiberios = tuple(self._extension_list.get(extension, ())) for fiber_io in extension_fiberios: if fiber_io in potential_fiberios: yield fiber_io diff --git a/dascore/io/index/catalog.py b/dascore/io/index/catalog.py index 6f026385a..9a658bd9a 100644 --- a/dascore/io/index/catalog.py +++ b/dascore/io/index/catalog.py @@ -18,10 +18,12 @@ import abc import json +import warnings from collections.abc import Mapping, Sequence from dataclasses import dataclass, field, replace from pathlib import Path from threading import RLock +from typing import Any import numpy as np import pandas as pd @@ -435,6 +437,32 @@ def __setstate__(self, state: dict) -> None: self.lock = RLock() +@dataclass +class _RevisionCache: + """ + A value cached against the catalog revision which produced it. + + Reads and writes happen under the revision lock; a value built at an + older revision is simply not returned. + """ + + value: Any = None + revision: int = -1 + + def get(self, revision: int): + """Return the cached value, or None when empty or stale.""" + return self.value if self.revision == revision else None + + def set(self, value, revision: int): + """Store (and return) a value built at the given revision.""" + self.value, self.revision = value, revision + return value + + def clear(self) -> None: + """Drop the cached value.""" + self.value, self.revision = None, -1 + + # sentinel: _view keeps the current order/ids spec unless told otherwise _KEEP = object() @@ -476,10 +504,8 @@ def __init__( self._default_order = default_order self._ids = None if ids is None else tuple(int(x) for x in ids) self._revision = revision or _CatalogRevision() - self._df_cache: pd.DataFrame | None = None - self._df_cache_revision = -1 - self._live_cache: tuple | None = None - self._live_cache_revision = -1 + self._df_cache = _RevisionCache() + self._live_cache = _RevisionCache() # Source records for rebuilding an in-memory backend (set by # __getstate__ so pickled catalogs survive losing the connection). self._rebuild_records: tuple = () @@ -593,6 +619,12 @@ def backend(self): where a brand-new directory index gets its one automatic update. Bootstrapping runs under the revision lock so concurrent first use cannot build (or ingest into) two backends. + + The one-time directory scan stays under the lock as well: it is + what serializes the build, and there is nothing for a blocked + reader to read until it finishes. Once done, ensure_updated is a + flag check. A later explicit update() re-scans an index which is + already usable, so that one runs outside the lock. """ with self._revision.lock: if self._backend is None: @@ -737,10 +769,8 @@ def restrict(self, indices) -> PatchCatalog: def _invalidate(self) -> None: with self._revision.lock: self._revision.value += 1 - self._df_cache = None - self._df_cache_revision = -1 - self._live_cache = None - self._live_cache_revision = -1 + self._df_cache.clear() + self._live_cache.clear() def _cold_live_values(self) -> tuple | None: """ @@ -762,13 +792,12 @@ def _cold_live_values(self) -> tuple | None: ) if not cold: return None - if ( - self._live_cache is None - or self._live_cache_revision != self._revision.value - ): - self._live_cache = tuple(self.resolver.live_entries().values()) - self._live_cache_revision = self._revision.value - return self._live_cache + revision = self._revision.value + if (live := self._live_cache.get(revision)) is None: + live = self._live_cache.set( + tuple(self.resolver.live_entries().values()), revision + ) + return live def __deepcopy__(self, memo) -> PatchCatalog: """ @@ -854,47 +883,45 @@ def to_df(self) -> pd.DataFrame: compares all non-private columns) is not spuriously blocked. """ with self._revision.lock: - if ( - self._df_cache is None - or self._df_cache_revision != self._revision.value - ): - df = self.backend.query( - list(self._queries) or None, - order_by=self._effective_order, - patch_ids=self._ids, + if (cached := self._df_cache.get(self._revision.value)) is not None: + return cached + df = self.backend.query( + list(self._queries) or None, + order_by=self._effective_order, + patch_ids=self._ids, + ) + if self._ids is not None and self._order is None: + # id membership presents in its own (window/array) order + position = {pid: i for i, pid in enumerate(self._ids)} + df = df.sort_values( + "patch_id", key=lambda s: s.map(position), kind="stable" + ).reset_index(drop=True) + df = df.drop(columns=list(SPOOL_HIDDEN_COLUMNS), errors="ignore").rename( + columns={"patch_id": "_patch_id"} + ) + # SQL identifies overlapping source patches. Expose the selected + # envelopes, matching spool.get_contents() and the exact trim + # applied when each patch is materialized. Each pass copies the + # frame, so disjoint-name range sets collapse into one pass. + range_dicts = [ + ranges + for query in self._queries + if ( + ranges := { + name: _envelope_range(value) + for name, value in query.coords.items() + if is_range(value) + } ) - if self._ids is not None and self._order is None: - # id membership presents in its own (window/array) order - position = {pid: i for i, pid in enumerate(self._ids)} - df = df.sort_values( - "patch_id", key=lambda s: s.map(position), kind="stable" - ).reset_index(drop=True) - df = df.drop( - columns=list(SPOOL_HIDDEN_COLUMNS), errors="ignore" - ).rename(columns={"patch_id": "_patch_id"}) - # SQL identifies overlapping source patches. Expose the selected - # envelopes, matching spool.get_contents() and the exact trim - # applied when each patch is materialized. Each pass copies the - # frame, so disjoint-name range sets collapse into one pass. - range_dicts = [ - ranges - for query in self._queries - if ( - ranges := { - name: _envelope_range(value) - for name, value in query.coords.items() - if is_range(value) - } - ) - ] - names = [name for ranges in range_dicts for name in ranges] - if range_dicts and len(set(names)) == len(names): - range_dicts = [{k: v for d in range_dicts for k, v in d.items()}] - for ranges in range_dicts: - df = adjust_segments(df, ignore_bad_kwargs=True, **ranges) - self._df_cache = df - self._df_cache_revision = self._revision.value - return self._df_cache + ] + names = [name for ranges in range_dicts for name in ranges] + if range_dicts and len(set(names)) == len(names): + range_dicts = [{k: v for d in range_dicts for k, v in d.items()}] + for ranges in range_dicts: + df = adjust_segments(df, ignore_bad_kwargs=True, **ranges) + # Re-read the revision: bootstrapping the backend above can + # bump it, and this frame reflects the state after that. + return self._df_cache.set(df, self._revision.value) def __len__(self) -> int: with self._revision.lock: @@ -904,11 +931,8 @@ def __len__(self) -> int: # range residuals only drop patches the SQL candidacy already # excludes and samples/relative residuals never drop patches, so # the count matches len(to_df()) without projecting or pivoting. - if ( - self._df_cache is not None - and self._df_cache_revision == self._revision.value - ): - return len(self._df_cache) + if (df := self._df_cache.get(self._revision.value)) is not None: + return len(df) return self.backend.count(list(self._queries) or None, patch_ids=self._ids) def get_patch(self, index: int) -> dc.Patch: @@ -949,8 +973,27 @@ def resolve_row(self, row: Mapping, extra_trim: Mapping | None = None) -> dc.Pat return apply_exact_residuals(patch, self._residuals) def __iter__(self): - for index in range(len(self)): - yield self.get_patch(index) + """ + Yield every patch under the selection. + + The relation is snapshotted once, then each patch is read and + trimmed outside the revision lock. A patch which cannot be + resolved is skipped with a warning rather than ending iteration + (see #583); indexing with get_patch still raises. + """ + with self._revision.lock: + live = self._cold_live_values() + df = None if live is not None else self.to_df() + if live is not None: + yield from live + return + for index in range(len(df)): + try: + yield self.resolve_row(df.iloc[index].to_dict()) + except MissingPatchError as e: + # Usually a coordinate mismatch trimmed the patch to nothing. + msg = f"Skipping patch at index {index} (see #583): {e}" + warnings.warn(msg, UserWarning, stacklevel=2) # --- mutation (root only) ---------------------------------------------- diff --git a/dascore/utils/namespace.py b/dascore/utils/namespace.py index d5a747fb5..efaf1183d 100644 --- a/dascore/utils/namespace.py +++ b/dascore/utils/namespace.py @@ -47,7 +47,13 @@ def _load_plugin_registry( if not csv_path.exists(): return FrozenDict() df = pd.read_csv(csv_path) - return FrozenDict(zip(df["namespace"], zip(df["package_name"], df["package_url"]))) + return FrozenDict( + zip( + df["namespace"], + zip(df["package_name"], df["package_url"], strict=True), + strict=True, + ) + ) def _pass_to_host_method(func): diff --git a/tests/conftest.py b/tests/conftest.py index 685168222..edfcedba3 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -4,6 +4,7 @@ import os import shutil +import threading import warnings from contextlib import contextmanager, suppress from pathlib import Path @@ -109,6 +110,35 @@ def permanent_config(): return _permanent_config +@pytest.fixture(scope="session") +def run_in_threads(): + """ + Return a helper which runs func(index) in several threads at once. + + A barrier releases every thread together, so concurrency tests do not + need sleeps. The timeouts turn a deadlock into a failure rather than a + hung test run. + """ + + def _run(func, count=4, timeout=60): + barrier = threading.Barrier(count, timeout=timeout) + results = [None] * count + + def worker(index): + barrier.wait() + results[index] = func(index) + + threads = [threading.Thread(target=worker, args=(i,)) for i in range(count)] + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout) + assert not thread.is_alive(), "thread never finished; possible deadlock" + return results + + return _run + + @pytest.fixture(autouse=True) def use_test_config(): """Run tests with debug mode enabled unless overridden locally.""" diff --git a/tests/test_core/test_spool.py b/tests/test_core/test_spool.py index f3c79982a..d842492e5 100644 --- a/tests/test_core/test_spool.py +++ b/tests/test_core/test_spool.py @@ -445,6 +445,20 @@ def test_contents_are_caller_owned(self, random_spool): df["tag"] = "modified" assert (random_spool.get_contents()["tag"] != "modified").all() + def test_contents_owned_under_copy_on_write_warn(self, random_spool): + """The truthy 'warn' setting does not enable copy-on-write isolation.""" + with pd.option_context("mode.copy_on_write", "warn"): + df = random_spool.get_contents() + df["tag"] = "modified" + assert (random_spool.get_contents()["tag"] != "modified").all() + + def test_contents_shallow_copy_when_cow_always_on(self, random_spool, monkeypatch): + """Pandas 3 has copy-on-write always on, so no eager block copy.""" + monkeypatch.setattr(pd, "__version__", "3.0.0") + df = random_spool.get_contents() + assert df is not random_spool._df + assert len(df) == len(random_spool) + class TestSelect: """Tests for selecting/trimming spools.""" @@ -961,11 +975,14 @@ def test_union_of_chunked_spool(self, many_contiguous): def test_iteration_skips_unresolvable_patch(self, monkeypatch): """A patch that fails to resolve is skipped with a #583 warning.""" spool = dc.spool([dc.get_example_patch()]) + # Realize the relation so iteration resolves rows rather than + # serving the live registry (which cannot fail to resolve). + spool.get_contents() - def _raise(_ind): + def _raise(*args, **kwargs): raise MissingPatchError("not available in this session") - monkeypatch.setattr(spool._catalog, "get_patch", _raise) + monkeypatch.setattr(spool._catalog, "resolve_row", _raise) with pytest.warns(UserWarning, match="Skipping patch"): assert list(spool) == [] diff --git a/tests/test_io/test_index/test_catalog.py b/tests/test_io/test_index/test_catalog.py index daba877a7..4c44f8d9c 100644 --- a/tests/test_io/test_index/test_catalog.py +++ b/tests/test_io/test_index/test_catalog.py @@ -3,7 +3,6 @@ from __future__ import annotations import pickle -import threading import numpy as np import pytest @@ -103,7 +102,7 @@ def test_unknown_name_raises_at_select(self, live_catalog): def test_no_sql_at_select(self, live_catalog): """Selection composes without realizing the dataframe.""" view = live_catalog.select(distance=(0, 10)) - assert view._df_cache is None + assert view._df_cache.get(view._revision.value) is None def test_views_cannot_mutate(self, live_catalog, patches): """Mutation only on the root.""" @@ -314,37 +313,21 @@ def test_view_pickles_membership_only(self): class TestCatalogConcurrency: """Catalog caches must stay coherent under concurrent readers.""" - def _run(self, func, count=4): - """Run func(index) in count threads, all released together.""" - barrier = threading.Barrier(count) - results = [None] * count - - def worker(index): - barrier.wait() - results[index] = func(index) - - threads = [threading.Thread(target=worker, args=(i,)) for i in range(count)] - for thread in threads: - thread.start() - for thread in threads: - thread.join() - return results - - def test_concurrent_realization(self, live_catalog, patches): + def test_concurrent_realization(self, live_catalog, patches, run_in_threads): """Racing first realizations build one backend and one relation.""" - results = self._run(lambda _: len(live_catalog.to_df())) + results = run_in_threads(lambda _: len(live_catalog.to_df())) assert set(results) == {len(patches)} # one cached relation, not one per thread assert live_catalog.to_df() is live_catalog.to_df() - def test_concurrent_reads(self, live_catalog, patches): + def test_concurrent_reads(self, live_catalog, patches, run_in_threads): """Mixed len/patch access from several threads agrees.""" def read(index): """Read the catalog a few different ways.""" return (len(live_catalog), live_catalog.get_patch(index).shape) - results = self._run(lambda index: read(index), len(patches)) + results = run_in_threads(read, len(patches)) assert {x[0] for x in results} == {len(patches)} assert [x[1] for x in results] == [x.shape for x in patches] diff --git a/tests/test_io/test_io_core.py b/tests/test_io/test_io_core.py index 5ac4605e3..15e441c75 100644 --- a/tests/test_io/test_io_core.py +++ b/tests/test_io/test_io_core.py @@ -30,6 +30,7 @@ _FiberIOManager, _get_reloadable_source_path, _make_scan_payload, + _reinit_manager_lock, _scan_result_to_summary, _validate_scan_payload, ) @@ -492,6 +493,9 @@ def bad_loader(): manager.__dict__.pop("known_formats", None) # clear the method cache so load_plugins runs again for this instance. manager.__dict__.pop("_cache", None) + # the copy inherits the original's "everything is loaded" state, + # which the swapped-in entry points invalidate. + manager._all_loaded = False return manager def test_load_plugins_warns_and_skips(self, broken_ep_manager): @@ -519,23 +523,7 @@ def _make_manager(self, eps): manager.__dict__["_eps"] = pd.Series(eps) return manager - def _run(self, func, count): - """Run func(index) in count threads, all released together.""" - barrier = threading.Barrier(count) - results = [None] * count - - def worker(index): - barrier.wait() - results[index] = func(index) - - threads = [threading.Thread(target=worker, args=(i,)) for i in range(count)] - for thread in threads: - thread.start() - for thread in threads: - thread.join() - return results - - def test_multi_version_format_never_partial(self): + def test_multi_version_format_never_partial(self, run_in_threads): """Every thread sees all versions, even mid-load.""" entered, release, calls = threading.Event(), threading.Event(), [] @@ -556,7 +544,7 @@ def slow_loader(): # the other threads are queued behind the manager lock by then. releaser = threading.Thread(target=lambda: (entered.wait(), release.set())) releaser.start() - results = self._run( + results = run_in_threads( lambda _: tuple( x.version for x in manager.yield_fiberio(format="_TestFormatter") ), @@ -567,7 +555,7 @@ def slow_loader(): assert all(x == ("2", "1") for x in results) assert calls == ["v1"] - def test_concurrent_full_load_runs_each_loader_once(self): + def test_concurrent_full_load_runs_each_loader_once(self, run_in_threads): """Loading all formats from several threads loads each entry point once.""" calls = [] @@ -586,7 +574,7 @@ def loader(): "_TESTFORMATTER__V2": make_loader(_FiberFormatTestV2), } ) - results = self._run(lambda _: len(list(manager.yield_fiberio())), 4) + results = run_in_threads(lambda _: len(list(manager.yield_fiberio()))) assert set(results) == {2} assert sorted(calls) == ["1", "2"] @@ -604,6 +592,21 @@ def test_copy_gets_own_lock(self): assert copied._lock is not manager._lock assert list(copied.yield_fiberio(format="_TestFormatter")) + def test_fork_handler_replaces_held_lock(self): + """A lock held at fork time is replaced so the child cannot deadlock.""" + manager = FiberIO.manager + old_lock = manager._lock + try: + with old_lock: + _reinit_manager_lock() + new_lock = manager._lock + # The replacement is free even while the old lock is held. + assert new_lock.acquire(blocking=False) + new_lock.release() + assert new_lock is not old_lock + finally: + manager._lock = old_lock + class TestFormatter: """Tests for adding file supports through Formatter.""" diff --git a/tests/test_units.py b/tests/test_units.py index 825bf18fa..b5c0284e5 100644 --- a/tests/test_units.py +++ b/tests/test_units.py @@ -2,8 +2,6 @@ from __future__ import annotations -import threading - import numpy as np import pint import pytest @@ -474,36 +472,34 @@ def test_fractional_percentage(self): class TestUnitConcurrency: """The pint registry must initialize once and parse safely in threads.""" - def _run(self, func, count=4): - """Run func(index) in count threads, all released together.""" - barrier = threading.Barrier(count) - results = [None] * count - - def worker(index): - barrier.wait() - results[index] = func(index) - - threads = [threading.Thread(target=worker, args=(i,)) for i in range(count)] - for thread in threads: - thread.start() - for thread in threads: - thread.join() - return results - - def test_registry_created_once(self, monkeypatch): + def test_registry_created_once(self, monkeypatch, run_in_threads): """Racing threads all get the same registry instance.""" # The registry is process-wide; restore it so quantities created by # other tests keep belonging to the active registry. monkeypatch.setattr(units_module, "_UNIT_REGISTRY", None) original = pint.get_application_registry().get() try: - results = self._run(lambda _: units_module.get_registry()) + results = run_in_threads(lambda _: units_module.get_registry()) finally: pint.set_application_registry(original) assert len({id(x) for x in results}) == 1 - def test_concurrent_parsing(self): + def test_concurrent_parsing(self, run_in_threads): """Parsing distinct quantities in threads returns correct values.""" strings = ["m/s", "1/s", "furlong/fortnight", "strain"] - results = self._run(lambda index: get_quantity(strings[index])) + results = run_in_threads(lambda index: get_quantity(strings[index])) assert results == [get_quantity(x) for x in strings] + + def test_fork_handler_replaces_held_lock(self): + """A lock held at fork time is replaced so the child cannot deadlock.""" + old_lock = units_module._UNIT_LOCK + try: + with old_lock: + units_module._reinit_unit_lock() + new_lock = units_module._UNIT_LOCK + # The replacement is free even while the old lock is held. + assert new_lock.acquire(blocking=False) + new_lock.release() + assert new_lock is not old_lock + finally: + units_module._UNIT_LOCK = old_lock diff --git a/tests/test_utils/test_namespace.py b/tests/test_utils/test_namespace.py index a843c67b5..7bc730baa 100644 --- a/tests/test_utils/test_namespace.py +++ b/tests/test_utils/test_namespace.py @@ -2,7 +2,6 @@ from __future__ import annotations -import threading from typing import ClassVar import pandas as pd @@ -264,36 +263,20 @@ def test_getattr_unknown_attr_raises_default_error(self, monkeypatch, tmp_path): class TestNamespaceConcurrency: """Lazy namespace attachment must be safe under concurrent first use.""" - def _run(self, func, count=4): - """Run func(index) in count threads, all released together.""" - barrier = threading.Barrier(count) - results = [None] * count - - def worker(index): - barrier.wait() - results[index] = func(index) - - threads = [threading.Thread(target=worker, args=(i,)) for i in range(count)] - for thread in threads: - thread.start() - for thread in threads: - thread.join() - return results - - def test_one_instance_per_host(self): + def test_one_instance_per_host(self, run_in_threads): """Concurrent first access on one object returns a single namespace.""" inst = ParentClass() - results = self._run(lambda _: inst.bob) + results = run_in_threads(lambda _: inst.bob) assert len({id(x) for x in results}) == 1 assert results[0] is inst.bob - def test_distinct_hosts_get_own_namespace(self): + def test_distinct_hosts_get_own_namespace(self, run_in_threads): """Concurrent first access on distinct objects binds each host.""" instances = [ParentClass() for _ in range(4)] - results = self._run(lambda index: instances[index].bob) + results = run_in_threads(lambda index: instances[index].bob) assert [x.return_self() for x in results] == instances - def test_concurrent_registration_and_lookup(self): + def test_concurrent_registration_and_lookup(self, run_in_threads): """Registering namespaces while reading the registry stays consistent.""" class ConcurrentBase(_MethodNameSpace): @@ -302,8 +285,25 @@ class ConcurrentBase(_MethodNameSpace): def register(index): """Define a new namespace and read the registry back.""" type(f"Namespace{index}", (ConcurrentBase,), {"name": f"ns_{index}"}) - return set(_MethodNameSpace._registry["dascore.concurrent_test"]) + # Snapshot under the lock; sibling threads are inserting. + with _MethodNameSpace._registry_lock: + return set(_MethodNameSpace._registry["dascore.concurrent_test"]) - self._run(register) + run_in_threads(register) registered = _MethodNameSpace._registry["dascore.concurrent_test"] assert set(registered) == {f"ns_{i}" for i in range(4)} + + def test_fork_handler_replaces_held_locks(self): + """Locks held at fork time are replaced so the child cannot deadlock.""" + old_attachment = ns_module._ATTACHMENT_LOCK + old_registry = _MethodNameSpace._registry_lock + try: + with old_attachment, old_registry: + ns_module._reinit_namespace_locks() + new_attachment = ns_module._ATTACHMENT_LOCK + new_registry = _MethodNameSpace._registry_lock + assert new_attachment is not old_attachment + assert new_registry is not old_registry + finally: + ns_module._ATTACHMENT_LOCK = old_attachment + _MethodNameSpace._registry_lock = old_registry From 449c569ee50792efa0fd0756f6e71de7d9c07320 Mon Sep 17 00:00:00 2001 From: Derrick Chambers Date: Sat, 25 Jul 2026 15:13:12 +0200 Subject: [PATCH 3/3] Decide copy-on-write by version once, not per call Reading pd.options.mode.copy_on_write emits a deprecation warning on every access under pandas 3, where copy-on-write can no longer be disabled. Settle that at import from the version and only consult the option on pandas 2, so get_contents() stays quiet on pandas 3 and the helper keeps a single, always-executed body (the previous version branch was the one line codecov reported as uncovered). The copy-mode test is now parametrized over every setting pandas 2 accepts and skips where the option no longer applies. --- dascore/core/spool.py | 18 +++++++++++------- tests/test_core/test_spool.py | 21 ++++++++++----------- 2 files changed, 21 insertions(+), 18 deletions(-) diff --git a/dascore/core/spool.py b/dascore/core/spool.py index 07529b56b..3781250b5 100644 --- a/dascore/core/spool.py +++ b/dascore/core/spool.py @@ -47,18 +47,22 @@ T = TypeVar("T") +# Copy-on-write is always on from pandas 3, which also deprecates the +# option: reading it there warns on every access, so settle it by version +# once and only consult the option on pandas 2. +_COPY_ON_WRITE_ALWAYS = int(pd.__version__.split(".", maxsplit=1)[0]) >= 3 + + def _copy_public_dataframe(frame: pd.DataFrame) -> pd.DataFrame: """ Return a caller-owned view of an internally cached dataframe. - With copy-on-write (always on in pandas 3) a shallow copy already - detaches on the first write; without it the blocks must be copied. - Only the literal True enables isolation: the "warn" setting keeps the - old (sharing) semantics, so it still needs a deep copy. + Copy-on-write makes a shallow copy enough, since the frames detach on + the first write. Only the literal True enables it; pandas 2 also + accepts "warn", which keeps the old sharing semantics. """ - if int(pd.__version__.split(".", maxsplit=1)[0]) >= 3: - return frame.copy(deep=False) - return frame.copy(deep=pd.options.mode.copy_on_write is not True) + copy_on_write = _COPY_ON_WRITE_ALWAYS or pd.options.mode.copy_on_write is True + return frame.copy(deep=not copy_on_write) class BaseSpool(NamespaceOwner, abc.ABC): diff --git a/tests/test_core/test_spool.py b/tests/test_core/test_spool.py index d842492e5..45320cef7 100644 --- a/tests/test_core/test_spool.py +++ b/tests/test_core/test_spool.py @@ -11,7 +11,7 @@ import pytest import dascore as dc -from dascore.core.spool import BaseSpool, Spool +from dascore.core.spool import _COPY_ON_WRITE_ALWAYS, BaseSpool, Spool from dascore.exceptions import ( InvalidSpoolError, MissingOptionalDependencyError, @@ -445,20 +445,19 @@ def test_contents_are_caller_owned(self, random_spool): df["tag"] = "modified" assert (random_spool.get_contents()["tag"] != "modified").all() - def test_contents_owned_under_copy_on_write_warn(self, random_spool): - """The truthy 'warn' setting does not enable copy-on-write isolation.""" - with pd.option_context("mode.copy_on_write", "warn"): + @pytest.mark.parametrize("copy_on_write", [False, True, "warn"]) + def test_contents_owned_in_every_copy_mode(self, random_spool, copy_on_write): + """Ownership holds for each copy-on-write setting pandas 2 allows. + + Notably "warn" is truthy but keeps the old sharing semantics. + """ + if _COPY_ON_WRITE_ALWAYS: + pytest.skip("pandas 3 has copy-on-write always on") + with pd.option_context("mode.copy_on_write", copy_on_write): df = random_spool.get_contents() df["tag"] = "modified" assert (random_spool.get_contents()["tag"] != "modified").all() - def test_contents_shallow_copy_when_cow_always_on(self, random_spool, monkeypatch): - """Pandas 3 has copy-on-write always on, so no eager block copy.""" - monkeypatch.setattr(pd, "__version__", "3.0.0") - df = random_spool.get_contents() - assert df is not random_spool._df - assert len(df) == len(random_spool) - class TestSelect: """Tests for selecting/trimming spools."""