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 closing, contextmanager, suppress
17from dataclasses import dataclass
18from pathlib import Path
19from typing import TYPE_CHECKING, Final, Literal
20from weakref import WeakValueDictionary
22from filelock._api import (
23 AcquireReturnProxy,
24 _ensure_current_process,
25 _fork_transition,
26 _raise_grouped_errors,
27 _register_fork_class,
28 _register_fork_object,
29 _register_owned_descriptor,
30 _unregister_owned_descriptor,
31)
32from filelock._error import Timeout
33from filelock._soft import SoftFileLock
34from filelock._util import ensure_directory_exists, touch, write_all
36if TYPE_CHECKING:
37 from collections.abc import Callable, Generator
40_Mode = Literal["read", "write"]
41_BREAK_SUFFIX: Final[str] = ".break"
42_MAX_MARKER_SIZE: Final[int] = 1024
43_O_NOFOLLOW: Final[int] = getattr(os, "O_NOFOLLOW", 0)
44_O_NONBLOCK: Final[int] = getattr(os, "O_NONBLOCK", 0)
45# dirfd-relative I/O is a Unix-only optimization; Windows cannot ``os.open()`` a directory at all, and
46# its ``os`` module skips dir_fd support entirely. When disabled, callers fall back to full-path ops.
47_SUPPORTS_DIR_FD: Final[bool] = sys.platform != "win32" and os.open in os.supports_dir_fd
49_ALL_INSTANCES: Final[WeakValueDictionary[int, SoftReadWriteLock]] = WeakValueDictionary()
50_ALL_INSTANCES_LOCK: threading.Lock = threading.Lock()
51_SINGLETONS_UNDER_CONSTRUCTION: Final[set[Path]] = set()
54class _SoftRWMeta(type):
55 _instances: WeakValueDictionary[Path, SoftReadWriteLock]
56 _instances_lock: threading.RLock
58 def __call__( # ruff:ignore[too-many-arguments] # forwards the public constructor's documented parameters
59 cls,
60 lock_file: str | os.PathLike[str],
61 timeout: float = -1,
62 *,
63 blocking: bool = True,
64 is_singleton: bool = True,
65 heartbeat_interval: float = 30.0,
66 stale_threshold: float | None = None,
67 poll_interval: float = 0.25,
68 ) -> SoftReadWriteLock:
69 _ensure_current_process()
70 if not is_singleton:
71 return super().__call__(
72 lock_file,
73 timeout,
74 blocking=blocking,
75 is_singleton=is_singleton,
76 heartbeat_interval=heartbeat_interval,
77 stale_threshold=stale_threshold,
78 poll_interval=poll_interval,
79 )
81 normalized = Path(lock_file).resolve()
82 with cls._instances_lock:
83 instance = cls._instances.get(normalized)
84 if instance is None:
85 if normalized in _SINGLETONS_UNDER_CONSTRUCTION: # pragma: needs fork
86 msg = f"Singleton lock construction is already active for {lock_file!s}"
87 raise RuntimeError(msg)
88 construction_pid = os.getpid()
89 _SINGLETONS_UNDER_CONSTRUCTION.add(normalized)
90 try:
91 instance = super().__call__(
92 lock_file,
93 timeout,
94 blocking=blocking,
95 is_singleton=is_singleton,
96 heartbeat_interval=heartbeat_interval,
97 stale_threshold=stale_threshold,
98 poll_interval=poll_interval,
99 )
100 finally:
101 _SINGLETONS_UNDER_CONSTRUCTION.discard(normalized)
102 if os.getpid() != construction_pid: # pragma: needs fork
103 msg = "Lock construction cannot continue after fork; construct a new lock in the child"
104 raise RuntimeError(msg)
105 cls._instances[normalized] = instance
106 elif instance.timeout != timeout or instance.blocking != blocking:
107 msg = (
108 f"Singleton lock created with timeout={instance.timeout}, blocking={instance.blocking},"
109 f" cannot be changed to timeout={timeout}, blocking={blocking}"
110 )
111 raise ValueError(msg)
112 return instance
115class SoftReadWriteLock(metaclass=_SoftRWMeta):
116 """
117 Cross-process and cross-host reader/writer lock built on :class:`SoftFileLock` primitives.
119 Use this class instead of :class:`~filelock.ReadWriteLock` when the lock file lives on a network
120 filesystem (NFS, Lustre with ``-o flock``, HPC cluster shared storage). ``ReadWriteLock`` is backed
121 by SQLite and cannot run on NFS because SQLite's ``fcntl`` locking is unreliable there.
123 Layout on disk for a lock at ``foo.lock``:
125 - ``foo.lock.state`` — a :class:`SoftFileLock` taken only during state transitions (microseconds).
126 - ``foo.lock.write`` — writer marker; its presence means a writer is claiming or holding the lock.
127 - ``foo.lock.readers/<host>.<pid>.<uuid>`` — one file per reader.
129 Each marker stores a random token (``secrets.token_hex(16)``), the holder's pid, and the holder's
130 hostname. A daemon heartbeat thread refreshes ``mtime`` on every held marker. A marker whose mtime
131 has not advanced in ``stale_threshold`` seconds may be evicted by any process on any host, giving
132 correct behavior when a compute node crashes with a lock held.
134 Writer acquire is two-phase and writer-preferring: phase 1 claims ``.write`` (blocking any new
135 reader), phase 2 waits for existing readers to drain. Writer starvation is impossible.
137 Reentrancy, upgrade/downgrade rules, thread pinning, and singleton caching by resolved path match
138 :class:`~filelock.ReadWriteLock`.
140 Forking invalidates the inherited instance in the child so the child cannot double-own the lock with its parent;
141 ``release()`` on that instance is a no-op, and the child must construct a new instance if it needs a lock.
143 Trust boundary: protects against same-UID non-cooperating processes (one host or cross-host) and
144 same-host different-UID users via ``0o600`` / ``0o700`` permissions. Does not protect against root
145 compromise, NTP tampering on same-UID cross-host nodes, or multi-tenant mounts where hostile
146 co-tenants share the UID.
148 :param lock_file: path to the lock file; sidecar state/write/readers live next to it
149 :param timeout: maximum wait time in seconds; ``-1`` means block indefinitely
150 :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately on contention
151 :param is_singleton: if ``True``, reuse existing instances for the same resolved path
152 :param heartbeat_interval: seconds between heartbeat refreshes; default 30 s
153 :param stale_threshold: seconds of ``mtime`` inactivity before a marker is stale; defaults to
154 ``3 * heartbeat_interval``, matching etcd's ``LeaseKeepAlive`` convention
155 :param poll_interval: seconds between acquire retries under contention; default 0.25 s
157 .. versionadded:: 3.27.0
159 """
161 _instances: WeakValueDictionary[Path, SoftReadWriteLock] = WeakValueDictionary()
162 _instances_lock = threading.RLock()
164 def __init__( # ruff:ignore[too-many-arguments] # public constructor: one parameter per documented lock option
165 self,
166 lock_file: str | os.PathLike[str],
167 timeout: float = -1,
168 *,
169 blocking: bool = True,
170 is_singleton: bool = True, # ruff:ignore[unused-method-argument] # consumed by _SoftRWMeta.__call__
171 heartbeat_interval: float = 30.0,
172 stale_threshold: float | None = None,
173 poll_interval: float = 0.25,
174 ) -> None:
175 self._creator_pid = os.getpid()
176 if heartbeat_interval <= 0:
177 msg = f"heartbeat_interval must be positive, got {heartbeat_interval}"
178 raise ValueError(msg)
179 if stale_threshold is None:
180 stale_threshold = heartbeat_interval * 3
181 if stale_threshold <= heartbeat_interval:
182 msg = f"stale_threshold must exceed heartbeat_interval ({stale_threshold} <= {heartbeat_interval})"
183 raise ValueError(msg)
184 if poll_interval <= 0:
185 msg = f"poll_interval must be positive, got {poll_interval}"
186 raise ValueError(msg)
188 self.lock_file: str = os.fspath(lock_file)
189 self.timeout: float = timeout
190 self.blocking: bool = blocking
191 self.heartbeat_interval: float = heartbeat_interval
192 self.stale_threshold: float = stale_threshold
193 self.poll_interval: float = poll_interval
195 self._paths = _Paths(
196 state=f"{self.lock_file}.state",
197 write=f"{self.lock_file}.write",
198 readers=f"{self.lock_file}.readers",
199 )
200 ensure_directory_exists(self.lock_file)
201 self._locks = _Locks(
202 internal=threading.Lock(),
203 transaction=threading.Lock(),
204 state=SoftFileLock(self._paths.state, timeout=-1),
205 )
206 self._readers_dir_fd: int | None = None
207 self._readers_dir_fd_token: int | None = None
208 self._hold: _Hold | None = None
209 self._closed: bool = False
211 with _ALL_INSTANCES_LOCK:
212 _ALL_INSTANCES[id(self)] = self
213 _register_fork_object(self)
215 @classmethod
216 def _reset_class_after_fork(cls) -> None: # pragma: forked child
217 global _ALL_INSTANCES_LOCK # ruff:ignore[global-statement] # rebinds the module lock to a fresh one in the fork child
218 _ALL_INSTANCES_LOCK = threading.Lock()
219 cls._instances = WeakValueDictionary()
220 cls._instances_lock = threading.RLock()
221 _SINGLETONS_UNDER_CONSTRUCTION.clear()
223 @contextmanager
224 def read_lock(self, timeout: float | None = None, *, blocking: bool | None = None) -> Generator[None]:
225 """
226 Context manager that acquires and releases a shared read lock.
228 Falls back to instance defaults for *timeout* and *blocking* when ``None``.
230 :param timeout: maximum wait time in seconds, or ``None`` to use the instance default
231 :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately; ``None`` uses the instance default
233 :raises RuntimeError: if a write lock is already held on this instance
234 :raises Timeout: if the lock cannot be acquired within *timeout* seconds
236 """
237 self.acquire_read(timeout, blocking=blocking)
238 try:
239 yield
240 finally:
241 self.release()
243 @contextmanager
244 def write_lock(self, timeout: float | None = None, *, blocking: bool | None = None) -> Generator[None]:
245 """
246 Context manager that acquires and releases an exclusive write lock.
248 Falls back to instance defaults for *timeout* and *blocking* when ``None``.
250 :param timeout: maximum wait time in seconds, or ``None`` to use the instance default
251 :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately; ``None`` uses the instance default
253 :raises RuntimeError: if a read lock is already held, or a write lock is held by a different thread
254 :raises Timeout: if the lock cannot be acquired within *timeout* seconds
256 """
257 self.acquire_write(timeout, blocking=blocking)
258 try:
259 yield
260 finally:
261 self.release()
263 def acquire_read(self, timeout: float | None = None, *, blocking: bool | None = None) -> AcquireReturnProxy:
264 """
265 Acquire a shared read lock.
267 If this instance already holds a read lock, the lock level is incremented (reentrant). Attempting to acquire a
268 read lock while holding a write lock raises :class:`RuntimeError` (downgrade not allowed). On the 0→1
269 transition a daemon heartbeat thread is started that refreshes the reader marker's ``mtime`` every
270 ``heartbeat_interval`` seconds so peers on other hosts do not evict the marker as stale.
272 :param timeout: maximum wait time in seconds, or ``None`` to use the instance default; ``-1`` means block
273 indefinitely
274 :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately when the lock is unavailable;
275 ``None`` uses the instance default
277 :returns: a proxy that can be used as a context manager to release the lock
279 :raises RuntimeError: if a write lock is already held on this instance, if this instance was invalidated by
280 :func:`os.fork`, or if :meth:`close` was called
281 :raises Timeout: if the lock cannot be acquired within *timeout* seconds
283 """
284 return self._acquire("read", timeout, blocking=blocking)
286 def acquire_write(self, timeout: float | None = None, *, blocking: bool | None = None) -> AcquireReturnProxy:
287 """
288 Acquire an exclusive write lock.
290 If this instance already holds a write lock from the same thread, the lock level is incremented (reentrant).
291 Attempting to acquire a write lock while holding a read lock raises :class:`RuntimeError` (upgrade not
292 allowed). Write locks are pinned to the acquiring thread: a different thread trying to re-enter also raises
293 :class:`RuntimeError`.
295 Writer acquisition runs in two phases. Phase 1 atomically claims ``<path>.write`` via ``O_CREAT | O_EXCL``,
296 which immediately blocks any new reader on any host. Phase 2 waits for existing readers to drain. Writer
297 starvation is impossible: new readers see ``<path>.write`` during phase 2 and wait behind the pending writer.
299 :param timeout: maximum wait time in seconds, or ``None`` to use the instance default; ``-1`` means block
300 indefinitely
301 :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately when the lock is unavailable;
302 ``None`` uses the instance default
304 :returns: a proxy that can be used as a context manager to release the lock
306 :raises RuntimeError: if a read lock is already held, if a write lock is held by a different thread, if this
307 instance was invalidated by :func:`os.fork`, or if :meth:`close` was called
308 :raises Timeout: if the lock cannot be acquired within *timeout* seconds
310 """
311 return self._acquire("write", timeout, blocking=blocking)
313 @classmethod
314 def get_lock(
315 cls,
316 lock_file: str | os.PathLike[str],
317 timeout: float = -1,
318 *,
319 blocking: bool = True,
320 ) -> SoftReadWriteLock:
321 """
322 Return the singleton :class:`SoftReadWriteLock` for *lock_file*.
324 :param lock_file: path to the lock file; sidecar state/write/readers live next to it
325 :param timeout: maximum wait time in seconds; ``-1`` means block indefinitely
326 :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately when the lock is unavailable
328 :returns: the singleton lock instance
330 :raises ValueError: if an instance already exists for this path with different *timeout* or *blocking* values
332 """
333 return cls(lock_file, timeout, blocking=blocking)
335 def close(self) -> None:
336 """
337 Release any held lock and release internal filesystem resources.
339 Idempotent. After calling this method the instance can no longer acquire locks — subsequent acquires raise
340 :class:`RuntimeError`. A fork-invalidated instance is closed without raising.
341 """
342 if self._creator_pid != os.getpid(): # pragma: forked child
343 return
344 self.release(force=True)
345 with self._locks.internal:
346 if self._closed:
347 return
348 self._closed = True
349 if self._readers_dir_fd is not None: # pragma: needs dir-fd
350 with _fork_transition():
351 if self._readers_dir_fd_token is not None: # pragma: needs dir-fd
352 _unregister_owned_descriptor(self._readers_dir_fd_token)
353 self._readers_dir_fd_token = None
354 fd, self._readers_dir_fd = self._readers_dir_fd, None
355 with suppress(OSError): # pragma: needs dir-fd
356 os.close(fd)
358 def release(self, *, force: bool = False) -> None:
359 """
360 Release one level of the current lock.
362 When the lock level reaches zero the heartbeat thread is stopped and the held marker file is unlinked. On a
363 fork-invalidated instance (that is, the child of a :func:`os.fork` call made while the parent held a lock)
364 this method is a no-op so inherited ``with`` blocks can unwind cleanly in the child.
366 :param force: if ``True``, release the lock completely regardless of the current lock level
368 :raises RuntimeError: if no lock is currently held and *force* is ``False``
370 """
371 if self._creator_pid != os.getpid(): # pragma: forked child
372 return
373 with self._locks.internal:
374 hold = self._hold
375 if hold is None:
376 if force:
377 return
378 msg = f"Cannot release a lock on {self.lock_file} (lock id: {id(self)}) that is not held"
379 raise RuntimeError(msg)
380 if force:
381 hold.level = 0
382 else:
383 hold.level -= 1
384 if hold.level > 0:
385 return
386 self._hold = None
388 # Order matters: signal → join → unlink. A late tick on a deleted marker is harmless and the
389 # heartbeat's token check would catch a re-acquisition race, but joining first removes that race.
390 hold.heartbeat_stop.set()
391 hold.heartbeat_thread.join(timeout=self.heartbeat_interval + 1.0)
392 if hold.is_reader:
393 _unlink(hold.marker_name, dir_fd=self._readers_dir_fd)
394 else:
395 self._unlink_writer_marker_if_ours(hold.token)
397 def _unlink_writer_marker_if_ours(self, token: str) -> None:
398 # Remove the writer marker only while it still carries our token. If this holder was paused long
399 # enough (a stop-the-world GC pause, SIGSTOP, a suspended VM) for a peer to evict the marker as
400 # stale and claim the writer slot itself, the file now at <path>.write is the peer's live marker;
401 # unlinking it by path would let a second writer through and break mutual exclusion. The state lock
402 # serializes this against a concurrent break/claim, and the heartbeat is already stopped, so the
403 # token we read is authoritative. Mirrors the token re-check the stale-break path already does.
404 with self._locks.state:
405 if (read := _read_marker(self._paths.write)) is None:
406 return
407 info, _ = read
408 if info is None or not hmac.compare_digest(info.token, token):
409 return
410 _unlink(self._paths.write)
412 def _acquire(
413 self,
414 mode: _Mode,
415 timeout: float | None,
416 *,
417 blocking: bool | None,
418 ) -> AcquireReturnProxy:
419 if self._creator_pid != os.getpid(): # pragma: forked child
420 msg = f"SoftReadWriteLock on {self.lock_file} was invalidated by fork(); construct a new instance"
421 raise RuntimeError(msg)
422 timeout = self.timeout if timeout is None else timeout
423 blocking = self.blocking if blocking is None else blocking
425 with self._locks.internal:
426 if self._closed:
427 msg = f"SoftReadWriteLock on {self.lock_file} has been closed"
428 raise RuntimeError(msg)
429 if self._hold is not None:
430 return self._validate_reentrant(mode)
432 start = time.perf_counter()
433 if not blocking:
434 acquired = self._locks.transaction.acquire(blocking=False)
435 elif timeout == -1:
436 acquired = self._locks.transaction.acquire(blocking=True)
437 else:
438 acquired = self._locks.transaction.acquire(blocking=True, timeout=timeout)
439 if not acquired:
440 raise Timeout(self.lock_file) from None
441 try:
442 return self._do_acquire_inner(mode, timeout, start, blocking=blocking)
443 finally:
444 self._locks.transaction.release()
446 def _do_acquire_inner(
447 self,
448 mode: _Mode,
449 effective_timeout: float,
450 start: float,
451 *,
452 blocking: bool,
453 ) -> AcquireReturnProxy:
454 with self._locks.internal:
455 if self._hold is not None:
456 return self._validate_reentrant(mode)
457 deadline = None if effective_timeout == -1 else start + effective_timeout
458 token = secrets.token_hex(16)
459 if mode == "write":
460 marker_name, is_reader = self._acquire_writer_slot(token, deadline=deadline, blocking=blocking)
461 else:
462 marker_name, is_reader = self._acquire_reader_slot(token, deadline=deadline, blocking=blocking)
463 stop_event = threading.Event()
464 heartbeat = _HeartbeatThread(
465 refresh=self._refresh_marker,
466 interval=self.heartbeat_interval,
467 stop_event=stop_event,
468 name=f"filelock-heartbeat-{id(self):x}",
469 )
470 # Publish the hold and start its heartbeat under one internal-lock section, so a concurrent release() never
471 # observes a hold whose thread has not started and joins it. If the OS refuses the thread, clear the hold and
472 # unlink the marker we claimed: left in place, a peer evicts it as stale and acquires while this instance still
473 # believes it holds the lock.
474 start_error: BaseException | None = None
475 with self._locks.internal:
476 self._hold = _Hold(
477 level=1,
478 mode=mode,
479 write_thread_id=threading.get_ident() if mode == "write" else None,
480 marker_name=marker_name,
481 is_reader=is_reader,
482 token=token,
483 heartbeat_thread=heartbeat,
484 heartbeat_stop=stop_event,
485 )
486 try:
487 heartbeat.start()
488 except BaseException as error: # ruff:ignore[blind-except] # clear the slot below and re-raise
489 self._hold = None
490 start_error = error
491 if start_error is not None:
492 if is_reader:
493 _unlink(marker_name, dir_fd=self._readers_dir_fd)
494 else:
495 self._unlink_writer_marker_if_ours(token)
496 raise start_error
497 return AcquireReturnProxy(lock=self)
499 def _validate_reentrant(self, mode: _Mode) -> AcquireReturnProxy:
500 hold = self._hold
501 assert hold is not None # ruff:ignore[assert] # callers dispatch here only inside the self._hold is not None branch
502 if hold.mode != mode:
503 opposite = "write" if mode == "read" else "read"
504 direction = "downgrade" if mode == "read" else "upgrade"
505 msg = (
506 f"Cannot acquire {mode} lock on {self.lock_file} (lock id: {id(self)}): "
507 f"already holding a {opposite} lock ({direction} not allowed)"
508 )
509 raise RuntimeError(msg)
510 if mode == "write" and (cur := threading.get_ident()) != hold.write_thread_id:
511 msg = (
512 f"Cannot acquire write lock on {self.lock_file} (lock id: {id(self)}) "
513 f"from thread {cur} while it is held by thread {hold.write_thread_id}"
514 )
515 raise RuntimeError(msg)
516 hold.level += 1
517 return AcquireReturnProxy(lock=self)
519 def _acquire_writer_slot(
520 self,
521 token: str,
522 *,
523 deadline: float | None,
524 blocking: bool,
525 ) -> tuple[str, bool]:
526 # Phase 2 scans readers/ via dirfd (where supported), so we need it open even though writers never
527 # create files inside.
528 self._open_readers_dir()
530 def try_claim_writer() -> bool:
531 with self._locks.state:
532 return self._claim_writer_marker(token)
534 def readers_drained_touching() -> bool:
535 with self._locks.state:
536 # A peer may replace an expired marker while this process pauses. Refresh only our token; touching a
537 # successor's marker would let this acquisition proceed without owning the writer slot.
538 if not self._touch_writer_marker_if_ours(token) and not self._claim_writer_marker(token):
539 return False
540 self._break_stale_readers(time.time())
541 return not self._any_readers()
543 self._wait_for(try_claim_writer, deadline=deadline, blocking=blocking)
544 try:
545 self._wait_for(readers_drained_touching, deadline=deadline, blocking=blocking)
546 except Timeout:
547 # Give up our writer claim so readers can make progress again, but only while the marker is
548 # still ours: a peer may have evicted it as stale and claimed the slot while phase 2 waited.
549 self._unlink_writer_marker_if_ours(token)
550 raise
551 return self._paths.write, False
553 def _claim_writer_marker(self, token: str) -> bool:
554 # Claim the writer slot for ``token``. Must be called holding ``self._locks.state``. Evicts a
555 # stale marker first, then refuses to claim while a live ``.write`` exists so a peer holding the
556 # slot is waited out instead of overwritten.
557 _break_stale_marker(self._paths.write, stale_threshold=self.stale_threshold, now=time.time())
558 if _file_exists(self._paths.write):
559 return False
560 try:
561 _atomic_create_marker(self._paths.write, token)
562 except FileExistsError:
563 return False
564 return True
566 def _touch_writer_marker_if_ours(self, token: str) -> bool:
567 # Refresh the writer marker through a single O_NOFOLLOW fd, but only while it still carries our
568 # token. Returns False when the marker is gone or now belongs to a peer that reclaimed the slot,
569 # so the caller can re-claim rather than keep a stranger's marker alive. Mirrors _refresh_marker.
570 fd = _open_marker(self._paths.write)
571 if fd is None:
572 return False
573 try:
574 try:
575 data = os.read(fd, _MAX_MARKER_SIZE + 1)
576 except OSError: # pragma: no cover - e.g. EAGAIN from a hostile FIFO that has a writer attached
577 return False
578 info = _parse_marker_bytes(data)
579 if info is None or not hmac.compare_digest(info.token, token):
580 return False
581 with suppress(OSError):
582 touch(self._paths.write, fd=fd)
583 return True
584 finally:
585 os.close(fd)
587 def _acquire_reader_slot(
588 self,
589 token: str,
590 *,
591 deadline: float | None,
592 blocking: bool,
593 ) -> tuple[str, bool]:
594 self._open_readers_dir()
595 reader_name = f"{uuid.uuid4().hex}.{os.getpid()}"
596 dir_fd = self._readers_dir_fd
597 full_reader_path = str(Path(self._paths.readers) / reader_name)
599 def try_claim_reader() -> bool:
600 with self._locks.state:
601 _break_stale_marker(self._paths.write, stale_threshold=self.stale_threshold, now=time.time())
602 if _file_exists(self._paths.write):
603 return False
604 if dir_fd is not None: # pragma: needs dir-fd
605 _atomic_create_marker(reader_name, token, dir_fd=dir_fd)
606 else: # pragma: win32 cover
607 _atomic_create_marker(full_reader_path, token)
608 return True
610 self._wait_for(try_claim_reader, deadline=deadline, blocking=blocking)
611 return (reader_name if dir_fd is not None else full_reader_path), True
613 def _wait_for(
614 self,
615 predicate: Callable[[], bool],
616 *,
617 deadline: float | None,
618 blocking: bool,
619 ) -> None:
620 while True:
621 if predicate():
622 return
623 now = time.perf_counter()
624 if not blocking:
625 raise Timeout(self.lock_file)
626 if deadline is not None and now >= deadline:
627 raise Timeout(self.lock_file)
628 sleep_for = self.poll_interval
629 if deadline is not None:
630 sleep_for = min(sleep_for, max(deadline - now, 0.0))
631 time.sleep(sleep_for)
633 def _open_readers_dir(self) -> None:
634 readers_path = Path(self._paths.readers)
635 with suppress(FileExistsError):
636 readers_path.mkdir(mode=0o700)
637 # mkdir has no O_NOFOLLOW, so verify via lstat that we did not land on an attacker-placed symlink
638 # or a regular file before we open or scan inside.
639 st = os.lstat(self._paths.readers)
640 if stat.S_ISLNK(st.st_mode) or not stat.S_ISDIR(st.st_mode):
641 msg = f"{self._paths.readers} exists but is not a directory or is a symlink; refusing to use it"
642 raise RuntimeError(msg)
643 if self._readers_dir_fd is None and _SUPPORTS_DIR_FD: # pragma: needs dir-fd
644 with _fork_transition():
645 fd = os.open(self._paths.readers, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | _O_NOFOLLOW)
646 try:
647 token = _register_owned_descriptor(fd)
648 except BaseException as registration_error:
649 try:
650 os.close(fd)
651 except BaseException as close_error: # ruff:ignore[blind-except] # both errors surface via the group below
652 _raise_grouped_errors(
653 "reader directory registration and descriptor close both failed",
654 registration_error,
655 close_error,
656 )
657 raise
658 self._readers_dir_fd = fd
659 self._readers_dir_fd_token = token
661 def _any_readers(self) -> bool:
662 with closing(self._iter_reader_entries()) as entries:
663 for _ in entries:
664 return True
665 return False
667 def _iter_reader_entries(self) -> Generator[tuple[str, bool]]:
668 """
669 Yield ``(name, dirfd_relative)`` pairs for every live reader marker.
671 ``dirfd_relative`` is ``True`` when *name* should be passed to ``dir_fd=``-aware syscalls; ``False``
672 when *name* is a full path because dirfd-relative I/O is unavailable on this platform.
674 A consumer that stops early must close this generator: while suspended it holds the ``scandir`` handle open,
675 and leaving that to the collector surfaces as an unraisable exception inside whatever runs next.
676 """
677 if self._readers_dir_fd is not None: # pragma: needs dir-fd
678 with os.scandir(self._readers_dir_fd) as it:
679 for entry in it:
680 if not _is_housekeeping_name(entry.name):
681 yield entry.name, True
682 return
683 readers_path = Path(self._paths.readers) # pragma: win32 cover
684 with os.scandir(readers_path) as it: # pragma: win32 cover
685 for entry in it: # pragma: win32 cover
686 if not _is_housekeeping_name(entry.name): # pragma: win32 cover
687 yield str(readers_path / entry.name), False # pragma: win32 cover
689 def _break_stale_readers(self, now: float) -> None:
690 names: list[tuple[str, int | None]] = []
691 try:
692 with closing(self._iter_reader_entries()) as entries:
693 for name, dirfd_relative in entries:
694 names.append((name, self._readers_dir_fd if dirfd_relative else None))
695 except OSError: # pragma: no cover - transient NFS scandir hiccup
696 return
697 for name, fd in names:
698 _break_stale_marker(name, stale_threshold=self.stale_threshold, now=now, dir_fd=fd)
700 def _refresh_marker(self) -> bool:
701 with self._locks.internal:
702 hold = self._hold
703 if hold is None: # pragma: no cover - race between stop_event.set and join
704 return False
705 marker_name = hold.marker_name
706 token = hold.token
707 dir_fd = self._readers_dir_fd if hold.is_reader else None
709 # Open once with O_NOFOLLOW and touch that exact descriptor. Refreshing through the verified fd
710 # (instead of re-opening by name) closes the window where a peer unlinks our marker and drops a symlink
711 # or a different file at the path between the read and the touch: utime then lands on the inode we
712 # verified, or nowhere. Only an unambiguous loss stops the heartbeat: the marker gone, or a peer's token
713 # in its place. A transient filesystem error (ESTALE / EIO on the NFS-style filesystems this lock targets)
714 # keeps the heartbeat alive to retry next tick, the way the touch below already does, so one blip does not
715 # silently drop a held lock.
716 try:
717 fd = _open_marker_fd(marker_name, dir_fd=dir_fd)
718 except FileNotFoundError:
719 return False
720 except OSError:
721 return True
722 try:
723 try:
724 data = _read_marker_fd(fd)
725 except OSError: # a transient read error or EAGAIN from a hostile FIFO; retry rather than drop the lock
726 return True
727 info = _parse_marker_bytes(data)
728 # Token mismatch means another process already evicted our marker and created its own; stop the
729 # thread so it does not keep a stranger's file alive.
730 if info is None or not hmac.compare_digest(info.token, token):
731 return False
732 # A transient touch failure (ESTALE / EIO on the NFS-style filesystems this lock targets) must not
733 # kill the heartbeat thread: the read above just confirmed the marker is still ours, so swallow the
734 # error and retry on the next tick rather than letting the lease lapse while we still hold the lock.
735 with suppress(OSError):
736 touch(marker_name, fd=fd)
737 return True
738 finally:
739 os.close(fd)
741 def _reset_after_fork_in_child(self) -> None: # pragma: forked child
742 self._locks = _Locks(
743 internal=threading.Lock(),
744 transaction=threading.Lock(),
745 state=self._locks.state,
746 )
747 self._hold = None
748 self._readers_dir_fd = None
749 self._readers_dir_fd_token = None
752class _HeartbeatThread(threading.Thread):
753 def __init__(
754 self,
755 refresh: Callable[[], bool],
756 interval: float,
757 stop_event: threading.Event,
758 name: str,
759 ) -> None:
760 super().__init__(name=name, daemon=True)
761 self._refresh = refresh
762 self._interval = interval
763 self._stop_event = stop_event
765 def run(self) -> None:
766 while not self._stop_event.wait(self._interval):
767 if not self._refresh():
768 self._stop_event.set()
769 return
772def _read_marker(name: str, *, dir_fd: int | None = None) -> tuple[_MarkerInfo | None, float] | None:
773 fd = _open_marker(name, dir_fd=dir_fd)
774 if fd is None:
775 return None
776 try:
777 st = os.fstat(fd)
778 # A legitimate marker is a regular file, so anything else at the path (a FIFO, say) is reported as a
779 # malformed marker (its mtime still drives stale eviction) without being read. Reading is where
780 # platforms diverge: an empty non-blocking read yields 0 bytes on Linux/macOS but EAGAIN on FreeBSD,
781 # and the EAGAIN used to abort the stale-break and wedge the acquire until timeout (#587).
782 if not stat.S_ISREG(st.st_mode): # pragma: needs fifo
783 return None, st.st_mtime
784 data = os.read(fd, _MAX_MARKER_SIZE + 1)
785 except OSError: # pragma: no cover - marker vanished or turned unreadable between open and read
786 return None
787 finally:
788 os.close(fd)
789 return _parse_marker_bytes(data), st.st_mtime
792def _read_marker_fd(fd: int) -> bytes:
793 return os.read(fd, _MAX_MARKER_SIZE + 1)
796def _open_marker_fd(name: str, *, dir_fd: int | None = None) -> int:
797 # The file is ours; these guard a hostile mid-flight swap. O_NOFOLLOW rejects a symlink; O_NONBLOCK keeps
798 # a real FIFO from blocking the open forever, so it reads as a malformed marker instead of wedging a peer
799 # that holds the state lock.
800 flags = os.O_RDONLY | _O_NOFOLLOW | _O_NONBLOCK
801 return os.open(name, flags, dir_fd=dir_fd) if _SUPPORTS_DIR_FD and dir_fd is not None else os.open(name, flags)
804def _open_marker(name: str, *, dir_fd: int | None = None) -> int | None:
805 try:
806 return _open_marker_fd(name, dir_fd=dir_fd)
807 except OSError:
808 return None
811def _parse_marker_bytes(data: bytes) -> _MarkerInfo | None:
812 # Trust nothing about attacker-controlled markers; any deviation returns None so callers fall through
813 # to stale cleanup. ``re.match`` caches compiled patterns internally, so the regex is built only once
814 # despite being defined inline.
815 if not data or len(data) > _MAX_MARKER_SIZE:
816 return None
817 try:
818 text = data.decode("ascii")
819 except UnicodeDecodeError:
820 return None
821 match = re.match(
822 r"""
823 \A # start of string
824 (?P<token> [0-9a-f]{32} ) \n # 128-bit hex token
825 (?P<pid> [1-9][0-9]{0,9} ) \n # decimal pid: no leading zero, ≤ 10 digits
826 (?P<hostname> [\x21-\x7e]{1,253}) # printable non-whitespace ASCII (RFC 1123 hostname limit)
827 \n* # tolerate sloppy writers that append extra newlines
828 \Z # end of string
829 """,
830 text,
831 re.VERBOSE,
832 )
833 if match is None:
834 return None
835 pid = int(match["pid"], 10)
836 if pid > 2**31 - 1:
837 return None
838 return _MarkerInfo(token=match["token"], pid=pid, hostname=match["hostname"])
841def _unlink(name: str, *, dir_fd: int | None = None) -> None:
842 with suppress(FileNotFoundError):
843 if _SUPPORTS_DIR_FD and dir_fd is not None: # pragma: needs dir-fd
844 # Path.unlink has no dir_fd support, so we stay on os.unlink for the dirfd path.
845 os.unlink(name, dir_fd=dir_fd)
846 else:
847 Path(name).unlink()
850def _break_stale_marker( # ruff:ignore[too-many-return-statements] # each return is a distinct abort/commit point in the break protocol
851 name: str,
852 *,
853 stale_threshold: float,
854 now: float,
855 dir_fd: int | None = None,
856) -> bool:
857 # Atomic break pattern: read → rename to unique break-name → re-verify → unlink. The rename gives us a
858 # private name nobody else can touch; if the re-verify sees a newer mtime or a different token, the
859 # legitimate holder's heartbeat fired between read and rename and we must abort (leaving the .break.*
860 # file behind rather than rollback-renaming, because rollback is itself racy).
861 if (read_result := _read_marker(name, dir_fd=dir_fd)) is None:
862 return False
863 info_before, mtime_before = read_result
864 if now - mtime_before <= stale_threshold:
865 return False
866 if info_before is None:
867 _unlink(name, dir_fd=dir_fd)
868 return True
870 break_name = f"{name}{_BREAK_SUFFIX}.{os.getpid()}.{secrets.token_hex(16)}"
871 try:
872 if _SUPPORTS_DIR_FD and dir_fd is not None: # pragma: needs dir-fd
873 os.rename(name, break_name, src_dir_fd=dir_fd, dst_dir_fd=dir_fd)
874 else:
875 Path(name).rename(break_name)
876 except OSError: # pragma: no cover - race where the marker vanishes between read and rename
877 return False
879 read_after = _read_marker(break_name, dir_fd=dir_fd)
880 if read_after is None: # pragma: no cover - race where a peer unlinks the break-name file
881 return False
882 info_after, mtime_after = read_after
883 if info_after is None: # pragma: no cover - content replaced post-rename by a racing peer
884 _unlink(break_name, dir_fd=dir_fd)
885 return True
886 if not hmac.compare_digest(info_before.token, info_after.token): # pragma: no cover - race only
887 return False
888 if mtime_after > mtime_before: # pragma: no cover - heartbeat raced our rename
889 return False
890 _unlink(break_name, dir_fd=dir_fd)
891 return True
894def _atomic_create_marker(name: str, token: str, *, dir_fd: int | None = None) -> None:
895 # O_NOFOLLOW blocks the symlink-overwrite attack where an attacker pre-creates the marker path as a
896 # symlink pointing at a victim file. Mode 0o600 keeps the token unreadable to other users.
897 flags = os.O_CREAT | os.O_EXCL | os.O_WRONLY | _O_NOFOLLOW
898 if _SUPPORTS_DIR_FD and dir_fd is not None: # pragma: needs dir-fd
899 fd = os.open(name, flags, 0o600, dir_fd=dir_fd)
900 else:
901 fd = os.open(name, flags, 0o600)
902 # Write the whole record before the marker counts as created. On failure remove it only while the path still names
903 # the file we opened, so a rollback never deletes a marker a concurrent reader recreated at this name.
904 identity: tuple[int, int] | None = None
905 try:
906 st = os.fstat(fd)
907 identity = st.st_dev, st.st_ino
908 write_all(fd, f"{token}\n{os.getpid()}\n{socket.gethostname()}\n".encode("ascii"))
909 except BaseException:
910 os.close(fd)
911 if identity is not None and _same_file(name, identity, dir_fd=dir_fd):
912 _unlink(name, dir_fd=dir_fd)
913 raise
914 else:
915 os.close(fd)
918def _same_file(name: str, identity: tuple[int, int], *, dir_fd: int | None) -> bool:
919 try:
920 st = os.lstat(name, dir_fd=dir_fd) if _SUPPORTS_DIR_FD and dir_fd is not None else os.lstat(name)
921 except OSError:
922 return False
923 return (st.st_dev, st.st_ino) == identity
926def _file_exists(path: str) -> bool:
927 try:
928 st = os.lstat(path)
929 except FileNotFoundError:
930 return False
931 return stat.S_ISREG(st.st_mode)
934def _is_housekeeping_name(name: str) -> bool:
935 return name.startswith(".") or _BREAK_SUFFIX in name
938@dataclass(frozen=True)
939class _Paths:
940 state: str
941 write: str
942 readers: str
945@dataclass
946class _Locks:
947 internal: threading.Lock
948 transaction: threading.Lock
949 state: SoftFileLock
952@dataclass(frozen=True)
953class _MarkerInfo:
954 token: str
955 pid: int
956 hostname: str
959@dataclass
960class _Hold:
961 """Everything that exists only while a lock is held; ``None`` when the instance has no lock."""
963 level: int
964 mode: _Mode
965 write_thread_id: int | None
966 marker_name: str
967 is_reader: bool
968 token: str
969 heartbeat_thread: _HeartbeatThread
970 heartbeat_stop: threading.Event
973def _cleanup_all_instances() -> None: # pragma: no cover - runs from atexit at interpreter shutdown
974 for instance in list(_ALL_INSTANCES.values()):
975 with suppress(Exception):
976 instance.release(force=True)
979atexit.register(_cleanup_all_instances)
980_register_fork_class(SoftReadWriteLock)
983__all__ = [
984 "SoftReadWriteLock",
985]