Retry HDF5 opens on transient file-lock failures (fixes zoomify --balance -p N>1 on NFS)

CHANGES.md

@@ -1,5 +1,21 @@
 # Release notes #
 
+## [Unreleased]
+
+### Bug fixes
+* fix: Retry `h5py.File` opens on `BlockingIOError` in the `create`/`coarsen`/`balance`
+  write paths, to tolerate transient HDF5 file-locking failures on some
+  NFS-mounted filesystems (e.g. `cooler zoomify --balance -p N>1`). The
+  `zoomify`/`balance` help text now also documents the
+  `HDF5_USE_FILE_LOCKING=FALSE` environment variable as a workaround.
+  Fixes #486.
+  * Note for future work: the retry is a targeted mitigation, not a fix for
+    the underlying cause of the lock churn. `write_pixels()` in
+    `cooler/create/_create.py` opens and closes the output file once per
+    pixel chunk; holding a single file handle open across the whole write
+    loop instead would reduce lock-acquisition frequency at the source and
+    is a more thorough (but riskier, performance-path-touching) follow-up.
+
 ## [v0.10.4](https://github.com/open2c/cooler/compare/v0.10.3...v0.10.4)
 
 ### Bug fixes

src/cooler/cli/balance.py

@@ -1,14 +1,13 @@
 import sys
 
 import click
-import h5py
 import numpy as np
 import pandas as pd
 from multiprocess import Pool
 
 from .._balance import balance_cooler
 from ..api import Cooler
-from ..util import bedslice, parse_cooler_uri
+from ..util import bedslice, open_hdf5_with_retry, parse_cooler_uri
 from . import cli, get_logger
 
 
@@ -173,12 +172,16 @@ def balance(
 
     COOL_PATH : Path to a COOL file.
 
+    If you encounter HDF5 file-locking errors (``BlockingIOError``) on a
+    networked/NFS-mounted filesystem, try setting the environment variable
+    ``HDF5_USE_FILE_LOCKING=FALSE`` before running cooler.
+
     """
     logger = get_logger(__name__)
     cool_path, group_path = parse_cooler_uri(cool_uri)
 
     if check:
-        with h5py.File(cool_path, "r") as h5:
+        with open_hdf5_with_retry(cool_path, "r") as h5:
             grp = h5[group_path]
             if name not in grp["bins"]:
                 click.echo(f"{cool_path}: No '{name}' column found.")
@@ -192,7 +195,7 @@ def balance(
             "Provide at most one of --cis-only and --trans-only flags"
         )
 
-    with h5py.File(cool_path, "r+") as h5:
+    with open_hdf5_with_retry(cool_path, "r+") as h5:
         grp = h5[group_path]
         if name in grp["bins"] and not stdout:
             if not force:
@@ -281,7 +284,7 @@ def balance(
             sys.stdout, header=False, index=False, na_rep="", float_format="%g"
         )
     else:
-        with h5py.File(cool_path, "r+") as h5:
+        with open_hdf5_with_retry(cool_path, "r+") as h5:
             grp = h5[group_path]
             # add the bias column to the file
             h5opts = {"compression": "gzip", "compression_opts": 6}

src/cooler/cli/zoomify.py

@@ -128,6 +128,10 @@ def zoomify(
 
     COOL_PATH : Path to a COOL file or Cooler URI.
 
+    If you encounter HDF5 file-locking errors (``BlockingIOError``) on a
+    networked/NFS-mounted filesystem, try setting the environment variable
+    ``HDF5_USE_FILE_LOCKING=FALSE`` before running cooler.
+
     """
     logger = get_logger(__name__)
     infile, _ = parse_cooler_uri(cool_uri)

src/cooler/create/_create.py

@@ -23,6 +23,7 @@
     get_chromsizes,
     get_meta,
     infer_meta,
+    open_hdf5_with_retry,
     parse_cooler_uri,
     rlencode,
 )
@@ -243,7 +244,16 @@ def write_pixels(
 
             logger.debug(f"writing chunk {i}")
 
-            with h5py.File(filepath, "r+") as fw:
+            # This opens and closes the file once per chunk, which on an
+            # NFS-mounted file means one lock-acquire/release cycle per
+            # chunk -- for the finest/largest resolution (most chunks) this
+            # is the actual source of the transient BlockingIOErrors that
+            # open_hdf5_with_retry works around below. Holding a single
+            # open handle across the whole write loop instead (still
+            # bounded by `lock`) would cut lock-acquisition frequency at
+            # the source rather than just tolerating its failures -- left
+            # as follow-up work (see CHANGES.md).
+            with open_hdf5_with_retry(filepath, "r+") as fw:
                 grp = fw[grouppath]
                 dsets = [grp[col] for col in columns]
 
@@ -609,7 +619,7 @@ def create(
         iterable = map(validator, iterable)
 
     # Create root group
-    with h5py.File(file_path, mode) as f:
+    with open_hdf5_with_retry(file_path, mode) as f:
         logger.info(f'Creating cooler at "{file_path}::{group_path}"')
         if group_path == "/":
             for name in ["chroms", "bins", "pixels", "indexes"]:
@@ -627,7 +637,10 @@ def create(
         src_path, _src_group = parse_cooler_uri(scool_root_uri)
         dst_path, dst_group = parse_cooler_uri(cool_uri)
 
-        with h5py.File(src_path, "r+") as src, h5py.File(dst_path, "r+") as dst:
+        with (
+            open_hdf5_with_retry(src_path, "r+") as src,
+            open_hdf5_with_retry(dst_path, "r+") as dst,
+        ):
             dst[dst_group]["chroms"] = src["chroms"]
 
             # hard link to root bins table, but only the three main datasets
@@ -642,7 +655,7 @@ def create(
                 columns.remove(col)
             if columns:
                 put(dst[dst_group]["bins"], bins[columns])
-        with h5py.File(file_path, "r+") as f:
+        with open_hdf5_with_retry(file_path, "r+") as f:
             h5 = f[group_path]
             grp = h5.create_group("pixels")
             if symmetric_upper:
@@ -653,7 +666,7 @@ def create(
                 grp, n_bins, max_size, meta.columns, dict(meta.dtypes), h5opts
             )
     else:
-        with h5py.File(file_path, "r+") as f:
+        with open_hdf5_with_retry(file_path, "r+") as f:
             h5 = f[group_path]
 
             logger.info("Writing chroms")
@@ -687,7 +700,7 @@ def create(
     )
 
     # Write indexes
-    with h5py.File(file_path, "r+") as f:
+    with open_hdf5_with_retry(file_path, "r+") as f:
         h5 = f[group_path]
 
         logger.info("Writing indexes")

src/cooler/util.py

@@ -2,6 +2,7 @@
 
 import os
 import re
+import time
 from collections import OrderedDict, defaultdict
 from collections.abc import Generator, Iterable, Iterator
 from contextlib import contextmanager
@@ -15,6 +16,43 @@
 from ._typing import GenomicRangeSpecifier, GenomicRangeTuple
 
 
+def open_hdf5_with_retry(
+    filepath: str, mode: str = "r", retries: int = 5, delay: float = 0.2
+) -> h5py.File:
+    """Open an HDF5 file, retrying on transient OS-level lock failures.
+
+    Drop-in replacement for ``h5py.File(filepath, mode)``. On some
+    NFS-mounted filesystems, the ``flock()`` call HDF5 issues to acquire
+    its file lock can fail transiently with ``BlockingIOError`` even when
+    no other cooler process is actually writing to the file at that
+    instant -- this is a known NFS/HDF5 locking compatibility issue, not a
+    sign of a real read/write race (cooler's own multiprocessing lock,
+    see :mod:`cooler.parallel`, already prevents those). Retrying with
+    exponential backoff is enough to ride out the transient failure; the
+    original error is re-raised if all attempts are exhausted.
+
+    Parameters
+    ----------
+    filepath : str
+        Path to the HDF5 file.
+    mode : str, optional
+        File mode, as accepted by ``h5py.File``.
+    retries : int, optional
+        Maximum number of attempts before giving up.
+    delay : float, optional
+        Base delay in seconds before the first retry; doubles after each
+        subsequent failed attempt.
+
+    """
+    for attempt in range(retries):
+        try:
+            return h5py.File(filepath, mode)
+        except BlockingIOError:
+            if attempt == retries - 1:
+                raise
+            time.sleep(delay * 2**attempt)
+
+
 def partition(start: int, stop: int, step: int) -> Iterator[tuple[int, int]]:
     """Partition an integer interval into equally-sized subintervals.
     Like builtin :py:func:`range`, but yields pairs of end points.

tests/test_util.py

@@ -26,6 +26,41 @@ def test_buffered():
     assert len(next(it)) == 3
 
 
+def test_open_hdf5_with_retry(tmp_path, monkeypatch):
+    path = tmp_path / "test.h5"
+    with h5py.File(path, "w") as f:
+        f.create_dataset("x", data=[1, 2, 3])
+
+    real_open = h5py.File
+    calls = {"n": 0}
+
+    def flaky_open(filepath, mode):
+        calls["n"] += 1
+        if calls["n"] < 3:
+            raise BlockingIOError(11, "Resource temporarily unavailable")
+        return real_open(filepath, mode)
+
+    monkeypatch.setattr(util.h5py, "File", flaky_open)
+
+    with util.open_hdf5_with_retry(str(path), "r", retries=5, delay=0) as f:
+        assert list(f["x"][:]) == [1, 2, 3]
+    assert calls["n"] == 3
+
+
+def test_open_hdf5_with_retry_exhausted(tmp_path, monkeypatch):
+    path = tmp_path / "test.h5"
+    with h5py.File(path, "w"):
+        pass
+
+    def always_fails(filepath, mode):
+        raise BlockingIOError(11, "Resource temporarily unavailable")
+
+    monkeypatch.setattr(util.h5py, "File", always_fails)
+
+    with pytest.raises(BlockingIOError):
+        util.open_hdf5_with_retry(str(path), "r", retries=3, delay=0)
+
+
 def test_rlencode():
     s, l, v = util.rlencode([1, 1, 1, 1, 5, 5, 5, 5, 3, 3, 8, 9, 9])  # noqa
     assert list(s) == [0, 4, 8, 10, 11]