ENH: synchronize registries, units, and catalogs for free-threading

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

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

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

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
@@ -48,6 +47,24 @@
 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.
+
+    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.
+    """
+    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):
     """Spool Abstract Base Class (ABC) for defining Spool interface."""
 
@@ -260,6 +277,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 +502,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
@@ -514,14 +537,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 -------------------------------
 

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,38 @@ 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()
+        # 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] = {}
+
+    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 +426,43 @@ 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
-
-    @cache
-    def _get_prioritized_list(self, input_type="file"):
+        key = ("input_type", input_type)
+        if (cached := self._lookup_cache.get(key)) is None:
+            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
+            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."""
+        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:
-            unsorted = self._format_version[format_name]
-            if not unsorted:
+            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 +474,43 @@ 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 (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
-        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
+        with self._lock:
+            # 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)
+                    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 formats.isdisjoint(self.unloaded_formats)
 
     def _load_entry_point(self, name: str, loader) -> FiberIO | None:
         """Load one FiberIO entry point, skipping broken registrations."""
@@ -474,6 +527,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
@@ -482,10 +536,12 @@ 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()
 
     @cached_method
     def get_fiberio(
@@ -574,7 +630,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 +659,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.get(extension, ()))
+        for fiber_io in extension_fiberios:
             if fiber_io in potential_fiberios:
                 yield fiber_io
             has_yielded.add(fiber_io)
@@ -931,6 +991,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,