Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1"""Cross-process and cross-host reader/writer lock built on :class:`SoftFileLock` primitives."""
3from __future__ import annotations
5import atexit
6import hmac
7import os
8import re
9import secrets
10import socket
11import stat
12import sys
13import threading
14import time
15import uuid
16from contextlib import contextmanager, suppress
17from dataclasses import dataclass
18from pathlib import Path
19from typing import TYPE_CHECKING, Literal
20from weakref import WeakValueDictionary
22from filelock._api import AcquireReturnProxy
23from filelock._error import Timeout
24from filelock._soft import SoftFileLock
25from filelock._util import ensure_directory_exists
27if TYPE_CHECKING:
28 from collections.abc import Callable, Generator
31_Mode = Literal["read", "write"]
32_BREAK_SUFFIX = ".break"
33_MAX_MARKER_SIZE = 1024
34_O_NOFOLLOW = getattr(os, "O_NOFOLLOW", 0)
35_O_NONBLOCK = getattr(os, "O_NONBLOCK", 0)
36# Retargeting os.utime to an open fd lets the heartbeat refresh the exact inode it just verified instead of
37# re-resolving the path; Windows read-only handles cannot set times that way, so keep it Unix-only.
38_SUPPORTS_UTIME_FD = sys.platform != "win32" and os.utime in os.supports_fd
39# os.utime follows symlinks unless told not to; not every platform can refuse the follow, so probe support.
40_SUPPORTS_UTIME_NOFOLLOW = os.utime in os.supports_follow_symlinks
41# dirfd-relative I/O is a Unix-only optimization; Windows cannot ``os.open()`` a directory at all, and
42# its ``os`` module skips dir_fd support entirely. When disabled, callers fall back to full-path ops.
43_SUPPORTS_DIR_FD = sys.platform != "win32" and os.open in os.supports_dir_fd
45_all_instances: WeakValueDictionary[Path, SoftReadWriteLock] = WeakValueDictionary()
46_all_instances_lock = threading.Lock()
47_atexit_registered = False
48_fork_registered = False
51@dataclass(frozen=True)
52class _Paths:
53 state: str
54 write: str
55 readers: str
58@dataclass
59class _Locks:
60 internal: threading.Lock
61 transaction: threading.Lock
62 state: SoftFileLock
65@dataclass(frozen=True)
66class _MarkerInfo:
67 token: str
68 pid: int
69 hostname: str
72@dataclass
73class _Hold:
74 """Everything that exists only while a lock is held; ``None`` when the instance has no lock."""
76 level: int
77 mode: _Mode
78 write_thread_id: int | None
79 marker_name: str
80 is_reader: bool
81 token: str
82 heartbeat_thread: _HeartbeatThread
83 heartbeat_stop: threading.Event
86class _SoftRWMeta(type):
87 _instances: WeakValueDictionary[Path, SoftReadWriteLock]
88 _instances_lock: threading.Lock
90 def __call__( # noqa: PLR0913
91 cls,
92 lock_file: str | os.PathLike[str],
93 timeout: float = -1,
94 *,
95 blocking: bool = True,
96 is_singleton: bool = True,
97 heartbeat_interval: float = 30.0,
98 stale_threshold: float | None = None,
99 poll_interval: float = 0.25,
100 ) -> SoftReadWriteLock:
101 if not is_singleton:
102 return super().__call__(
103 lock_file,
104 timeout,
105 blocking=blocking,
106 is_singleton=is_singleton,
107 heartbeat_interval=heartbeat_interval,
108 stale_threshold=stale_threshold,
109 poll_interval=poll_interval,
110 )
112 normalized = Path(lock_file).resolve()
113 with cls._instances_lock:
114 instance = cls._instances.get(normalized)
115 if instance is None:
116 instance = super().__call__(
117 lock_file,
118 timeout,
119 blocking=blocking,
120 is_singleton=is_singleton,
121 heartbeat_interval=heartbeat_interval,
122 stale_threshold=stale_threshold,
123 poll_interval=poll_interval,
124 )
125 cls._instances[normalized] = instance
126 elif instance.timeout != timeout or instance.blocking != blocking:
127 msg = (
128 f"Singleton lock created with timeout={instance.timeout}, blocking={instance.blocking},"
129 f" cannot be changed to timeout={timeout}, blocking={blocking}"
130 )
131 raise ValueError(msg)
132 return instance
135class SoftReadWriteLock(metaclass=_SoftRWMeta):
136 """
137 Cross-process and cross-host reader/writer lock built on :class:`SoftFileLock` primitives.
139 Use this class instead of :class:`~filelock.ReadWriteLock` when the lock file lives on a network
140 filesystem (NFS, Lustre with ``-o flock``, HPC cluster shared storage). ``ReadWriteLock`` is backed
141 by SQLite and cannot run on NFS because SQLite's ``fcntl`` locking is unreliable there.
143 Layout on disk for a lock at ``foo.lock``:
145 - ``foo.lock.state`` — a :class:`SoftFileLock` taken only during state transitions (microseconds).
146 - ``foo.lock.write`` — writer marker; its presence means a writer is claiming or holding the lock.
147 - ``foo.lock.readers/<host>.<pid>.<uuid>`` — one file per reader.
149 Each marker stores a random token (``secrets.token_hex(16)``), the holder's pid, and the holder's
150 hostname. A daemon heartbeat thread refreshes ``mtime`` on every held marker. A marker whose mtime
151 has not advanced in ``stale_threshold`` seconds may be evicted by any process on any host, giving
152 correct behavior when a compute node crashes with a lock held.
154 Writer acquire is two-phase and writer-preferring: phase 1 claims ``.write`` (blocking any new
155 reader), phase 2 waits for existing readers to drain. Writer starvation is impossible.
157 Reentrancy, upgrade/downgrade rules, thread pinning, and singleton caching by resolved path match
158 :class:`~filelock.ReadWriteLock`.
160 Forking while holding a lock invalidates the inherited instance in the child so the child cannot
161 double-own the lock with its parent; ``release()`` on a fork-invalidated instance is a no-op, and
162 the child must re-acquire if it needs a lock.
164 Trust boundary: protects against same-UID non-cooperating processes (one host or cross-host) and
165 same-host different-UID users via ``0o600`` / ``0o700`` permissions. Does not protect against root
166 compromise, NTP tampering on same-UID cross-host nodes, or multi-tenant mounts where hostile
167 co-tenants share the UID.
169 :param lock_file: path to the lock file; sidecar state/write/readers live next to it
170 :param timeout: maximum wait time in seconds; ``-1`` means block indefinitely
171 :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately on contention
172 :param is_singleton: if ``True``, reuse existing instances for the same resolved path
173 :param heartbeat_interval: seconds between heartbeat refreshes; default 30 s
174 :param stale_threshold: seconds of ``mtime`` inactivity before a marker is stale; defaults to
175 ``3 * heartbeat_interval``, matching etcd's ``LeaseKeepAlive`` convention
176 :param poll_interval: seconds between acquire retries under contention; default 0.25 s
178 .. versionadded:: 3.27.0
180 """
182 _instances: WeakValueDictionary[Path, SoftReadWriteLock] = WeakValueDictionary()
183 _instances_lock = threading.Lock()
185 def __init__( # noqa: PLR0913
186 self,
187 lock_file: str | os.PathLike[str],
188 timeout: float = -1,
189 *,
190 blocking: bool = True,
191 is_singleton: bool = True, # noqa: ARG002
192 heartbeat_interval: float = 30.0,
193 stale_threshold: float | None = None,
194 poll_interval: float = 0.25,
195 ) -> None:
196 if heartbeat_interval <= 0:
197 msg = f"heartbeat_interval must be positive, got {heartbeat_interval}"
198 raise ValueError(msg)
199 if stale_threshold is None:
200 stale_threshold = heartbeat_interval * 3
201 if stale_threshold <= heartbeat_interval:
202 msg = f"stale_threshold must exceed heartbeat_interval ({stale_threshold} <= {heartbeat_interval})"
203 raise ValueError(msg)
204 if poll_interval <= 0:
205 msg = f"poll_interval must be positive, got {poll_interval}"
206 raise ValueError(msg)
208 self.lock_file: str = os.fspath(lock_file)
209 self.timeout: float = timeout
210 self.blocking: bool = blocking
211 self.heartbeat_interval: float = heartbeat_interval
212 self.stale_threshold: float = stale_threshold
213 self.poll_interval: float = poll_interval
215 self._paths = _Paths(
216 state=f"{self.lock_file}.state",
217 write=f"{self.lock_file}.write",
218 readers=f"{self.lock_file}.readers",
219 )
220 ensure_directory_exists(self.lock_file)
221 self._locks = _Locks(
222 internal=threading.Lock(),
223 transaction=threading.Lock(),
224 state=SoftFileLock(self._paths.state, timeout=-1),
225 )
226 self._readers_dir_fd: int | None = None
227 self._hold: _Hold | None = None
228 self._fork_invalidated: bool = False
229 self._closed: bool = False
231 with _all_instances_lock:
232 _all_instances[Path(self.lock_file).resolve()] = self
233 _register_hooks()
235 @contextmanager
236 def read_lock(self, timeout: float | None = None, *, blocking: bool | None = None) -> Generator[None]:
237 """
238 Context manager that acquires and releases a shared read lock.
240 Falls back to instance defaults for *timeout* and *blocking* when ``None``.
242 :param timeout: maximum wait time in seconds, or ``None`` to use the instance default
243 :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately; ``None`` uses the instance default
245 :raises RuntimeError: if a write lock is already held on this instance
246 :raises Timeout: if the lock cannot be acquired within *timeout* seconds
248 """
249 self.acquire_read(timeout, blocking=blocking)
250 try:
251 yield
252 finally:
253 self.release()
255 @contextmanager
256 def write_lock(self, timeout: float | None = None, *, blocking: bool | None = None) -> Generator[None]:
257 """
258 Context manager that acquires and releases an exclusive write lock.
260 Falls back to instance defaults for *timeout* and *blocking* when ``None``.
262 :param timeout: maximum wait time in seconds, or ``None`` to use the instance default
263 :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately; ``None`` uses the instance default
265 :raises RuntimeError: if a read lock is already held, or a write lock is held by a different thread
266 :raises Timeout: if the lock cannot be acquired within *timeout* seconds
268 """
269 self.acquire_write(timeout, blocking=blocking)
270 try:
271 yield
272 finally:
273 self.release()
275 def acquire_read(self, timeout: float | None = None, *, blocking: bool | None = None) -> AcquireReturnProxy:
276 """
277 Acquire a shared read lock.
279 If this instance already holds a read lock, the lock level is incremented (reentrant). Attempting to acquire a
280 read lock while holding a write lock raises :class:`RuntimeError` (downgrade not allowed). On the 0→1
281 transition a daemon heartbeat thread is started that refreshes the reader marker's ``mtime`` every
282 ``heartbeat_interval`` seconds so peers on other hosts do not evict the marker as stale.
284 :param timeout: maximum wait time in seconds, or ``None`` to use the instance default; ``-1`` means block
285 indefinitely
286 :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately when the lock is unavailable;
287 ``None`` uses the instance default
289 :returns: a proxy that can be used as a context manager to release the lock
291 :raises RuntimeError: if a write lock is already held on this instance, if this instance was invalidated by
292 :func:`os.fork`, or if :meth:`close` was called
293 :raises Timeout: if the lock cannot be acquired within *timeout* seconds
295 """
296 return self._acquire("read", timeout, blocking=blocking)
298 def acquire_write(self, timeout: float | None = None, *, blocking: bool | None = None) -> AcquireReturnProxy:
299 """
300 Acquire an exclusive write lock.
302 If this instance already holds a write lock from the same thread, the lock level is incremented (reentrant).
303 Attempting to acquire a write lock while holding a read lock raises :class:`RuntimeError` (upgrade not
304 allowed). Write locks are pinned to the acquiring thread: a different thread trying to re-enter also raises
305 :class:`RuntimeError`.
307 Writer acquisition runs in two phases. Phase 1 atomically claims ``<path>.write`` via ``O_CREAT | O_EXCL``,
308 which immediately blocks any new reader on any host. Phase 2 waits for existing readers to drain. Writer
309 starvation is impossible: new readers see ``<path>.write`` during phase 2 and wait behind the pending writer.
311 :param timeout: maximum wait time in seconds, or ``None`` to use the instance default; ``-1`` means block
312 indefinitely
313 :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately when the lock is unavailable;
314 ``None`` uses the instance default
316 :returns: a proxy that can be used as a context manager to release the lock
318 :raises RuntimeError: if a read lock is already held, if a write lock is held by a different thread, if this
319 instance was invalidated by :func:`os.fork`, or if :meth:`close` was called
320 :raises Timeout: if the lock cannot be acquired within *timeout* seconds
322 """
323 return self._acquire("write", timeout, blocking=blocking)
325 def close(self) -> None:
326 """
327 Release any held lock and release internal filesystem resources.
329 Idempotent. After calling this method the instance can no longer acquire locks — subsequent acquires raise
330 :class:`RuntimeError`. A fork-invalidated instance is closed without raising.
331 """
332 self.release(force=True)
333 with self._locks.internal:
334 if self._closed:
335 return
336 self._closed = True
337 if self._readers_dir_fd is not None:
338 with suppress(OSError):
339 os.close(self._readers_dir_fd)
340 self._readers_dir_fd = None
342 def release(self, *, force: bool = False) -> None:
343 """
344 Release one level of the current lock.
346 When the lock level reaches zero the heartbeat thread is stopped and the held marker file is unlinked. On a
347 fork-invalidated instance (that is, the child of a :func:`os.fork` call made while the parent held a lock)
348 this method is a no-op so inherited ``with`` blocks can unwind cleanly in the child.
350 :param force: if ``True``, release the lock completely regardless of the current lock level
352 :raises RuntimeError: if no lock is currently held and *force* is ``False``
354 """
355 with self._locks.internal:
356 if self._fork_invalidated:
357 # Inherited state from the parent is meaningless in the child; clear any counters and return.
358 self._hold = None
359 return
360 hold = self._hold
361 if hold is None:
362 if force:
363 return
364 msg = f"Cannot release a lock on {self.lock_file} (lock id: {id(self)}) that is not held"
365 raise RuntimeError(msg)
366 if force:
367 hold.level = 0
368 else:
369 hold.level -= 1
370 if hold.level > 0:
371 return
372 self._hold = None
374 # Order matters: signal → join → unlink. A late tick on a deleted marker is harmless, and the
375 # token check in the heartbeat callback would catch any re-acquisition race, but joining first
376 # removes even that theoretical race.
377 hold.heartbeat_stop.set()
378 hold.heartbeat_thread.join(timeout=self.heartbeat_interval + 1.0)
379 if hold.is_reader:
380 _unlink(hold.marker_name, dir_fd=self._readers_dir_fd)
381 else:
382 self._unlink_writer_marker_if_ours(hold.token)
384 def _unlink_writer_marker_if_ours(self, token: str) -> None:
385 # Remove the writer marker only while it still carries our token. If this holder was paused long
386 # enough (a stop-the-world GC pause, SIGSTOP, a suspended VM) for a peer to evict the marker as
387 # stale and claim the writer slot itself, the file now at <path>.write is the peer's live marker;
388 # unlinking it by path would let a second writer through and break mutual exclusion. The state lock
389 # serializes this against a concurrent break/claim, and the heartbeat is already stopped, so the
390 # token we read is authoritative. Mirrors the token re-check the stale-break path already does.
391 with self._locks.state:
392 read = _read_marker(self._paths.write)
393 if read is None:
394 return
395 info, _ = read
396 if info is None or not hmac.compare_digest(info.token, token):
397 return
398 _unlink(self._paths.write)
400 def _claim_writer_marker(self, token: str) -> bool:
401 # Claim the writer slot for ``token``. Must be called holding ``self._locks.state``. Evicts a
402 # stale marker first, then refuses to claim while a live ``.write`` exists so a peer holding the
403 # slot is waited out instead of overwritten.
404 _break_stale_marker(self._paths.write, stale_threshold=self.stale_threshold, now=time.time())
405 if _file_exists(self._paths.write):
406 return False
407 try:
408 _atomic_create_marker(self._paths.write, token)
409 except FileExistsError:
410 return False
411 return True
413 def _touch_writer_marker_if_ours(self, token: str) -> bool:
414 # Refresh the writer marker through a single O_NOFOLLOW fd, but only while it still carries our
415 # token. Returns False when the marker is gone or now belongs to a peer that reclaimed the slot,
416 # so the caller can re-claim rather than keep a stranger's marker alive. Mirrors _refresh_marker.
417 fd = _open_marker(self._paths.write)
418 if fd is None:
419 return False
420 try:
421 try:
422 data = os.read(fd, _MAX_MARKER_SIZE + 1)
423 except OSError: # pragma: no cover - e.g. EAGAIN from a hostile FIFO that has a writer attached
424 return False
425 info = _parse_marker_bytes(data)
426 if info is None or not hmac.compare_digest(info.token, token):
427 return False
428 with suppress(OSError):
429 _touch(self._paths.write, fd=fd)
430 return True
431 finally:
432 os.close(fd)
434 @classmethod
435 def get_lock(
436 cls,
437 lock_file: str | os.PathLike[str],
438 timeout: float = -1,
439 *,
440 blocking: bool = True,
441 ) -> SoftReadWriteLock:
442 """
443 Return the singleton :class:`SoftReadWriteLock` for *lock_file*.
445 :param lock_file: path to the lock file; sidecar state/write/readers live next to it
446 :param timeout: maximum wait time in seconds; ``-1`` means block indefinitely
447 :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately when the lock is unavailable
449 :returns: the singleton lock instance
451 :raises ValueError: if an instance already exists for this path with different *timeout* or *blocking* values
453 """
454 return cls(lock_file, timeout, blocking=blocking)
456 def _acquire(
457 self,
458 mode: _Mode,
459 timeout: float | None,
460 *,
461 blocking: bool | None,
462 ) -> AcquireReturnProxy:
463 timeout = self.timeout if timeout is None else timeout
464 blocking = self.blocking if blocking is None else blocking
466 with self._locks.internal:
467 if self._fork_invalidated:
468 msg = f"SoftReadWriteLock on {self.lock_file} was invalidated by fork(); construct a new instance"
469 raise RuntimeError(msg)
470 if self._closed:
471 msg = f"SoftReadWriteLock on {self.lock_file} has been closed"
472 raise RuntimeError(msg)
473 if self._hold is not None:
474 return self._validate_reentrant(mode)
476 start = time.perf_counter()
477 if not blocking:
478 acquired = self._locks.transaction.acquire(blocking=False)
479 elif timeout == -1:
480 acquired = self._locks.transaction.acquire(blocking=True)
481 else:
482 acquired = self._locks.transaction.acquire(blocking=True, timeout=timeout)
483 if not acquired:
484 raise Timeout(self.lock_file) from None
485 try:
486 return self._do_acquire_inner(mode, timeout, start, blocking=blocking)
487 finally:
488 self._locks.transaction.release()
490 def _do_acquire_inner(
491 self,
492 mode: _Mode,
493 effective_timeout: float,
494 start: float,
495 *,
496 blocking: bool,
497 ) -> AcquireReturnProxy:
498 with self._locks.internal:
499 if self._hold is not None:
500 return self._validate_reentrant(mode)
501 deadline = None if effective_timeout == -1 else start + effective_timeout
502 token = secrets.token_hex(16)
503 if mode == "write":
504 marker_name, is_reader = self._acquire_writer_slot(token, deadline=deadline, blocking=blocking)
505 else:
506 marker_name, is_reader = self._acquire_reader_slot(token, deadline=deadline, blocking=blocking)
507 stop_event = threading.Event()
508 heartbeat = _HeartbeatThread(
509 refresh=self._refresh_marker,
510 interval=self.heartbeat_interval,
511 stop_event=stop_event,
512 name=f"filelock-heartbeat-{id(self):x}",
513 )
514 with self._locks.internal:
515 self._hold = _Hold(
516 level=1,
517 mode=mode,
518 write_thread_id=threading.get_ident() if mode == "write" else None,
519 marker_name=marker_name,
520 is_reader=is_reader,
521 token=token,
522 heartbeat_thread=heartbeat,
523 heartbeat_stop=stop_event,
524 )
525 heartbeat.start()
526 return AcquireReturnProxy(lock=self)
528 def _validate_reentrant(self, mode: _Mode) -> AcquireReturnProxy:
529 hold = self._hold
530 assert hold is not None # noqa: S101
531 if hold.mode != mode:
532 opposite = "write" if mode == "read" else "read"
533 direction = "downgrade" if mode == "read" else "upgrade"
534 msg = (
535 f"Cannot acquire {mode} lock on {self.lock_file} (lock id: {id(self)}): "
536 f"already holding a {opposite} lock ({direction} not allowed)"
537 )
538 raise RuntimeError(msg)
539 if mode == "write" and (cur := threading.get_ident()) != hold.write_thread_id:
540 msg = (
541 f"Cannot acquire write lock on {self.lock_file} (lock id: {id(self)}) "
542 f"from thread {cur} while it is held by thread {hold.write_thread_id}"
543 )
544 raise RuntimeError(msg)
545 hold.level += 1
546 return AcquireReturnProxy(lock=self)
548 def _acquire_writer_slot(
549 self,
550 token: str,
551 *,
552 deadline: float | None,
553 blocking: bool,
554 ) -> tuple[str, bool]:
555 # Phase 2 scans readers/ via dirfd (where supported), so we need it open even though writers never
556 # create files inside.
557 self._open_readers_dir()
559 def try_claim_writer() -> bool:
560 with self._locks.state:
561 return self._claim_writer_marker(token)
563 def readers_drained_touching() -> bool:
564 with self._locks.state:
565 # Refresh our writer marker every scan iteration so phase 2 does not exceed
566 # ``stale_threshold`` under contention and get evicted. The refresh only happens while
567 # the marker is still ours: if we were paused past ``stale_threshold`` a peer can evict
568 # the stale marker and reclaim ``.write`` with its own token, and touching that path
569 # blindly would keep the peer's live marker alive and let this acquire finish as though
570 # we still held the slot, admitting a second writer. When the claim is no longer ours we
571 # re-claim the slot here (waiting behind the peer if it currently holds it) rather than
572 # trusting the foreign marker, mirroring the token re-check the stale-break and release
573 # paths already rely on.
574 if not self._touch_writer_marker_if_ours(token) and not self._claim_writer_marker(token):
575 return False
576 self._break_stale_readers(time.time())
577 return not self._any_readers()
579 self._wait_for(try_claim_writer, deadline=deadline, blocking=blocking)
580 try:
581 self._wait_for(readers_drained_touching, deadline=deadline, blocking=blocking)
582 except Timeout:
583 # Give up our writer claim so readers can make progress again, but only while the marker is
584 # still ours: a peer may have evicted it as stale and claimed the slot while phase 2 waited.
585 self._unlink_writer_marker_if_ours(token)
586 raise
587 return self._paths.write, False
589 def _acquire_reader_slot(
590 self,
591 token: str,
592 *,
593 deadline: float | None,
594 blocking: bool,
595 ) -> tuple[str, bool]:
596 self._open_readers_dir()
597 reader_name = f"{uuid.uuid4().hex}.{os.getpid()}"
598 dir_fd = self._readers_dir_fd
599 full_reader_path = str(Path(self._paths.readers) / reader_name)
601 def try_claim_reader() -> bool:
602 with self._locks.state:
603 _break_stale_marker(self._paths.write, stale_threshold=self.stale_threshold, now=time.time())
604 if _file_exists(self._paths.write):
605 return False
606 if dir_fd is not None:
607 _atomic_create_marker(reader_name, token, dir_fd=dir_fd)
608 else: # pragma: win32 cover
609 _atomic_create_marker(full_reader_path, token)
610 return True
612 self._wait_for(try_claim_reader, deadline=deadline, blocking=blocking)
613 return (reader_name if dir_fd is not None else full_reader_path), True
615 def _wait_for(
616 self,
617 predicate: Callable[[], bool],
618 *,
619 deadline: float | None,
620 blocking: bool,
621 ) -> None:
622 while True:
623 if predicate():
624 return
625 now = time.perf_counter()
626 if not blocking:
627 raise Timeout(self.lock_file)
628 if deadline is not None and now >= deadline:
629 raise Timeout(self.lock_file)
630 sleep_for = self.poll_interval
631 if deadline is not None:
632 sleep_for = min(sleep_for, max(deadline - now, 0.0))
633 time.sleep(sleep_for)
635 def _open_readers_dir(self) -> None:
636 readers_path = Path(self._paths.readers)
637 with suppress(FileExistsError):
638 readers_path.mkdir(mode=0o700)
639 # mkdir has no O_NOFOLLOW, so verify via lstat that we did not land on an attacker-placed symlink
640 # or a regular file before we open or scan inside.
641 st = os.lstat(self._paths.readers)
642 if stat.S_ISLNK(st.st_mode) or not stat.S_ISDIR(st.st_mode):
643 msg = f"{self._paths.readers} exists but is not a directory or is a symlink; refusing to use it"
644 raise RuntimeError(msg)
645 if self._readers_dir_fd is None and _SUPPORTS_DIR_FD:
646 flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | _O_NOFOLLOW
647 self._readers_dir_fd = os.open(self._paths.readers, flags)
649 def _any_readers(self) -> bool:
650 for _ in self._iter_reader_entries():
651 return True
652 return False
654 def _iter_reader_entries(self) -> Generator[tuple[str, bool]]:
655 """
656 Yield ``(name, dirfd_relative)`` pairs for every live reader marker.
658 ``dirfd_relative`` is ``True`` when *name* should be passed to ``dir_fd=``-aware syscalls; ``False``
659 when *name* is a full path because dirfd-relative I/O is unavailable on this platform.
660 """
661 if self._readers_dir_fd is not None:
662 with os.scandir(self._readers_dir_fd) as it:
663 for entry in it:
664 if not _is_housekeeping_name(entry.name):
665 yield entry.name, True
666 return
667 readers_path = Path(self._paths.readers) # pragma: win32 cover
668 with os.scandir(readers_path) as it: # pragma: win32 cover
669 for entry in it: # pragma: win32 cover
670 if not _is_housekeeping_name(entry.name): # pragma: win32 cover
671 yield str(readers_path / entry.name), False # pragma: win32 cover
673 def _break_stale_readers(self, now: float) -> None:
674 names: list[tuple[str, int | None]] = []
675 try:
676 for name, dirfd_relative in self._iter_reader_entries():
677 names.append((name, self._readers_dir_fd if dirfd_relative else None))
678 except OSError: # pragma: no cover - transient NFS scandir hiccup
679 return
680 for name, fd in names:
681 _break_stale_marker(name, stale_threshold=self.stale_threshold, now=now, dir_fd=fd)
683 def _refresh_marker(self) -> bool:
684 with self._locks.internal:
685 hold = self._hold
686 if hold is None: # pragma: no cover - race between stop_event.set and join
687 return False
688 marker_name = hold.marker_name
689 token = hold.token
690 dir_fd = self._readers_dir_fd if hold.is_reader else None
692 # Open once with O_NOFOLLOW and touch that exact descriptor. Doing the refresh through the verified fd
693 # (instead of re-opening by name) closes the window where a peer unlinks our marker and drops a symlink
694 # or a different file at the path between the read and the touch: utime then lands on the inode we
695 # verified, or nowhere. A failed open means the marker is gone or was swapped out -- a peer evicted us
696 # -- so stop the heartbeat at once instead of waiting a tick.
697 fd = _open_marker(marker_name, dir_fd=dir_fd)
698 if fd is None:
699 return False
700 try:
701 try:
702 data = os.read(fd, _MAX_MARKER_SIZE + 1)
703 except OSError: # pragma: no cover - e.g. EAGAIN from a hostile FIFO that has a writer attached
704 return False
705 info = _parse_marker_bytes(data)
706 # Token mismatch means another process already evicted our marker and created its own; stop the
707 # thread so it does not keep a stranger's file alive.
708 if info is None or not hmac.compare_digest(info.token, token):
709 return False
710 # A transient touch failure (ESTALE / EIO on the NFS-style filesystems this lock targets) must not
711 # kill the heartbeat thread: the read above just confirmed the marker is still ours, so swallow the
712 # error and retry on the next tick rather than letting the lease lapse while we still hold the lock.
713 with suppress(OSError):
714 _touch(marker_name, fd=fd)
715 return True
716 finally:
717 os.close(fd)
719 def _reset_after_fork_in_child(self) -> None: # pragma: no cover - fork child not tracked
720 # Replace every lock this instance owns with a fresh one; the inherited locks may still be held
721 # by threads that no longer exist in the child. The readers dirfd and the SoftFileLock state
722 # mutex both get dropped for the same reason — the child re-creates them on its next acquire.
723 self._locks = _Locks(
724 internal=threading.Lock(),
725 transaction=threading.Lock(),
726 state=SoftFileLock(self._paths.state, timeout=-1),
727 )
728 self._hold = None
729 self._readers_dir_fd = None
730 self._fork_invalidated = True
733class _HeartbeatThread(threading.Thread):
734 def __init__(
735 self,
736 refresh: Callable[[], bool],
737 interval: float,
738 stop_event: threading.Event,
739 name: str,
740 ) -> None:
741 super().__init__(name=name, daemon=True)
742 self._refresh = refresh
743 self._interval = interval
744 self._stop_event = stop_event
746 def run(self) -> None:
747 while not self._stop_event.wait(self._interval):
748 if not self._refresh():
749 self._stop_event.set()
750 return
753def _atomic_create_marker(name: str, token: str, *, dir_fd: int | None = None) -> None:
754 # O_NOFOLLOW blocks the symlink-overwrite attack where an attacker pre-creates the marker path as a
755 # symlink pointing at a victim file. Mode 0o600 keeps the token unreadable to other users.
756 flags = os.O_CREAT | os.O_EXCL | os.O_WRONLY | _O_NOFOLLOW
757 if _SUPPORTS_DIR_FD and dir_fd is not None:
758 fd = os.open(name, flags, 0o600, dir_fd=dir_fd)
759 else:
760 fd = os.open(name, flags, 0o600)
761 try:
762 content = f"{token}\n{os.getpid()}\n{socket.gethostname()}\n".encode("ascii")
763 os.write(fd, content)
764 finally:
765 os.close(fd)
768def _open_marker(name: str, *, dir_fd: int | None = None) -> int | None:
769 # The file is ours; these guard a hostile mid-flight swap. O_NOFOLLOW rejects a symlink; O_NONBLOCK keeps
770 # a real FIFO from blocking the open forever, so it reads as a malformed marker instead of wedging a peer
771 # that holds the state lock.
772 flags = os.O_RDONLY | _O_NOFOLLOW | _O_NONBLOCK
773 try:
774 return os.open(name, flags, dir_fd=dir_fd) if _SUPPORTS_DIR_FD and dir_fd is not None else os.open(name, flags)
775 except OSError:
776 return None
779def _read_marker(name: str, *, dir_fd: int | None = None) -> tuple[_MarkerInfo | None, float] | None:
780 fd = _open_marker(name, dir_fd=dir_fd)
781 if fd is None:
782 return None
783 try:
784 st = os.fstat(fd)
785 # A legitimate marker is always a regular file, so anything else at the path -- above all a FIFO -- is
786 # reported as a malformed marker (its mtime still drives stale eviction) without being read. Reading is
787 # where platforms diverge: an empty non-blocking read yields 0 bytes on Linux/macOS but EAGAIN on
788 # FreeBSD, and the EAGAIN used to abort the stale-break and wedge the acquire until timeout (#587).
789 if not stat.S_ISREG(st.st_mode):
790 return None, st.st_mtime
791 data = os.read(fd, _MAX_MARKER_SIZE + 1)
792 except OSError: # pragma: no cover - marker vanished or turned unreadable between open and read
793 return None
794 finally:
795 os.close(fd)
796 return _parse_marker_bytes(data), st.st_mtime
799def _parse_marker_bytes(data: bytes) -> _MarkerInfo | None:
800 # Trust nothing about attacker-controlled markers; any deviation returns None so callers fall through
801 # to stale cleanup. ``re.match`` caches compiled patterns internally, so the regex is built only once
802 # despite being defined inline.
803 if not data or len(data) > _MAX_MARKER_SIZE:
804 return None
805 try:
806 text = data.decode("ascii")
807 except UnicodeDecodeError:
808 return None
809 match = re.match(
810 r"""
811 \A # start of string
812 (?P<token> [0-9a-f]{32} ) \n # 128-bit hex token
813 (?P<pid> [1-9][0-9]{0,9} ) \n # decimal pid: no leading zero, ≤ 10 digits
814 (?P<hostname> [\x21-\x7e]{1,253}) # printable non-whitespace ASCII (RFC 1123 hostname limit)
815 \n* # tolerate sloppy writers that append extra newlines
816 \Z # end of string
817 """,
818 text,
819 re.VERBOSE,
820 )
821 if match is None:
822 return None
823 pid = int(match["pid"], 10)
824 if pid > 2**31 - 1:
825 return None
826 return _MarkerInfo(token=match["token"], pid=pid, hostname=match["hostname"])
829def _break_stale_marker( # noqa: PLR0911
830 name: str,
831 *,
832 stale_threshold: float,
833 now: float,
834 dir_fd: int | None = None,
835) -> bool:
836 # Atomic break pattern: read → rename to unique break-name → re-verify → unlink. The rename gives us a
837 # private name nobody else can touch; if the re-verify sees a newer mtime or a different token, the
838 # legitimate holder's heartbeat fired between read and rename and we must abort (leaving the .break.*
839 # file behind rather than rollback-renaming, because rollback is itself racy).
840 read_result = _read_marker(name, dir_fd=dir_fd)
841 if read_result is None:
842 return False
843 info_before, mtime_before = read_result
844 if now - mtime_before <= stale_threshold:
845 return False
846 if info_before is None:
847 _unlink(name, dir_fd=dir_fd)
848 return True
850 break_name = f"{name}{_BREAK_SUFFIX}.{os.getpid()}.{secrets.token_hex(16)}"
851 try:
852 if _SUPPORTS_DIR_FD and dir_fd is not None:
853 os.rename(name, break_name, src_dir_fd=dir_fd, dst_dir_fd=dir_fd)
854 else:
855 Path(name).rename(break_name)
856 except OSError: # pragma: no cover - race where the marker vanishes between read and rename
857 return False
859 read_after = _read_marker(break_name, dir_fd=dir_fd)
860 if read_after is None: # pragma: no cover - race where a peer unlinks the break-name file
861 return False
862 info_after, mtime_after = read_after
863 if info_after is None: # pragma: no cover - content replaced post-rename by a racing peer
864 _unlink(break_name, dir_fd=dir_fd)
865 return True
866 if not hmac.compare_digest(info_before.token, info_after.token): # pragma: no cover - race only
867 return False
868 if mtime_after > mtime_before: # pragma: no cover - heartbeat raced our rename
869 return False
870 _unlink(break_name, dir_fd=dir_fd)
871 return True
874def _unlink(name: str, *, dir_fd: int | None = None) -> None:
875 with suppress(FileNotFoundError):
876 if _SUPPORTS_DIR_FD and dir_fd is not None:
877 # Path.unlink has no dir_fd support, so we stay on os.unlink for the dirfd path.
878 os.unlink(name, dir_fd=dir_fd)
879 else:
880 Path(name).unlink()
883def _touch(name: str, *, fd: int | None = None) -> None:
884 # Prefer the already-open, already-verified fd so a peer that swaps a symlink or a different file in at the
885 # path after our O_NOFOLLOW read cannot redirect the touch: utime then targets the inode behind the fd.
886 # Where the platform cannot utime an fd, fall back to a path-based touch that still refuses to follow a
887 # symlink where supported -- matching the O_NOFOLLOW reads used everywhere else here.
888 if fd is not None and _SUPPORTS_UTIME_FD:
889 os.utime(fd, None)
890 return
891 os.utime(name, None, follow_symlinks=not _SUPPORTS_UTIME_NOFOLLOW)
894def _file_exists(path: str) -> bool:
895 try:
896 st = os.lstat(path)
897 except FileNotFoundError:
898 return False
899 return stat.S_ISREG(st.st_mode)
902def _is_housekeeping_name(name: str) -> bool:
903 return name.startswith(".") or _BREAK_SUFFIX in name
906def _reset_all_after_fork() -> None: # pragma: no cover - fork child, not tracked by coverage
907 global _all_instances_lock # noqa: PLW0603
908 # User-created threading locks do not auto-reset across fork: any lock held by a parent thread stays
909 # locked in the child with no owner to release it. Replace the module-level lock and every instance's
910 # locks with fresh ones; the child is single-threaded at this point so no synchronization is needed.
911 _all_instances_lock = threading.Lock()
912 for instance in list(_all_instances.values()):
913 instance._reset_after_fork_in_child() # noqa: SLF001
916def _cleanup_all_instances() -> None: # pragma: no cover - runs from atexit at interpreter shutdown
917 for instance in list(_all_instances.values()):
918 with suppress(Exception):
919 instance.release(force=True)
922def _register_hooks() -> None:
923 global _atexit_registered, _fork_registered # noqa: PLW0603
924 if not _atexit_registered:
925 atexit.register(_cleanup_all_instances)
926 _atexit_registered = True
927 # after_in_child replaces inherited state so the child cannot double-own any lock the parent held.
928 if not _fork_registered and hasattr(os, "register_at_fork"):
929 os.register_at_fork(after_in_child=_reset_all_after_fork)
930 _fork_registered = True
933__all__ = [
934 "SoftReadWriteLock",
935]