codekingpro/portable-devtools

1from __future__ import annotations2 3import os4import sys5import warnings6from contextlib import suppress7from errno import EAGAIN, ENOSYS, EWOULDBLOCK8from pathlib import Path9from typing import cast10 11from ._api import BaseFileLock12from ._util import ensure_directory_exists13 14#: a flag to indicate if the fcntl API is available15has_fcntl = False16if sys.platform == "win32":  # pragma: win32 cover17 18    class UnixFileLock(BaseFileLock):19        """Uses the :func:`fcntl.flock` to hard lock the lock file on unix systems."""20 21        def _acquire(self) -> None:22            raise NotImplementedError23 24        def _release(self) -> None:25            raise NotImplementedError26 27else:  # pragma: win32 no cover28    try:29        import fcntl30 31        _ = (fcntl.flock, fcntl.LOCK_EX, fcntl.LOCK_NB, fcntl.LOCK_UN)32    except (ImportError, AttributeError):33        pass34    else:35        has_fcntl = True36 37    class UnixFileLock(BaseFileLock):38        """39        Uses the :func:`fcntl.flock` to hard lock the lock file on unix systems.40 41        Lock file cleanup: Unix and macOS delete the lock file reliably after release, even in42        multi-threaded scenarios. Unlike Windows, Unix allows unlinking files that other processes43        have open.44        """45 46        def _acquire(self) -> None:  # noqa: C901, PLR091247            ensure_directory_exists(self.lock_file)48            open_flags = os.O_RDWR | os.O_TRUNC49            o_nofollow = getattr(os, "O_NOFOLLOW", None)50            if o_nofollow is not None:51                open_flags |= o_nofollow52            open_flags |= os.O_CREAT53            open_mode = self._open_mode()54            try:55                fd = os.open(self.lock_file, open_flags, open_mode)56            except FileNotFoundError:57                # On FUSE/NFS, os.open(O_CREAT) is not atomic: LOOKUP + CREATE can be split, allowing a concurrent58                # unlink() to delete the file between them. For valid paths, treat ENOENT as transient contention.59                # For invalid paths (e.g., empty string), re-raise to avoid infinite retry loops.60                if self.lock_file and Path(self.lock_file).parent.exists():61                    return62                raise63            except PermissionError:64                # Sticky-bit dirs (e.g. /tmp): O_CREAT fails if the file is owned by another user (#317).65                # Fall back to opening the existing file without O_CREAT.66                if not Path(self.lock_file).exists():67                    raise68                try:69                    fd = os.open(self.lock_file, open_flags & ~os.O_CREAT, open_mode)70                except FileNotFoundError:71                    return72            if self.has_explicit_mode:73                with suppress(PermissionError):74                    os.fchmod(fd, self._context.mode)75            try:76                fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)77            except OSError as exception:78                os.close(fd)79                if exception.errno == ENOSYS:80                    with suppress(OSError):81                        Path(self.lock_file).unlink()82                    self._fallback_to_soft_lock()83                    self._acquire()84                    return85                if exception.errno not in {EAGAIN, EWOULDBLOCK}:86                    raise87            else:88                # The file may have been unlinked by a concurrent _release() between our open() and flock().89                # A lock on an unlinked inode is useless — discard and let the retry loop start fresh.90                if os.fstat(fd).st_nlink == 0:91                    os.close(fd)92                else:93                    self._context.lock_file_fd = fd94 95        def _fallback_to_soft_lock(self) -> None:96            from ._soft import SoftFileLock  # noqa: PLC041597 98            warnings.warn("flock not supported on this filesystem, falling back to SoftFileLock", stacklevel=2)99            from .asyncio import AsyncSoftFileLock, BaseAsyncFileLock  # noqa: PLC0415100 101            self.__class__ = AsyncSoftFileLock if isinstance(self, BaseAsyncFileLock) else SoftFileLock102 103        def _release(self) -> None:104            fd = cast("int", self._context.lock_file_fd)105            self._context.lock_file_fd = None106            with suppress(OSError):107                Path(self.lock_file).unlink()108            fcntl.flock(fd, fcntl.LOCK_UN)109            with suppress(OSError):  # close can raise EIO on FUSE/Docker bind-mount filesystems after unlink110                os.close(fd)111 112 113__all__ = [114    "UnixFileLock",115    "has_fcntl",116]117