Fix CombinedLock deadlocks & flaky tests

.github/workflows/ci.yaml

@@ -64,6 +64,8 @@ jobs:
     name: "${{ matrix.os }} | ${{ matrix.pixi-env }}${{ matrix.pytest-addopts && format(' ({0})', matrix.pytest-addopts) || '' }}"
     runs-on: ${{ matrix.os }}
     needs: cache-pixi-lock
+    # The slowest jobs take ~20 minutes; a hung test should not burn the 6h default.
+    timeout-minutes: 60
     defaults:
       run:
         shell: bash -l {0}
@@ -131,7 +133,9 @@ jobs:
           then
             if [[ "${{ matrix.pytest-addopts }}" == "flaky" ]] ;
             then
-              echo "PYTEST_ADDOPTS=-m 'flaky or network' --run-flaky --run-network-tests -W default" >> $GITHUB_ENV
+              # The thread timeout method also kills tests stuck in C code or deadlocked
+              # threads, which the default signal method cannot interrupt.
+              echo "PYTEST_ADDOPTS=-m 'flaky or network' --run-flaky --run-network-tests --timeout-method=thread -W default" >> $GITHUB_ENV
             elif [[ "${{ matrix.pytest-addopts }}" == "mypy" ]] ;
             then
               echo "PYTEST_ADDOPTS=-n 1 -m 'mypy' --run-mypy -W default" >> $GITHUB_ENV

doc/whats-new.rst

@@ -53,9 +53,20 @@ Deprecations
 Bug Fixes
 ~~~~~~~~~
 
+- Fix deadlocks when reading and writing netCDF files with dask at the same
+  time. Combined locks now always acquire their locks in the same order, which
+  previously depended on memory addresses, and a failed non-blocking acquire,
+  e.g. while garbage collecting an unclosed file, no longer leaves some of its
+  locks held forever (:pull:`11622`).
+  By `Michael Niklas <https://github.com/headtr1ck>`_.
 - Fix :py:func:`broadcast` failing on objects with an index spanning several
   dimensions, such as a custom index set on both ``x`` and ``y`` (:pull:`11615`).
   By `Matthias Schabel <https://github.com/matthiasschabel>`_.
+- Fix errors and crashes in :py:func:`open_mfdataset` with ``parallel=True``
+  when opening more files than ``file_cache_maxsize``. Files evicted from the
+  file cache by another thread are no longer closed while they are still being
+  read (:pull:`11622`).
+  By `Michael Niklas <https://github.com/headtr1ck>`_.
 - :py:meth:`DataArray.roll` and :py:meth:`Dataset.roll` now return unchanged
   empty results when rolling an empty dimension, including with
   ``roll_coords=True``, instead of raising ``ZeroDivisionError`` (:pull:`11613`).
@@ -64,7 +75,6 @@ Bug Fixes
   index instead of an empty index when the interval direction conflicts with
   the sign of ``step`` (:pull:`11623`).
   By `Ahmet Kamer Çivi <https://github.com/lowgame>`_.
-
 - Fix async zarr tests using ``wraps`` with ``autospec=True`` on async methods,
   which caused ``AsyncMock`` objects to leak through instead of real array data
   (:pull:`11232`).

xarray/backends/file_manager.py

@@ -14,12 +14,35 @@
 from xarray.core.options import OPTIONS
 from xarray.core.types import Closable, Lock
 
+# Files are pinned while in use inside CachingFileManager.acquire_context().
+# A pinned file evicted from FILE_CACHE by another thread is not closed right
+# away, as that thread may still be reading from it. Instead it is parked in
+# _EVICTED_PINNED until it is acquired again or the last user unpins it.
+_PIN_COUNTS: dict[Any, int] = {}
+_EVICTED_PINNED: dict[Any, Closable] = {}
+
+
+def _close_unless_pinned(key: Any, file: Closable) -> None:
+    with _PIN_LOCK:
+        if _PIN_COUNTS.get(key):
+            _EVICTED_PINNED[key] = file
+            return
+    file.close()
+
+
 # Global cache for storing open files.
 FILE_CACHE: LRUCache[Any, Closable] = LRUCache(
-    maxsize=OPTIONS["file_cache_maxsize"], on_evict=lambda k, v: v.close()
+    maxsize=OPTIONS["file_cache_maxsize"], on_evict=_close_unless_pinned
 )
 assert FILE_CACHE.maxsize, "file cache must be at least size one"
 
+# Guards the pin state. It is the reentrant lock of FILE_CACHE, as evicting a
+# file from the cache updates the pin state while holding the cache's lock, and
+# garbage collection can run a manager's __del__, which closes its file and
+# removes it from the cache, while a thread holds either. Separate locks could
+# deadlock there.
+_PIN_LOCK = FILE_CACHE._lock
+
 T_File = TypeVar("T_File", bound=Closable)
 
 REF_COUNTS: dict[Any, int] = {}
@@ -203,21 +226,51 @@ def acquire(self, needs_lock: bool = True) -> T_File:
 
     @contextmanager
     def acquire_context(self, needs_lock: bool = True) -> Iterator[T_File]:
-        """Context manager for acquiring a file."""
-        file, cached = self._acquire_with_cache_info(needs_lock)
+        """Context manager for acquiring a file.
+
+        The file stays open until the context exits, even if it is evicted
+        from the cache in the meantime.
+        """
+        with self._pinned():
+            file, cached = self._acquire_with_cache_info(needs_lock)
+            try:
+                yield file
+            except Exception:
+                if not cached:
+                    self.close(needs_lock)
+                raise
+
+    @contextmanager
+    def _pinned(self) -> Iterator[None]:
+        """Keep the file open while in use, even if it is evicted meanwhile."""
+        with _PIN_LOCK:
+            _PIN_COUNTS[self._key] = _PIN_COUNTS.get(self._key, 0) + 1
         try:
-            yield file
-        except Exception:
-            if not cached:
-                self.close(needs_lock)
-            raise
+            yield
+        finally:
+            with _PIN_LOCK:
+                count = _PIN_COUNTS.pop(self._key) - 1
+                if count:
+                    _PIN_COUNTS[self._key] = count
+                    file = None
+                else:
+                    file = _EVICTED_PINNED.pop(self._key, None)
+            if file is not None:
+                file.close()
 
     def _acquire_with_cache_info(self, needs_lock: bool = True) -> tuple[T_File, bool]:
         """Acquire a file, returning the file and whether it was cached."""
         with self._optional_lock(needs_lock):
             try:
                 file = self._cache[self._key]
             except KeyError:
+                with _PIN_LOCK:
+                    evicted = _EVICTED_PINNED.pop(self._key, None)
+                if evicted is not None:
+                    # still open because it is in use, so reuse it
+                    file = cast(T_File, evicted)
+                    self._cache[self._key] = file
+                    return file, True
                 kwargs = self._kwargs
                 if self._mode is not _OMIT_MODE:
                     kwargs = kwargs.copy()
@@ -238,8 +291,11 @@ def close(self, needs_lock: bool = True) -> None:
         with self._optional_lock(needs_lock):
             default = None
             file = self._cache.pop(self._key, default)
-            if file is not None:
-                file.close()
+            with _PIN_LOCK:
+                evicted = _EVICTED_PINNED.pop(self._key, None)
+            for f in (file, evicted):
+                if f is not None:
+                    f.close()
 
     def __del__(self) -> None:
         # If we're the only CachingFileManger referencing an unclosed file,

xarray/backends/h5netcdf_.py

@@ -149,7 +149,8 @@ def __init__(
         self.format = format or "NETCDF4"
         # todo: utilizing find_root_and_group seems a bit clunky
         #  making filename available on h5netcdf.Group seems better
-        self._filename = find_root_and_group(self.ds)[0].filename
+        with manager.acquire_context():
+            self._filename = find_root_and_group(self.ds)[0].filename
         self.is_remote = is_remote_uri(self._filename)
         self.lock = ensure_lock(lock)
         self.autoclose = autoclose
@@ -570,16 +571,17 @@ def open_dataset(
 
         store_entrypoint = StoreBackendEntrypoint()
 
-        ds = store_entrypoint.open_dataset(
-            store,
-            mask_and_scale=mask_and_scale,
-            decode_times=decode_times,
-            concat_characters=concat_characters,
-            decode_coords=decode_coords,
-            drop_variables=drop_variables,
-            use_cftime=use_cftime,
-            decode_timedelta=decode_timedelta,
-        )
+        with store._manager.acquire_context():
+            ds = store_entrypoint.open_dataset(
+                store,
+                mask_and_scale=mask_and_scale,
+                decode_times=decode_times,
+                concat_characters=concat_characters,
+                decode_coords=decode_coords,
+                drop_variables=drop_variables,
+                use_cftime=use_cftime,
+                decode_timedelta=decode_timedelta,
+            )
 
         # only warn if phony_dims exist in file
         # remove together with the above check
@@ -686,26 +688,27 @@ def open_groups_as_dict(
 
         manager = store._manager
         groups_dict = {}
-        for path_group in _iter_nc_groups(store.ds, parent=parent):
-            group_store = H5NetCDFStore(manager, group=path_group, **kwargs)
-            store_entrypoint = StoreBackendEntrypoint()
-            with close_on_error(group_store):
-                group_ds = store_entrypoint.open_dataset(
-                    group_store,
-                    mask_and_scale=mask_and_scale,
-                    decode_times=decode_times,
-                    concat_characters=concat_characters,
-                    decode_coords=decode_coords,
-                    drop_variables=drop_variables,
-                    use_cftime=use_cftime,
-                    decode_timedelta=decode_timedelta,
-                )
+        with manager.acquire_context():
+            for path_group in _iter_nc_groups(store.ds, parent=parent):
+                group_store = H5NetCDFStore(manager, group=path_group, **kwargs)
+                store_entrypoint = StoreBackendEntrypoint()
+                with close_on_error(group_store):
+                    group_ds = store_entrypoint.open_dataset(
+                        group_store,
+                        mask_and_scale=mask_and_scale,
+                        decode_times=decode_times,
+                        concat_characters=concat_characters,
+                        decode_coords=decode_coords,
+                        drop_variables=drop_variables,
+                        use_cftime=use_cftime,
+                        decode_timedelta=decode_timedelta,
+                    )
 
-            if group:
-                group_name = str(NodePath(path_group).relative_to(parent))
-            else:
-                group_name = str(NodePath(path_group))
-            groups_dict[group_name] = group_ds
+                if group:
+                    group_name = str(NodePath(path_group).relative_to(parent))
+                else:
+                    group_name = str(NodePath(path_group))
+                groups_dict[group_name] = group_ds
 
         # only warn if phony_dims exist in file
         # remove together with the above check

xarray/backends/locks.py

@@ -207,6 +207,16 @@ def acquire(lock, blocking=True):
         return lock.acquire(blocking)
 
 
+def _lock_identity(lock: Lock) -> int:
+    """Identity of the lock that is actually acquired.
+
+    Unpickled SerializableLocks are new objects wrapping the same threading.Lock.
+    """
+    if isinstance(lock, SerializableLock):
+        return id(lock.lock)
+    return id(lock)
+
+
 class CombinedLock(Lock):
     """A combination of multiple locks.
 
@@ -215,22 +225,38 @@ class CombinedLock(Lock):
     """
 
     def __init__(self, locks: Sequence[Lock]):
-        self.locks = tuple(set(locks))  # remove duplicates
+        # Remove duplicates and always acquire in one global order. If not careful,
+        # CombinedLocks sharing locks could acquire them in opposite orders and
+        # deadlock each other.
+        unique = {_lock_identity(lock): lock for lock in locks}
+        self.locks = tuple(lock for _, lock in sorted(unique.items()))
+
+    def __reduce__(self):
+        # The order depends on the ids of the locks, which differ between
+        # processes, so sort again when unpickling.
+        return (type(self), (list(self.locks),))
 
     def acquire(self, blocking=True):
-        return all(acquire(lock, blocking=blocking) for lock in self.locks)
+        acquired = []
+        for lock in self.locks:
+            if not acquire(lock, blocking=blocking):
+                # Release the locks we already hold, otherwise a failed
+                # non-blocking acquire leaves them locked forever.
+                for held in reversed(acquired):
+                    held.release()
+                return False
+            acquired.append(lock)
+        return True
 
     def release(self):
-        for lock in self.locks:
+        for lock in reversed(self.locks):
             lock.release()
 
     def __enter__(self):
-        for lock in self.locks:
-            lock.__enter__()
+        self.acquire()
 
     def __exit__(self, *args):
-        for lock in self.locks:
-            lock.__exit__(*args)
+        self.release()
 
     def locked(self):
         return any(lock.locked() for lock in self.locks)

xarray/backends/netCDF4_.py

@@ -451,8 +451,9 @@ def __init__(
         self._manager = manager
         self._group = group
         self._mode = mode
-        self.format = self.ds.data_model
-        self._filename = self.ds.filepath()
+        with manager.acquire_context():
+            self.format = self.ds.data_model
+            self._filename = self.ds.filepath()
         self.is_remote = is_remote_uri(self._filename)
         self.lock = ensure_lock(lock)
         self.autoclose = autoclose
@@ -807,7 +808,7 @@ def open_dataset(
         )
 
         store_entrypoint = StoreBackendEntrypoint()
-        with close_on_error(store):
+        with close_on_error(store), store._manager.acquire_context():
             ds = store_entrypoint.open_dataset(
                 store,
                 mask_and_scale=mask_and_scale,
@@ -908,25 +909,26 @@ def open_groups_as_dict(
 
         manager = store._manager
         groups_dict = {}
-        for path_group in _iter_nc_groups(store.ds, parent=parent):
-            group_store = NetCDF4DataStore(manager, group=path_group, **kwargs)
-            store_entrypoint = StoreBackendEntrypoint()
-            with close_on_error(group_store):
-                group_ds = store_entrypoint.open_dataset(
-                    group_store,
-                    mask_and_scale=mask_and_scale,
-                    decode_times=decode_times,
-                    concat_characters=concat_characters,
-                    decode_coords=decode_coords,
-                    drop_variables=drop_variables,
-                    use_cftime=use_cftime,
-                    decode_timedelta=decode_timedelta,
-                )
-            if group:
-                group_name = str(NodePath(path_group).relative_to(parent))
-            else:
-                group_name = str(NodePath(path_group))
-            groups_dict[group_name] = group_ds
+        with manager.acquire_context():
+            for path_group in _iter_nc_groups(store.ds, parent=parent):
+                group_store = NetCDF4DataStore(manager, group=path_group, **kwargs)
+                store_entrypoint = StoreBackendEntrypoint()
+                with close_on_error(group_store):
+                    group_ds = store_entrypoint.open_dataset(
+                        group_store,
+                        mask_and_scale=mask_and_scale,
+                        decode_times=decode_times,
+                        concat_characters=concat_characters,
+                        decode_coords=decode_coords,
+                        drop_variables=drop_variables,
+                        use_cftime=use_cftime,
+                        decode_timedelta=decode_timedelta,
+                    )
+                if group:
+                    group_name = str(NodePath(path_group).relative_to(parent))
+                else:
+                    group_name = str(NodePath(path_group))
+                groups_dict[group_name] = group_ds
 
         return groups_dict
 

xarray/backends/scipy_.py

@@ -446,7 +446,7 @@ def open_dataset(
         )
 
         store_entrypoint = StoreBackendEntrypoint()
-        with close_on_error(store):
+        with close_on_error(store), store._manager.acquire_context():
             ds = store_entrypoint.open_dataset(
                 store,
                 mask_and_scale=mask_and_scale,