"""Cross-process and cross-host reader/writer lock built on :class:`SoftFileLock` primitives.""" from __future__ import annotations import atexit import hmac import os import re import secrets import socket import stat import sys import threading import time import uuid from contextlib import contextmanager, suppress from dataclasses import dataclass from pathlib import Path from typing import TYPE_CHECKING, Final, Literal from weakref import WeakValueDictionary from filelock._api import ( AcquireReturnProxy, _ensure_current_process, _fork_transition, _raise_grouped_errors, _register_fork_class, _register_fork_object, _register_owned_descriptor, _unregister_owned_descriptor, ) from filelock._error import Timeout from filelock._soft import SoftFileLock from filelock._util import ensure_directory_exists, touch, write_all if TYPE_CHECKING: from collections.abc import Callable, Generator _Mode = Literal["read", "write"] _BREAK_SUFFIX: Final[str] = ".break" _MAX_MARKER_SIZE: Final[int] = 1024 _O_NOFOLLOW: Final[int] = getattr(os, "O_NOFOLLOW", 0) _O_NONBLOCK: Final[int] = getattr(os, "O_NONBLOCK", 0) # dirfd-relative I/O is a Unix-only optimization; Windows cannot ``os.open()`` a directory at all, and # its ``os`` module skips dir_fd support entirely. When disabled, callers fall back to full-path ops. _SUPPORTS_DIR_FD: Final[bool] = sys.platform != "win32" and os.open in os.supports_dir_fd _ALL_INSTANCES: Final[WeakValueDictionary[int, SoftReadWriteLock]] = WeakValueDictionary() _ALL_INSTANCES_LOCK: threading.Lock = threading.Lock() _SINGLETONS_UNDER_CONSTRUCTION: Final[set[Path]] = set() class _SoftRWMeta(type): _instances: WeakValueDictionary[Path, SoftReadWriteLock] _instances_lock: threading.RLock def __call__( # ruff:ignore[too-many-arguments] # forwards the public constructor's documented parameters cls, lock_file: str | os.PathLike[str], timeout: float = -1, *, blocking: bool = True, is_singleton: bool = True, heartbeat_interval: float = 30.0, stale_threshold: float | None = None, poll_interval: float = 0.25, ) -> SoftReadWriteLock: _ensure_current_process() if not is_singleton: return super().__call__( lock_file, timeout, blocking=blocking, is_singleton=is_singleton, heartbeat_interval=heartbeat_interval, stale_threshold=stale_threshold, poll_interval=poll_interval, ) normalized = Path(lock_file).resolve() with cls._instances_lock: instance = cls._instances.get(normalized) if instance is None: if normalized in _SINGLETONS_UNDER_CONSTRUCTION: # pragma: needs fork msg = f"Singleton lock construction is already active for {lock_file!s}" raise RuntimeError(msg) construction_pid = os.getpid() _SINGLETONS_UNDER_CONSTRUCTION.add(normalized) try: instance = super().__call__( lock_file, timeout, blocking=blocking, is_singleton=is_singleton, heartbeat_interval=heartbeat_interval, stale_threshold=stale_threshold, poll_interval=poll_interval, ) finally: _SINGLETONS_UNDER_CONSTRUCTION.discard(normalized) if os.getpid() != construction_pid: # pragma: needs fork msg = "Lock construction cannot continue after fork; construct a new lock in the child" raise RuntimeError(msg) cls._instances[normalized] = instance elif instance.timeout != timeout or instance.blocking != blocking: msg = ( f"Singleton lock created with timeout={instance.timeout}, blocking={instance.blocking}," f" cannot be changed to timeout={timeout}, blocking={blocking}" ) raise ValueError(msg) return instance class SoftReadWriteLock(metaclass=_SoftRWMeta): """ Cross-process and cross-host reader/writer lock built on :class:`SoftFileLock` primitives. Use this class instead of :class:`~filelock.ReadWriteLock` when the lock file lives on a network filesystem (NFS, Lustre with ``-o flock``, HPC cluster shared storage). ``ReadWriteLock`` is backed by SQLite and cannot run on NFS because SQLite's ``fcntl`` locking is unreliable there. Layout on disk for a lock at ``foo.lock``: - ``foo.lock.state`` — a :class:`SoftFileLock` taken only during state transitions (microseconds). - ``foo.lock.write`` — writer marker; its presence means a writer is claiming or holding the lock. - ``foo.lock.readers/..`` — one file per reader. Each marker stores a random token (``secrets.token_hex(16)``), the holder's pid, and the holder's hostname. A daemon heartbeat thread refreshes ``mtime`` on every held marker. A marker whose mtime has not advanced in ``stale_threshold`` seconds may be evicted by any process on any host, giving correct behavior when a compute node crashes with a lock held. Writer acquire is two-phase and writer-preferring: phase 1 claims ``.write`` (blocking any new reader), phase 2 waits for existing readers to drain. Writer starvation is impossible. Reentrancy, upgrade/downgrade rules, thread pinning, and singleton caching by resolved path match :class:`~filelock.ReadWriteLock`. Forking invalidates the inherited instance in the child so the child cannot double-own the lock with its parent; ``release()`` on that instance is a no-op, and the child must construct a new instance if it needs a lock. Trust boundary: protects against same-UID non-cooperating processes (one host or cross-host) and same-host different-UID users via ``0o600`` / ``0o700`` permissions. Does not protect against root compromise, NTP tampering on same-UID cross-host nodes, or multi-tenant mounts where hostile co-tenants share the UID. :param lock_file: path to the lock file; sidecar state/write/readers live next to it :param timeout: maximum wait time in seconds; ``-1`` means block indefinitely :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately on contention :param is_singleton: if ``True``, reuse existing instances for the same resolved path :param heartbeat_interval: seconds between heartbeat refreshes; default 30 s :param stale_threshold: seconds of ``mtime`` inactivity before a marker is stale; defaults to ``3 * heartbeat_interval``, matching etcd's ``LeaseKeepAlive`` convention :param poll_interval: seconds between acquire retries under contention; default 0.25 s .. versionadded:: 3.27.0 """ _instances: WeakValueDictionary[Path, SoftReadWriteLock] = WeakValueDictionary() _instances_lock = threading.RLock() def __init__( # ruff:ignore[too-many-arguments] # public constructor: one parameter per documented lock option self, lock_file: str | os.PathLike[str], timeout: float = -1, *, blocking: bool = True, is_singleton: bool = True, # ruff:ignore[unused-method-argument] # consumed by _SoftRWMeta.__call__ heartbeat_interval: float = 30.0, stale_threshold: float | None = None, poll_interval: float = 0.25, ) -> None: self._creator_pid = os.getpid() if heartbeat_interval <= 0: msg = f"heartbeat_interval must be positive, got {heartbeat_interval}" raise ValueError(msg) if stale_threshold is None: stale_threshold = heartbeat_interval * 3 if stale_threshold <= heartbeat_interval: msg = f"stale_threshold must exceed heartbeat_interval ({stale_threshold} <= {heartbeat_interval})" raise ValueError(msg) if poll_interval <= 0: msg = f"poll_interval must be positive, got {poll_interval}" raise ValueError(msg) self.lock_file: str = os.fspath(lock_file) self.timeout: float = timeout self.blocking: bool = blocking self.heartbeat_interval: float = heartbeat_interval self.stale_threshold: float = stale_threshold self.poll_interval: float = poll_interval self._paths = _Paths( state=f"{self.lock_file}.state", write=f"{self.lock_file}.write", readers=f"{self.lock_file}.readers", ) ensure_directory_exists(self.lock_file) self._locks = _Locks( internal=threading.Lock(), transaction=threading.Lock(), state=SoftFileLock(self._paths.state, timeout=-1), ) self._readers_dir_fd: int | None = None self._readers_dir_fd_token: int | None = None self._hold: _Hold | None = None self._closed: bool = False with _ALL_INSTANCES_LOCK: _ALL_INSTANCES[id(self)] = self _register_fork_object(self) @classmethod def _reset_class_after_fork(cls) -> None: # pragma: forked child global _ALL_INSTANCES_LOCK # ruff:ignore[global-statement] # rebinds the module lock to a fresh one in the fork child _ALL_INSTANCES_LOCK = threading.Lock() cls._instances = WeakValueDictionary() cls._instances_lock = threading.RLock() _SINGLETONS_UNDER_CONSTRUCTION.clear() @contextmanager def read_lock(self, timeout: float | None = None, *, blocking: bool | None = None) -> Generator[None]: """ Context manager that acquires and releases a shared read lock. Falls back to instance defaults for *timeout* and *blocking* when ``None``. :param timeout: maximum wait time in seconds, or ``None`` to use the instance default :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately; ``None`` uses the instance default :raises RuntimeError: if a write lock is already held on this instance :raises Timeout: if the lock cannot be acquired within *timeout* seconds """ self.acquire_read(timeout, blocking=blocking) try: yield finally: self.release() @contextmanager def write_lock(self, timeout: float | None = None, *, blocking: bool | None = None) -> Generator[None]: """ Context manager that acquires and releases an exclusive write lock. Falls back to instance defaults for *timeout* and *blocking* when ``None``. :param timeout: maximum wait time in seconds, or ``None`` to use the instance default :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately; ``None`` uses the instance default :raises RuntimeError: if a read lock is already held, or a write lock is held by a different thread :raises Timeout: if the lock cannot be acquired within *timeout* seconds """ self.acquire_write(timeout, blocking=blocking) try: yield finally: self.release() def acquire_read(self, timeout: float | None = None, *, blocking: bool | None = None) -> AcquireReturnProxy: """ Acquire a shared read lock. If this instance already holds a read lock, the lock level is incremented (reentrant). Attempting to acquire a read lock while holding a write lock raises :class:`RuntimeError` (downgrade not allowed). On the 0→1 transition a daemon heartbeat thread is started that refreshes the reader marker's ``mtime`` every ``heartbeat_interval`` seconds so peers on other hosts do not evict the marker as stale. :param timeout: maximum wait time in seconds, or ``None`` to use the instance default; ``-1`` means block indefinitely :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately when the lock is unavailable; ``None`` uses the instance default :returns: a proxy that can be used as a context manager to release the lock :raises RuntimeError: if a write lock is already held on this instance, if this instance was invalidated by :func:`os.fork`, or if :meth:`close` was called :raises Timeout: if the lock cannot be acquired within *timeout* seconds """ return self._acquire("read", timeout, blocking=blocking) def acquire_write(self, timeout: float | None = None, *, blocking: bool | None = None) -> AcquireReturnProxy: """ Acquire an exclusive write lock. If this instance already holds a write lock from the same thread, the lock level is incremented (reentrant). Attempting to acquire a write lock while holding a read lock raises :class:`RuntimeError` (upgrade not allowed). Write locks are pinned to the acquiring thread: a different thread trying to re-enter also raises :class:`RuntimeError`. Writer acquisition runs in two phases. Phase 1 atomically claims ``.write`` via ``O_CREAT | O_EXCL``, which immediately blocks any new reader on any host. Phase 2 waits for existing readers to drain. Writer starvation is impossible: new readers see ``.write`` during phase 2 and wait behind the pending writer. :param timeout: maximum wait time in seconds, or ``None`` to use the instance default; ``-1`` means block indefinitely :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately when the lock is unavailable; ``None`` uses the instance default :returns: a proxy that can be used as a context manager to release the lock :raises RuntimeError: if a read lock is already held, if a write lock is held by a different thread, if this instance was invalidated by :func:`os.fork`, or if :meth:`close` was called :raises Timeout: if the lock cannot be acquired within *timeout* seconds """ return self._acquire("write", timeout, blocking=blocking) @classmethod def get_lock( cls, lock_file: str | os.PathLike[str], timeout: float = -1, *, blocking: bool = True, ) -> SoftReadWriteLock: """ Return the singleton :class:`SoftReadWriteLock` for *lock_file*. :param lock_file: path to the lock file; sidecar state/write/readers live next to it :param timeout: maximum wait time in seconds; ``-1`` means block indefinitely :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately when the lock is unavailable :returns: the singleton lock instance :raises ValueError: if an instance already exists for this path with different *timeout* or *blocking* values """ return cls(lock_file, timeout, blocking=blocking) def close(self) -> None: """ Release any held lock and release internal filesystem resources. Idempotent. After calling this method the instance can no longer acquire locks — subsequent acquires raise :class:`RuntimeError`. A fork-invalidated instance is closed without raising. """ if self._creator_pid != os.getpid(): # pragma: forked child return self.release(force=True) with self._locks.internal: if self._closed: return self._closed = True if self._readers_dir_fd is not None: # pragma: needs dir-fd with _fork_transition(): if self._readers_dir_fd_token is not None: # pragma: needs dir-fd _unregister_owned_descriptor(self._readers_dir_fd_token) self._readers_dir_fd_token = None fd, self._readers_dir_fd = self._readers_dir_fd, None with suppress(OSError): # pragma: needs dir-fd os.close(fd) def release(self, *, force: bool = False) -> None: """ Release one level of the current lock. When the lock level reaches zero the heartbeat thread is stopped and the held marker file is unlinked. On a fork-invalidated instance (that is, the child of a :func:`os.fork` call made while the parent held a lock) this method is a no-op so inherited ``with`` blocks can unwind cleanly in the child. :param force: if ``True``, release the lock completely regardless of the current lock level :raises RuntimeError: if no lock is currently held and *force* is ``False`` """ if self._creator_pid != os.getpid(): # pragma: forked child return with self._locks.internal: hold = self._hold if hold is None: if force: return msg = f"Cannot release a lock on {self.lock_file} (lock id: {id(self)}) that is not held" raise RuntimeError(msg) if force: hold.level = 0 else: hold.level -= 1 if hold.level > 0: return self._hold = None # Order matters: signal → join → unlink. A late tick on a deleted marker is harmless and the # heartbeat's token check would catch a re-acquisition race, but joining first removes that race. hold.heartbeat_stop.set() hold.heartbeat_thread.join(timeout=self.heartbeat_interval + 1.0) if hold.is_reader: _unlink(hold.marker_name, dir_fd=self._readers_dir_fd) else: self._unlink_writer_marker_if_ours(hold.token) def _unlink_writer_marker_if_ours(self, token: str) -> None: # Remove the writer marker only while it still carries our token. If this holder was paused long # enough (a stop-the-world GC pause, SIGSTOP, a suspended VM) for a peer to evict the marker as # stale and claim the writer slot itself, the file now at .write is the peer's live marker; # unlinking it by path would let a second writer through and break mutual exclusion. The state lock # serializes this against a concurrent break/claim, and the heartbeat is already stopped, so the # token we read is authoritative. Mirrors the token re-check the stale-break path already does. with self._locks.state: if (read := _read_marker(self._paths.write)) is None: return info, _ = read if info is None or not hmac.compare_digest(info.token, token): return _unlink(self._paths.write) def _acquire( self, mode: _Mode, timeout: float | None, *, blocking: bool | None, ) -> AcquireReturnProxy: if self._creator_pid != os.getpid(): # pragma: forked child msg = f"SoftReadWriteLock on {self.lock_file} was invalidated by fork(); construct a new instance" raise RuntimeError(msg) timeout = self.timeout if timeout is None else timeout blocking = self.blocking if blocking is None else blocking with self._locks.internal: if self._closed: msg = f"SoftReadWriteLock on {self.lock_file} has been closed" raise RuntimeError(msg) if self._hold is not None: return self._validate_reentrant(mode) start = time.perf_counter() if not blocking: acquired = self._locks.transaction.acquire(blocking=False) elif timeout == -1: acquired = self._locks.transaction.acquire(blocking=True) else: acquired = self._locks.transaction.acquire(blocking=True, timeout=timeout) if not acquired: raise Timeout(self.lock_file) from None try: return self._do_acquire_inner(mode, timeout, start, blocking=blocking) finally: self._locks.transaction.release() def _do_acquire_inner( self, mode: _Mode, effective_timeout: float, start: float, *, blocking: bool, ) -> AcquireReturnProxy: with self._locks.internal: if self._hold is not None: return self._validate_reentrant(mode) deadline = None if effective_timeout == -1 else start + effective_timeout token = secrets.token_hex(16) if mode == "write": marker_name, is_reader = self._acquire_writer_slot(token, deadline=deadline, blocking=blocking) else: marker_name, is_reader = self._acquire_reader_slot(token, deadline=deadline, blocking=blocking) stop_event = threading.Event() heartbeat = _HeartbeatThread( refresh=self._refresh_marker, interval=self.heartbeat_interval, stop_event=stop_event, name=f"filelock-heartbeat-{id(self):x}", ) with self._locks.internal: self._hold = _Hold( level=1, mode=mode, write_thread_id=threading.get_ident() if mode == "write" else None, marker_name=marker_name, is_reader=is_reader, token=token, heartbeat_thread=heartbeat, heartbeat_stop=stop_event, ) heartbeat.start() return AcquireReturnProxy(lock=self) def _validate_reentrant(self, mode: _Mode) -> AcquireReturnProxy: hold = self._hold assert hold is not None # ruff:ignore[assert] # callers dispatch here only inside the self._hold is not None branch if hold.mode != mode: opposite = "write" if mode == "read" else "read" direction = "downgrade" if mode == "read" else "upgrade" msg = ( f"Cannot acquire {mode} lock on {self.lock_file} (lock id: {id(self)}): " f"already holding a {opposite} lock ({direction} not allowed)" ) raise RuntimeError(msg) if mode == "write" and (cur := threading.get_ident()) != hold.write_thread_id: msg = ( f"Cannot acquire write lock on {self.lock_file} (lock id: {id(self)}) " f"from thread {cur} while it is held by thread {hold.write_thread_id}" ) raise RuntimeError(msg) hold.level += 1 return AcquireReturnProxy(lock=self) def _acquire_writer_slot( self, token: str, *, deadline: float | None, blocking: bool, ) -> tuple[str, bool]: # Phase 2 scans readers/ via dirfd (where supported), so we need it open even though writers never # create files inside. self._open_readers_dir() def try_claim_writer() -> bool: with self._locks.state: return self._claim_writer_marker(token) def readers_drained_touching() -> bool: with self._locks.state: # A peer may replace an expired marker while this process pauses. Refresh only our token; touching a # successor's marker would let this acquisition proceed without owning the writer slot. if not self._touch_writer_marker_if_ours(token) and not self._claim_writer_marker(token): return False self._break_stale_readers(time.time()) return not self._any_readers() self._wait_for(try_claim_writer, deadline=deadline, blocking=blocking) try: self._wait_for(readers_drained_touching, deadline=deadline, blocking=blocking) except Timeout: # Give up our writer claim so readers can make progress again, but only while the marker is # still ours: a peer may have evicted it as stale and claimed the slot while phase 2 waited. self._unlink_writer_marker_if_ours(token) raise return self._paths.write, False def _claim_writer_marker(self, token: str) -> bool: # Claim the writer slot for ``token``. Must be called holding ``self._locks.state``. Evicts a # stale marker first, then refuses to claim while a live ``.write`` exists so a peer holding the # slot is waited out instead of overwritten. _break_stale_marker(self._paths.write, stale_threshold=self.stale_threshold, now=time.time()) if _file_exists(self._paths.write): return False try: _atomic_create_marker(self._paths.write, token) except FileExistsError: return False return True def _touch_writer_marker_if_ours(self, token: str) -> bool: # Refresh the writer marker through a single O_NOFOLLOW fd, but only while it still carries our # token. Returns False when the marker is gone or now belongs to a peer that reclaimed the slot, # so the caller can re-claim rather than keep a stranger's marker alive. Mirrors _refresh_marker. fd = _open_marker(self._paths.write) if fd is None: return False try: try: data = os.read(fd, _MAX_MARKER_SIZE + 1) except OSError: # pragma: no cover - e.g. EAGAIN from a hostile FIFO that has a writer attached return False info = _parse_marker_bytes(data) if info is None or not hmac.compare_digest(info.token, token): return False with suppress(OSError): touch(self._paths.write, fd=fd) return True finally: os.close(fd) def _acquire_reader_slot( self, token: str, *, deadline: float | None, blocking: bool, ) -> tuple[str, bool]: self._open_readers_dir() reader_name = f"{uuid.uuid4().hex}.{os.getpid()}" dir_fd = self._readers_dir_fd full_reader_path = str(Path(self._paths.readers) / reader_name) def try_claim_reader() -> bool: with self._locks.state: _break_stale_marker(self._paths.write, stale_threshold=self.stale_threshold, now=time.time()) if _file_exists(self._paths.write): return False if dir_fd is not None: # pragma: needs dir-fd _atomic_create_marker(reader_name, token, dir_fd=dir_fd) else: # pragma: win32 cover _atomic_create_marker(full_reader_path, token) return True self._wait_for(try_claim_reader, deadline=deadline, blocking=blocking) return (reader_name if dir_fd is not None else full_reader_path), True def _wait_for( self, predicate: Callable[[], bool], *, deadline: float | None, blocking: bool, ) -> None: while True: if predicate(): return now = time.perf_counter() if not blocking: raise Timeout(self.lock_file) if deadline is not None and now >= deadline: raise Timeout(self.lock_file) sleep_for = self.poll_interval if deadline is not None: sleep_for = min(sleep_for, max(deadline - now, 0.0)) time.sleep(sleep_for) def _open_readers_dir(self) -> None: readers_path = Path(self._paths.readers) with suppress(FileExistsError): readers_path.mkdir(mode=0o700) # mkdir has no O_NOFOLLOW, so verify via lstat that we did not land on an attacker-placed symlink # or a regular file before we open or scan inside. st = os.lstat(self._paths.readers) if stat.S_ISLNK(st.st_mode) or not stat.S_ISDIR(st.st_mode): msg = f"{self._paths.readers} exists but is not a directory or is a symlink; refusing to use it" raise RuntimeError(msg) if self._readers_dir_fd is None and _SUPPORTS_DIR_FD: # pragma: needs dir-fd with _fork_transition(): fd = os.open(self._paths.readers, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | _O_NOFOLLOW) try: token = _register_owned_descriptor(fd) except BaseException as registration_error: try: os.close(fd) except BaseException as close_error: # ruff:ignore[blind-except] # both errors surface via the group below _raise_grouped_errors( "reader directory registration and descriptor close both failed", registration_error, close_error, ) raise self._readers_dir_fd = fd self._readers_dir_fd_token = token def _any_readers(self) -> bool: for _ in self._iter_reader_entries(): return True return False def _iter_reader_entries(self) -> Generator[tuple[str, bool]]: """ Yield ``(name, dirfd_relative)`` pairs for every live reader marker. ``dirfd_relative`` is ``True`` when *name* should be passed to ``dir_fd=``-aware syscalls; ``False`` when *name* is a full path because dirfd-relative I/O is unavailable on this platform. """ if self._readers_dir_fd is not None: # pragma: needs dir-fd with os.scandir(self._readers_dir_fd) as it: for entry in it: if not _is_housekeeping_name(entry.name): yield entry.name, True return readers_path = Path(self._paths.readers) # pragma: win32 cover with os.scandir(readers_path) as it: # pragma: win32 cover for entry in it: # pragma: win32 cover if not _is_housekeeping_name(entry.name): # pragma: win32 cover yield str(readers_path / entry.name), False # pragma: win32 cover def _break_stale_readers(self, now: float) -> None: names: list[tuple[str, int | None]] = [] try: for name, dirfd_relative in self._iter_reader_entries(): names.append((name, self._readers_dir_fd if dirfd_relative else None)) except OSError: # pragma: no cover - transient NFS scandir hiccup return for name, fd in names: _break_stale_marker(name, stale_threshold=self.stale_threshold, now=now, dir_fd=fd) def _refresh_marker(self) -> bool: with self._locks.internal: hold = self._hold if hold is None: # pragma: no cover - race between stop_event.set and join return False marker_name = hold.marker_name token = hold.token dir_fd = self._readers_dir_fd if hold.is_reader else None # Open once with O_NOFOLLOW and touch that exact descriptor. Refreshing through the verified fd # (instead of re-opening by name) closes the window where a peer unlinks our marker and drops a symlink # or a different file at the path between the read and the touch: utime then lands on the inode we # verified, or nowhere. Only an unambiguous loss stops the heartbeat: the marker gone, or a peer's token # in its place. A transient filesystem error (ESTALE / EIO on the NFS-style filesystems this lock targets) # keeps the heartbeat alive to retry next tick, the way the touch below already does, so one blip does not # silently drop a held lock. try: fd = _open_marker_fd(marker_name, dir_fd=dir_fd) except FileNotFoundError: return False except OSError: return True try: try: data = _read_marker_fd(fd) except OSError: # a transient read error or EAGAIN from a hostile FIFO; retry rather than drop the lock return True info = _parse_marker_bytes(data) # Token mismatch means another process already evicted our marker and created its own; stop the # thread so it does not keep a stranger's file alive. if info is None or not hmac.compare_digest(info.token, token): return False # A transient touch failure (ESTALE / EIO on the NFS-style filesystems this lock targets) must not # kill the heartbeat thread: the read above just confirmed the marker is still ours, so swallow the # error and retry on the next tick rather than letting the lease lapse while we still hold the lock. with suppress(OSError): touch(marker_name, fd=fd) return True finally: os.close(fd) def _reset_after_fork_in_child(self) -> None: # pragma: forked child self._locks = _Locks( internal=threading.Lock(), transaction=threading.Lock(), state=self._locks.state, ) self._hold = None self._readers_dir_fd = None self._readers_dir_fd_token = None class _HeartbeatThread(threading.Thread): def __init__( self, refresh: Callable[[], bool], interval: float, stop_event: threading.Event, name: str, ) -> None: super().__init__(name=name, daemon=True) self._refresh = refresh self._interval = interval self._stop_event = stop_event def run(self) -> None: while not self._stop_event.wait(self._interval): if not self._refresh(): self._stop_event.set() return def _read_marker(name: str, *, dir_fd: int | None = None) -> tuple[_MarkerInfo | None, float] | None: fd = _open_marker(name, dir_fd=dir_fd) if fd is None: return None try: st = os.fstat(fd) # A legitimate marker is a regular file, so anything else at the path (a FIFO, say) is reported as a # malformed marker (its mtime still drives stale eviction) without being read. Reading is where # platforms diverge: an empty non-blocking read yields 0 bytes on Linux/macOS but EAGAIN on FreeBSD, # and the EAGAIN used to abort the stale-break and wedge the acquire until timeout (#587). if not stat.S_ISREG(st.st_mode): # pragma: needs fifo return None, st.st_mtime data = os.read(fd, _MAX_MARKER_SIZE + 1) except OSError: # pragma: no cover - marker vanished or turned unreadable between open and read return None finally: os.close(fd) return _parse_marker_bytes(data), st.st_mtime def _read_marker_fd(fd: int) -> bytes: return os.read(fd, _MAX_MARKER_SIZE + 1) def _open_marker_fd(name: str, *, dir_fd: int | None = None) -> int: # The file is ours; these guard a hostile mid-flight swap. O_NOFOLLOW rejects a symlink; O_NONBLOCK keeps # a real FIFO from blocking the open forever, so it reads as a malformed marker instead of wedging a peer # that holds the state lock. flags = os.O_RDONLY | _O_NOFOLLOW | _O_NONBLOCK return os.open(name, flags, dir_fd=dir_fd) if _SUPPORTS_DIR_FD and dir_fd is not None else os.open(name, flags) def _open_marker(name: str, *, dir_fd: int | None = None) -> int | None: try: return _open_marker_fd(name, dir_fd=dir_fd) except OSError: return None def _parse_marker_bytes(data: bytes) -> _MarkerInfo | None: # Trust nothing about attacker-controlled markers; any deviation returns None so callers fall through # to stale cleanup. ``re.match`` caches compiled patterns internally, so the regex is built only once # despite being defined inline. if not data or len(data) > _MAX_MARKER_SIZE: return None try: text = data.decode("ascii") except UnicodeDecodeError: return None match = re.match( r""" \A # start of string (?P [0-9a-f]{32} ) \n # 128-bit hex token (?P [1-9][0-9]{0,9} ) \n # decimal pid: no leading zero, ≤ 10 digits (?P [\x21-\x7e]{1,253}) # printable non-whitespace ASCII (RFC 1123 hostname limit) \n* # tolerate sloppy writers that append extra newlines \Z # end of string """, text, re.VERBOSE, ) if match is None: return None pid = int(match["pid"], 10) if pid > 2**31 - 1: return None return _MarkerInfo(token=match["token"], pid=pid, hostname=match["hostname"]) def _unlink(name: str, *, dir_fd: int | None = None) -> None: with suppress(FileNotFoundError): if _SUPPORTS_DIR_FD and dir_fd is not None: # pragma: needs dir-fd # Path.unlink has no dir_fd support, so we stay on os.unlink for the dirfd path. os.unlink(name, dir_fd=dir_fd) else: Path(name).unlink() def _break_stale_marker( # ruff:ignore[too-many-return-statements] # each return is a distinct abort/commit point in the break protocol name: str, *, stale_threshold: float, now: float, dir_fd: int | None = None, ) -> bool: # Atomic break pattern: read → rename to unique break-name → re-verify → unlink. The rename gives us a # private name nobody else can touch; if the re-verify sees a newer mtime or a different token, the # legitimate holder's heartbeat fired between read and rename and we must abort (leaving the .break.* # file behind rather than rollback-renaming, because rollback is itself racy). if (read_result := _read_marker(name, dir_fd=dir_fd)) is None: return False info_before, mtime_before = read_result if now - mtime_before <= stale_threshold: return False if info_before is None: _unlink(name, dir_fd=dir_fd) return True break_name = f"{name}{_BREAK_SUFFIX}.{os.getpid()}.{secrets.token_hex(16)}" try: if _SUPPORTS_DIR_FD and dir_fd is not None: # pragma: needs dir-fd os.rename(name, break_name, src_dir_fd=dir_fd, dst_dir_fd=dir_fd) else: Path(name).rename(break_name) except OSError: # pragma: no cover - race where the marker vanishes between read and rename return False read_after = _read_marker(break_name, dir_fd=dir_fd) if read_after is None: # pragma: no cover - race where a peer unlinks the break-name file return False info_after, mtime_after = read_after if info_after is None: # pragma: no cover - content replaced post-rename by a racing peer _unlink(break_name, dir_fd=dir_fd) return True if not hmac.compare_digest(info_before.token, info_after.token): # pragma: no cover - race only return False if mtime_after > mtime_before: # pragma: no cover - heartbeat raced our rename return False _unlink(break_name, dir_fd=dir_fd) return True def _atomic_create_marker(name: str, token: str, *, dir_fd: int | None = None) -> None: # O_NOFOLLOW blocks the symlink-overwrite attack where an attacker pre-creates the marker path as a # symlink pointing at a victim file. Mode 0o600 keeps the token unreadable to other users. flags = os.O_CREAT | os.O_EXCL | os.O_WRONLY | _O_NOFOLLOW if _SUPPORTS_DIR_FD and dir_fd is not None: # pragma: needs dir-fd fd = os.open(name, flags, 0o600, dir_fd=dir_fd) else: fd = os.open(name, flags, 0o600) # Write the whole record before the marker counts as created. On failure remove it only while the path still names # the file we opened, so a rollback never deletes a marker a concurrent reader recreated at this name. identity: tuple[int, int] | None = None try: st = os.fstat(fd) identity = st.st_dev, st.st_ino write_all(fd, f"{token}\n{os.getpid()}\n{socket.gethostname()}\n".encode("ascii")) except BaseException: os.close(fd) if identity is not None and _same_file(name, identity, dir_fd=dir_fd): _unlink(name, dir_fd=dir_fd) raise else: os.close(fd) def _same_file(name: str, identity: tuple[int, int], *, dir_fd: int | None) -> bool: try: st = os.lstat(name, dir_fd=dir_fd) if _SUPPORTS_DIR_FD and dir_fd is not None else os.lstat(name) except OSError: return False return (st.st_dev, st.st_ino) == identity def _file_exists(path: str) -> bool: try: st = os.lstat(path) except FileNotFoundError: return False return stat.S_ISREG(st.st_mode) def _is_housekeeping_name(name: str) -> bool: return name.startswith(".") or _BREAK_SUFFIX in name @dataclass(frozen=True) class _Paths: state: str write: str readers: str @dataclass class _Locks: internal: threading.Lock transaction: threading.Lock state: SoftFileLock @dataclass(frozen=True) class _MarkerInfo: token: str pid: int hostname: str @dataclass class _Hold: """Everything that exists only while a lock is held; ``None`` when the instance has no lock.""" level: int mode: _Mode write_thread_id: int | None marker_name: str is_reader: bool token: str heartbeat_thread: _HeartbeatThread heartbeat_stop: threading.Event def _cleanup_all_instances() -> None: # pragma: no cover - runs from atexit at interpreter shutdown for instance in list(_ALL_INSTANCES.values()): with suppress(Exception): instance.release(force=True) atexit.register(_cleanup_all_instances) _register_fork_class(SoftReadWriteLock) __all__ = [ "SoftReadWriteLock", ]