Skip to main content
`Ctrl`+`K`

# Source code for silx.io.h5py_utils

    # /*##########################################################################
    # Copyright (C) 2016-2024 European Synchrotron Radiation Facility
    #
    # Permission is hereby granted, free of charge, to any person obtaining a copy
    # of this software and associated documentation files (the "Software"), to deal
    # in the Software without restriction, including without limitation the rights
    # to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
    # copies of the Software, and to permit persons to whom the Software is
    # furnished to do so, subject to the following conditions:
    #
    # The above copyright notice and this permission notice shall be included in
    # all copies or substantial portions of the Software.
    #
    # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
    # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
    # FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
    # AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
    # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
    # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
    # THE SOFTWARE.
    #
    # ############################################################################*/
    from __future__ import annotations
    """
    This module provides utility methods on top of h5py, mainly to handle
    parallel writing and reading.
    """

    __authors__ = ["W. de Nolf"]
    __license__ = "MIT"
    __date__ = "28/11/2023"

    import os
    import sys
    import traceback
    import logging
    from typing import Sequence
    import h5py

    from .._version import calc_hexversion
    from ..utils import retry as retry_mod
    from silx.utils.deprecation import deprecated_warning

    _logger = logging.getLogger(__name__)

    IS_WINDOWS = sys.platform == "win32"

    H5PY_HEX_VERSION = calc_hexversion(*h5py.version.version_tuple[:3])
    HDF5_HEX_VERSION = calc_hexversion(*h5py.version.hdf5_version_tuple[:3])

    if h5py.version.version_tuple >= (3, 10):
        HDF5_SWMR_VERSION = 1, 9, 178
    else:
        HDF5_SWMR_VERSION = h5py.get_config().swmr_min_hdf5_version[:3]
    HAS_SWMR = HDF5_HEX_VERSION >= calc_hexversion(*HDF5_SWMR_VERSION)

    HAS_TRACK_ORDER = H5PY_HEX_VERSION >= calc_hexversion(2, 9, 0)

    if h5py.version.hdf5_version_tuple[:2] == (1, 10):
        HDF5_HAS_LOCKING_ARGUMENT = HDF5_HEX_VERSION >= calc_hexversion(1, 10, 7)
    else:
        HDF5_HAS_LOCKING_ARGUMENT = HDF5_HEX_VERSION >= calc_hexversion(1, 12, 1)
    H5PY_HAS_LOCKING_ARGUMENT = H5PY_HEX_VERSION >= calc_hexversion(3, 5, 0)
    HAS_LOCKING_ARGUMENT = HDF5_HAS_LOCKING_ARGUMENT & H5PY_HAS_LOCKING_ARGUMENT

    LATEST_LIBVER_IS_V108 = HDF5_HEX_VERSION < calc_hexversion(1, 10, 0)

    LOCKING_ARG_CHANGED_H5PY_VS_HDF5_WARMED = False

    def only_once(func):
        """
        simple decorator to make sure the warning is call at most one time.
        """

        def wrapper(*args, **kwargs):
            global LOCKING_ARG_CHANGED_H5PY_VS_HDF5_WARMED
            if not LOCKING_ARG_CHANGED_H5PY_VS_HDF5_WARMED:
                LOCKING_ARG_CHANGED_H5PY_VS_HDF5_WARMED = True
                return func(*args, **kwargs)

        return wrapper

    @only_once
    def warm_incoherent_locking_argument():
        """warm the user that file locking behavior has changed in recent versions of libhdf5"""
        _logger.critical(
            "The version of libhdf5 ({}) used by h5py ({}) is not supported: "
            "Do not expect file locking to work.".format(
                h5py.version.hdf5_version, h5py.version.version
            )
        )

    def _libver_low_bound_is_v108(libver) -> bool:
        if libver is None:
            return True
        if LATEST_LIBVER_IS_V108:
            return True
        if isinstance(libver, str):
            low = libver
        else:
            low = libver[0]
        if low == "latest":
            return False
        return low == "v108"

    def _hdf5_file_locking(
            mode: str | None = "r",
            locking: bool | str | None = None,
            swmr: bool | None = None,
            libver: str | Sequence[str] | None = None,
            **_
        ) -> str | bool | None:
        """Concurrent access by disabling file locking is not supported
        in these cases:

            * mode != "r": causes file corruption
            * SWMR: does not work
            * libver > v108 and file already locked: does not work
            * windows and HDF5_HAS_LOCKING_ARGUMENT and file already locked: does not work

        :param mode: read-only by default
        :param locking: by default it is disabled for `mode='r'`
                        and `swmr=False` and enabled when supported
                        for all other modes.
        :param swmr: try both modes when `mode='r'` and `swmr=None`
        :param libver:
        """
        if locking is None and mode == "r" and not swmr:
            locking = False
        if locking in (False, "false"):
            if mode != "r":
                raise ValueError("Locking is mandatory for HDF5 writing")
            if swmr:
                raise ValueError("Locking is mandatory for HDF5 SWMR mode")
            if IS_WINDOWS and HDF5_HAS_LOCKING_ARGUMENT:
                _logger.debug(
                    "Non-locking readers will fail when a writer has already locked the HDF5 file (this restriction applies to libhdf5 >= 1.12.1 or libhdf5 >= 1.10.7 on Windows)"
                )
            if not _libver_low_bound_is_v108(libver):
                _logger.debug(
                    "Non-locking readers will fail when a writer has already locked the HDF5 file (this restriction applies to libver >= v110)"
                )
        return locking

    def is_h5py_exception(e):
        """
        :param BaseException e:
        :returns bool:
        """
        if not isinstance(e, Exception):
            return False
        for frame in traceback.walk_tb(e.__traceback__):
            for namespace in (frame[0].f_locals, frame[0].f_globals):
                if namespace.get("__package__", None) == "h5py":
                    return True
        return False

    def retry_h5py_error(e):
        """
        :param BaseException e:
        :returns bool:
        """
        if is_h5py_exception(e):
            if isinstance(e, (OSError, RuntimeError)):
                return True
            elif isinstance(e, KeyError):
                # For example this needs to be retried:
                # KeyError: 'Unable to open object (bad object header version number)'
                return "Unable to open object" in str(e)
        elif isinstance(e, retry_mod.RetryError):
            return True
        return False

    def retry(**kw):
        r"""Decorator for a method that needs to be executed until it not longer
        fails on HDF5 IO. Mainly used for reading an HDF5 file that is being
        written.

        :param \**kw: see `silx.utils.retry`
        """
        kw.setdefault("retry_on_error", retry_h5py_error)
        return retry_mod.retry(**kw)

    def retry_contextmanager(**kw):
        r"""Decorator to make a context manager from a method that needs to be
        entered until it not longer fails on HDF5 IO. Mainly used for reading
        an HDF5 file that is being written.

        :param \**kw: see `silx.utils.retry_contextmanager`
        """
        kw.setdefault("retry_on_error", retry_h5py_error)
        return retry_mod.retry_contextmanager(**kw)

    def retry_in_subprocess(**kw):
        r"""Same as `retry` but it also retries segmentation faults.

        On Window you cannot use this decorator with the "@" syntax:

        .. code-block:: python

            def _method(*args, **kw):
                ...

            method = retry_in_subprocess()(_method)

        :param \**kw: see `silx.utils.retry_in_subprocess`
        """
        kw.setdefault("retry_on_error", retry_h5py_error)
        return retry_mod.retry_in_subprocess(**kw)

    def group_has_end_time(h5item):
        """Returns True when the HDF5 item is a Group with an "end_time"
        dataset. A reader can use this as an indication that the Group
        has been fully written (at least if the writer supports this).

        :param Union[h5py.Group,h5py.Dataset] h5item:
        :returns bool:
        """
        if isinstance(h5item, h5py.Group):
            return "end_time" in h5item
        else:
            return False

    @retry_contextmanager()
    def open_item(filename, name, retry_invalid=False, validate=None, **open_options):
        r"""Yield an HDF5 dataset or group (retry until it can be instantiated).

        :param str filename:
        :param str name: HDF5 path of the item
        :param bool retry_invalid: retry when item is missing or not valid
        :param callable or None validate:
        :param \**open_options: see `File.__init__`
        :yields Dataset, Group or None:
        """
        with File(filename, **open_options) as h5file:
            try:
                item = h5file[name]
            except KeyError as e:
                if "doesn't exist" in str(e):
                    if retry_invalid:
                        raise retry_mod.RetryError
                    else:
                        item = None
                else:
                    raise
            if callable(validate) and item is not None:
                if not validate(item):
                    if retry_invalid:
                        raise retry_mod.RetryError
                    else:
                        item = None
            yield item

    def _top_level_names(filename, include_only=group_has_end_time, **open_options):
        r"""Return all valid top-level HDF5 names.

        :param str filename:
        :param callable or None include_only:
        :param \**open_options: see `File.__init__`
        :returns list(str):
        """
        with File(filename, **open_options) as h5file:
            try:
                if callable(include_only):
                    return [name for name in h5file["/"] if include_only(h5file[name])]
                else:
                    return list(h5file["/"])
            except KeyError:
                raise retry_mod.RetryError

    top_level_names = retry()(_top_level_names)
    if hasattr(sys, "frozen") and sys.frozen:
        # multiprocessing not working on frozen binaries
        safe_top_level_names = top_level_names
    else:
        safe_top_level_names = retry_in_subprocess()(_top_level_names)

    class Hdf5FileLockingManager:
        """Manage HDF5 file locking in the current process through the HDF5_USE_FILE_LOCKING
        environment variable.
        """

        def __init__(self) -> None:
            self._hdf5_file_locking = None
            self._nfiles_open = 0

        def opened(self):
            self._add_nopen(1)

        def closed(self):
            self._add_nopen(-1)
            if not self._nfiles_open:
                self._restore_locking_env()

        def set_locking(self, locking):
            if self._nfiles_open:
                self._check_locking_env(locking)
            else:
                self._set_locking_env(locking)

        def _add_nopen(self, v):
            self._nfiles_open = max(self._nfiles_open + v, 0)

        def _set_locking_env(self, enable):
            self._backup_locking_env()
            if enable:
                os.environ["HDF5_USE_FILE_LOCKING"] = "TRUE"
            elif enable is None:
                try:
                    del os.environ["HDF5_USE_FILE_LOCKING"]
                except KeyError:
                    pass
            else:
                os.environ["HDF5_USE_FILE_LOCKING"] = "FALSE"

        def _get_locking_env(self):
            v = os.environ.get("HDF5_USE_FILE_LOCKING")
            if v == "TRUE":
                return True
            elif v is None:
                return None
            else:
                return False

        def _check_locking_env(self, enable):
            if enable != self._get_locking_env():
                if enable:
                    raise RuntimeError(
                        "Close all HDF5 files before enabling HDF5 file locking"
                    )
                else:
                    raise RuntimeError(
                        "Close all HDF5 files before disabling HDF5 file locking"
                    )

        def _backup_locking_env(self):
            v = os.environ.get("HDF5_USE_FILE_LOCKING")
            if v is None:
                self._hdf5_file_locking = None
            else:
                self._hdf5_file_locking = v == "TRUE"

        def _restore_locking_env(self):
            self._set_locking_env(self._hdf5_file_locking)
            self._hdf5_file_locking = None

    class File(h5py.File):
        """Takes care of HDF5 file locking and SWMR mode without the need
        to handle those explicitely.

        When file locking is managed through the HDF5_USE_FILE_LOCKING environment
        variable, you cannot open different files simultaneously with different modes.
        """

        _SWMR_LIBVER = "latest"

        if HAS_LOCKING_ARGUMENT:
            _LOCKING_MGR = None
        else:
            _LOCKING_MGR = Hdf5FileLockingManager()

        def __init__(
            self,
            filename: str,
            mode: str | None = None,
            locking: bool | str | None = None,
            enable_file_locking: bool | None = None,
            swmr: bool | None = None,
            libver: str | Sequence[str] | None = None,
            **kwargs,
        ):
            r"""The arguments `locking` and `swmr` should not be
            specified explicitly for normal use cases.

            :param filename:
            :param mode: read-only by default
            :param locking: by default it is disabled for `mode='r'`
                            and `swmr=False` and enabled when supported
                            for all other modes.
            :param enable_file_locking: deprecated
            :param swmr: try both modes when `mode='r'` and `swmr=None`
            :param libver:
            :param \**kwargs: see `h5py.File.__init__`
            """
            # File locking behavior has changed in recent versions of libhdf5
            if HDF5_HAS_LOCKING_ARGUMENT != H5PY_HAS_LOCKING_ARGUMENT:
                warm_incoherent_locking_argument()

            if mode is None:
                mode = "r"
            elif mode not in ("r", "w", "w-", "x", "a", "r+"):
                raise ValueError("invalid mode {}".format(mode))
            if not HAS_SWMR:
                swmr = False
            if swmr and libver is None:
                libver = self._SWMR_LIBVER

            if enable_file_locking is not None:
                deprecated_warning(
                    type_="argument",
                    name="enable_file_locking",
                    replacement="locking",
                    since_version="1.0",
                )
                if locking is None:
                    locking = enable_file_locking
            locking = _hdf5_file_locking(
                mode=mode, locking=locking, swmr=swmr, libver=libver
            )
            if self._LOCKING_MGR is None:
                kwargs.setdefault("locking", locking)
            else:
                self._LOCKING_MGR.set_locking(locking)

            if HAS_TRACK_ORDER:
                kwargs.setdefault("track_order", True)
            try:
                super().__init__(filename, mode=mode, swmr=swmr, libver=libver, **kwargs)
            except OSError as e:
                #   wlock   wSWMR   rlock   rSWMR   OSError: Unable to open file (...)
                # 1 TRUE    FALSE   FALSE   FALSE   -
                # 2 TRUE    FALSE   FALSE   TRUE    -
                # 3 TRUE    FALSE   TRUE    FALSE   unable to lock file, errno = 11, error message = 'Resource temporarily unavailable'
                # 4 TRUE    FALSE   TRUE    TRUE    unable to lock file, errno = 11, error message = 'Resource temporarily unavailable'
                # 5 TRUE    TRUE    FALSE   FALSE   file is already open for write (may use <h5clear file> to clear file consistency flags)
                # 6 TRUE    TRUE    FALSE   TRUE    -
                # 7 TRUE    TRUE    TRUE    FALSE   file is already open for write (may use <h5clear file> to clear file consistency flags)
                # 8 TRUE    TRUE    TRUE    TRUE    -
                if (
                    mode == "r"
                    and swmr is None
                    and "file is already open for write" in str(e)
                ):
                    # Try reading in SWMR mode (situation 5 and 7)
                    swmr = True
                    if libver is None:
                        libver = self._SWMR_LIBVER
                    super().__init__(
                        filename, mode=mode, swmr=swmr, libver=libver, **kwargs
                    )
                else:
                    raise
            else:
                self._file_open_callback()
                try:
                    if mode != "r" and swmr:
                        # Try setting writer in SWMR mode
                        self.swmr_mode = True
                except Exception:
                    self.close()
                    raise

        def close(self):
            super().close()
            self._file_close_callback()

        def _file_open_callback(self):
            if self._LOCKING_MGR is not None:
                self._LOCKING_MGR.opened()

        def _file_close_callback(self):
            if self._LOCKING_MGR is not None:
                self._LOCKING_MGR.closed()