Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/filelock/_soft_rw/_sync.py: 22%

1"""Cross-process and cross-host reader/writer lock built on :class:`SoftFileLock` primitives.""" 

2 

3from __future__ import annotations 

4 

5import atexit 

6import hmac 

7import os 

8import re 

9import secrets 

10import stat 

11import sys 

12import threading 

13import time 

14import uuid 

15from contextlib import closing, contextmanager, suppress 

16from dataclasses import dataclass 

17from math import isfinite 

18from pathlib import Path 

19from typing import TYPE_CHECKING, Final, Literal 

20from weakref import WeakValueDictionary 

21 

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._identity import host_name 

34from filelock._soft import SoftFileLock 

35from filelock._util import ensure_directory_exists, touch, write_all 

36 

37if TYPE_CHECKING: 

38 from collections.abc import Callable, Generator 

39 

40 

41_Mode = Literal["read", "write"] 

42_BREAK_SUFFIX: Final[str] = ".break" 

43_MAX_MARKER_SIZE: Final[int] = 1024 

44_O_NOFOLLOW: Final[int] = getattr(os, "O_NOFOLLOW", 0) 

45_O_NONBLOCK: Final[int] = getattr(os, "O_NONBLOCK", 0) 

46# dirfd-relative I/O is a Unix-only optimization; Windows cannot ``os.open()`` a directory at all, and 

47# its ``os`` module skips dir_fd support entirely. When disabled, callers fall back to full-path ops. 

48_SUPPORTS_DIR_FD: Final[bool] = sys.platform != "win32" and os.open in os.supports_dir_fd 

49 

50_ALL_INSTANCES: Final[WeakValueDictionary[int, SoftReadWriteLock]] = WeakValueDictionary() 

51_ALL_INSTANCES_LOCK: threading.Lock = threading.Lock() 

52_SINGLETONS_UNDER_CONSTRUCTION: Final[set[Path]] = set() 

53 

54 

55class _SoftRWMeta(type): 

56 _instances: WeakValueDictionary[Path, SoftReadWriteLock] 

57 _instances_lock: threading.RLock 

58 

59 def __call__( # ruff:ignore[too-many-arguments] # forwards the public constructor's documented parameters 

60 cls, 

61 lock_file: str | os.PathLike[str], 

62 timeout: float = -1, 

63 *, 

64 blocking: bool = True, 

65 is_singleton: bool = True, 

66 heartbeat_interval: float = 30.0, 

67 stale_threshold: float | None = None, 

68 poll_interval: float = 0.25, 

69 ) -> SoftReadWriteLock: 

70 _ensure_current_process() 

71 if not is_singleton: 

72 return super().__call__( 

73 lock_file, 

74 timeout, 

75 blocking=blocking, 

76 is_singleton=is_singleton, 

77 heartbeat_interval=heartbeat_interval, 

78 stale_threshold=stale_threshold, 

79 poll_interval=poll_interval, 

80 ) 

81 

82 normalized = Path(lock_file).resolve() 

83 with cls._instances_lock: 

84 instance = cls._instances.get(normalized) 

85 if instance is None: 

86 if normalized in _SINGLETONS_UNDER_CONSTRUCTION: # pragma: needs fork 

87 msg = f"Singleton lock construction is already active for {lock_file!s}" 

88 raise RuntimeError(msg) 

89 construction_pid = os.getpid() 

90 _SINGLETONS_UNDER_CONSTRUCTION.add(normalized) 

91 try: 

92 instance = super().__call__( 

93 lock_file, 

94 timeout, 

95 blocking=blocking, 

96 is_singleton=is_singleton, 

97 heartbeat_interval=heartbeat_interval, 

98 stale_threshold=stale_threshold, 

99 poll_interval=poll_interval, 

100 ) 

101 finally: 

102 _SINGLETONS_UNDER_CONSTRUCTION.discard(normalized) 

103 if os.getpid() != construction_pid: # pragma: needs fork 

104 msg = "Lock construction cannot continue after fork; construct a new lock in the child" 

105 raise RuntimeError(msg) 

106 cls._instances[normalized] = instance 

107 elif instance.timeout != timeout or instance.blocking != blocking: 

108 msg = ( 

109 f"Singleton lock created with timeout={instance.timeout}, blocking={instance.blocking}," 

110 f" cannot be changed to timeout={timeout}, blocking={blocking}" 

111 ) 

112 raise ValueError(msg) 

113 else: 

114 _validate_intervals(heartbeat_interval, stale_threshold, poll_interval) 

115 return instance 

116 

117 

118class SoftReadWriteLock(metaclass=_SoftRWMeta): 

119 """ 

120 Cross-process and cross-host reader/writer lock built on :class:`SoftFileLock` primitives. 

121 

122 Use this class instead of :class:`~filelock.ReadWriteLock` when the lock file lives on a network 

123 filesystem (NFS, Lustre with ``-o flock``, HPC cluster shared storage). ``ReadWriteLock`` is backed 

124 by SQLite and cannot run on NFS because SQLite's ``fcntl`` locking is unreliable there. 

125 

126 Layout on disk for a lock at ``foo.lock``: 

127 

128 - ``foo.lock.state`` — a :class:`SoftFileLock` taken only during state transitions (microseconds). 

129 - ``foo.lock.write`` — writer marker; its presence means a writer is claiming or holding the lock. 

130 - ``foo.lock.readers/<host>.<pid>.<uuid>`` — one file per reader. 

131 

132 Each marker stores a random token (``secrets.token_hex(16)``), the holder's pid, and the holder's 

133 hostname. A daemon heartbeat thread refreshes ``mtime`` on every held marker. A marker whose mtime 

134 has not advanced in ``stale_threshold`` seconds may be evicted by any process on any host, giving 

135 correct behavior when a compute node crashes with a lock held. 

136 

137 Writer acquire is two-phase and writer-preferring: phase 1 claims ``.write`` (blocking any new 

138 reader), phase 2 waits for existing readers to drain. Writer starvation is impossible. 

139 

140 Reentrancy, upgrade/downgrade rules, thread pinning, and singleton caching by resolved path match 

141 :class:`~filelock.ReadWriteLock`. 

142 

143 Forking invalidates the inherited instance in the child so the child cannot double-own the lock with its parent; 

144 ``release()`` on that instance is a no-op, and the child must construct a new instance if it needs a lock. 

145 

146 Trust boundary: protects against same-UID non-cooperating processes (one host or cross-host) and 

147 same-host different-UID users via ``0o600`` / ``0o700`` permissions. Does not protect against root 

148 compromise, NTP tampering on same-UID cross-host nodes, or multi-tenant mounts where hostile 

149 co-tenants share the UID. 

150 

151 :param lock_file: path to the lock file; sidecar state/write/readers live next to it 

152 :param timeout: maximum wait time in seconds; ``-1`` means block indefinitely 

153 :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately on contention 

154 :param is_singleton: if ``True``, reuse existing instances for the same resolved path 

155 :param heartbeat_interval: seconds between heartbeat refreshes; default 30 s 

156 :param stale_threshold: seconds of ``mtime`` inactivity before a marker is stale; defaults to 

157 ``3 * heartbeat_interval``, matching etcd's ``LeaseKeepAlive`` convention 

158 :param poll_interval: seconds between acquire retries under contention; default 0.25 s 

159 

160 .. versionadded:: 3.27.0 

161 

162 """ 

163 

164 _instances: WeakValueDictionary[Path, SoftReadWriteLock] = WeakValueDictionary() 

165 _instances_lock = threading.RLock() 

166 

167 def __init__( # ruff:ignore[too-many-arguments] # public constructor: one parameter per documented lock option 

168 self, 

169 lock_file: str | os.PathLike[str], 

170 timeout: float = -1, 

171 *, 

172 blocking: bool = True, 

173 is_singleton: bool = True, # ruff:ignore[unused-method-argument] # consumed by _SoftRWMeta.__call__ 

174 heartbeat_interval: float = 30.0, 

175 stale_threshold: float | None = None, 

176 poll_interval: float = 0.25, 

177 ) -> None: 

178 self._creator_pid = os.getpid() 

179 stale_threshold = _validate_intervals(heartbeat_interval, stale_threshold, poll_interval) 

180 

181 self.lock_file: str = os.fspath(lock_file) 

182 self.timeout: float = timeout 

183 self.blocking: bool = blocking 

184 self.heartbeat_interval: float = heartbeat_interval 

185 self.stale_threshold: float = stale_threshold 

186 self.poll_interval: float = poll_interval 

187 

188 self._paths = _Paths( 

189 state=f"{self.lock_file}.state", 

190 write=f"{self.lock_file}.write", 

191 readers=f"{self.lock_file}.readers", 

192 ) 

193 ensure_directory_exists(self.lock_file) 

194 self._locks = _Locks( 

195 internal=threading.Lock(), 

196 transaction=threading.Lock(), 

197 state=SoftFileLock(self._paths.state, timeout=-1), 

198 ) 

199 self._readers_dir_fd: int | None = None 

200 self._readers_dir_fd_token: int | None = None 

201 self._hold: _Hold | None = None 

202 self._closed: bool = False 

203 

204 with _ALL_INSTANCES_LOCK: 

205 _ALL_INSTANCES[id(self)] = self 

206 _register_fork_object(self) 

207 

208 @classmethod 

209 def _reset_class_after_fork(cls) -> None: # pragma: forked child 

210 global _ALL_INSTANCES_LOCK # ruff:ignore[global-statement] # rebinds the module lock to a fresh one in the fork child 

211 _ALL_INSTANCES_LOCK = threading.Lock() 

212 cls._instances = WeakValueDictionary() 

213 cls._instances_lock = threading.RLock() 

214 _SINGLETONS_UNDER_CONSTRUCTION.clear() 

215 

216 @contextmanager 

217 def read_lock(self, timeout: float | None = None, *, blocking: bool | None = None) -> Generator[None]: 

218 """ 

219 Context manager that acquires and releases a shared read lock. 

220 

221 Falls back to instance defaults for *timeout* and *blocking* when ``None``. 

222 

223 :param timeout: maximum wait time in seconds, or ``None`` to use the instance default 

224 :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately; ``None`` uses the instance default 

225 

226 :raises RuntimeError: if a write lock is already held on this instance 

227 :raises Timeout: if the lock cannot be acquired within *timeout* seconds 

228 

229 """ 

230 self.acquire_read(timeout, blocking=blocking) 

231 try: 

232 yield 

233 finally: 

234 self.release() 

235 

236 @contextmanager 

237 def write_lock(self, timeout: float | None = None, *, blocking: bool | None = None) -> Generator[None]: 

238 """ 

239 Context manager that acquires and releases an exclusive write lock. 

240 

241 Falls back to instance defaults for *timeout* and *blocking* when ``None``. 

242 

243 :param timeout: maximum wait time in seconds, or ``None`` to use the instance default 

244 :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately; ``None`` uses the instance default 

245 

246 :raises RuntimeError: if a read lock is already held, or a write lock is held by a different thread 

247 :raises Timeout: if the lock cannot be acquired within *timeout* seconds 

248 

249 """ 

250 self.acquire_write(timeout, blocking=blocking) 

251 try: 

252 yield 

253 finally: 

254 self.release() 

255 

256 def acquire_read(self, timeout: float | None = None, *, blocking: bool | None = None) -> AcquireReturnProxy: 

257 """ 

258 Acquire a shared read lock. 

259 

260 If this instance already holds a read lock, the lock level is incremented (reentrant). Attempting to acquire a 

261 read lock while holding a write lock raises :class:`RuntimeError` (downgrade not allowed). On the 0→1 

262 transition a daemon heartbeat thread is started that refreshes the reader marker's ``mtime`` every 

263 ``heartbeat_interval`` seconds so peers on other hosts do not evict the marker as stale. 

264 

265 :param timeout: maximum wait time in seconds, or ``None`` to use the instance default; ``-1`` means block 

266 indefinitely 

267 :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately when the lock is unavailable; 

268 ``None`` uses the instance default 

269 

270 :returns: a proxy that can be used as a context manager to release the lock 

271 

272 :raises RuntimeError: if a write lock is already held on this instance, if this instance was invalidated by 

273 :func:`os.fork`, or if :meth:`close` was called 

274 :raises Timeout: if the lock cannot be acquired within *timeout* seconds 

275 

276 """ 

277 return self._acquire("read", timeout, blocking=blocking) 

278 

279 def acquire_write(self, timeout: float | None = None, *, blocking: bool | None = None) -> AcquireReturnProxy: 

280 """ 

281 Acquire an exclusive write lock. 

282 

283 If this instance already holds a write lock from the same thread, the lock level is incremented (reentrant). 

284 Attempting to acquire a write lock while holding a read lock raises :class:`RuntimeError` (upgrade not 

285 allowed). Write locks are pinned to the acquiring thread: a different thread trying to re-enter also raises 

286 :class:`RuntimeError`. 

287 

288 Writer acquisition runs in two phases. Phase 1 atomically claims ``<path>.write`` via ``O_CREAT | O_EXCL``, 

289 which immediately blocks any new reader on any host. Phase 2 waits for existing readers to drain. Writer 

290 starvation is impossible: new readers see ``<path>.write`` during phase 2 and wait behind the pending writer. 

291 

292 :param timeout: maximum wait time in seconds, or ``None`` to use the instance default; ``-1`` means block 

293 indefinitely 

294 :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately when the lock is unavailable; 

295 ``None`` uses the instance default 

296 

297 :returns: a proxy that can be used as a context manager to release the lock 

298 

299 :raises RuntimeError: if a read lock is already held, if a write lock is held by a different thread, if this 

300 instance was invalidated by :func:`os.fork`, or if :meth:`close` was called 

301 :raises Timeout: if the lock cannot be acquired within *timeout* seconds 

302 

303 """ 

304 return self._acquire("write", timeout, blocking=blocking) 

305 

306 @classmethod 

307 def get_lock( 

308 cls, 

309 lock_file: str | os.PathLike[str], 

310 timeout: float = -1, 

311 *, 

312 blocking: bool = True, 

313 ) -> SoftReadWriteLock: 

314 """ 

315 Return the singleton :class:`SoftReadWriteLock` for *lock_file*. 

316 

317 :param lock_file: path to the lock file; sidecar state/write/readers live next to it 

318 :param timeout: maximum wait time in seconds; ``-1`` means block indefinitely 

319 :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately when the lock is unavailable 

320 

321 :returns: the singleton lock instance 

322 

323 :raises ValueError: if an instance already exists for this path with different *timeout* or *blocking* values 

324 

325 """ 

326 return cls(lock_file, timeout, blocking=blocking) 

327 

328 def close(self) -> None: 

329 """ 

330 Release any held lock and release internal filesystem resources. 

331 

332 Idempotent. After calling this method the instance can no longer acquire locks — subsequent acquires raise 

333 :class:`RuntimeError`. A fork-invalidated instance is closed without raising. 

334 """ 

335 if self._creator_pid != os.getpid(): # pragma: forked child 

336 return 

337 self.release(force=True) 

338 with self._locks.internal: 

339 if self._closed: 

340 return 

341 self._closed = True 

342 if self._readers_dir_fd is not None: # pragma: needs dir-fd 

343 with _fork_transition(): 

344 if self._readers_dir_fd_token is not None: # pragma: needs dir-fd 

345 _unregister_owned_descriptor(self._readers_dir_fd_token) 

346 self._readers_dir_fd_token = None 

347 fd, self._readers_dir_fd = self._readers_dir_fd, None 

348 with suppress(OSError): # pragma: needs dir-fd 

349 os.close(fd) 

350 

351 def release(self, *, force: bool = False) -> None: 

352 """ 

353 Release one level of the current lock. 

354 

355 When the lock level reaches zero the heartbeat thread is stopped and the held marker file is unlinked. On a 

356 fork-invalidated instance (that is, the child of a :func:`os.fork` call made while the parent held a lock) 

357 this method is a no-op so inherited ``with`` blocks can unwind cleanly in the child. 

358 

359 :param force: if ``True``, release the lock completely regardless of the current lock level 

360 

361 :raises RuntimeError: if no lock is currently held and *force* is ``False`` 

362 

363 """ 

364 if self._creator_pid != os.getpid(): # pragma: forked child 

365 return 

366 with self._locks.internal: 

367 hold = self._hold 

368 if hold is None: 

369 if force: 

370 return 

371 msg = f"Cannot release a lock on {self.lock_file} (lock id: {id(self)}) that is not held" 

372 raise RuntimeError(msg) 

373 if force: 

374 hold.level = 0 

375 else: 

376 hold.level -= 1 

377 if hold.level > 0: 

378 return 

379 self._hold = None 

380 

381 # Order matters: signal → join → unlink. A late tick on a deleted marker is harmless and the 

382 # heartbeat's token check would catch a re-acquisition race, but joining first removes that race. 

383 hold.heartbeat_stop.set() 

384 hold.heartbeat_thread.join(timeout=self.heartbeat_interval + 1.0) 

385 if hold.is_reader: 

386 _unlink(hold.marker_name, dir_fd=self._readers_dir_fd) 

387 else: 

388 self._unlink_writer_marker_if_ours(hold.token) 

389 

390 def _unlink_writer_marker_if_ours(self, token: str, *, blocking: bool = True) -> None: 

391 # Serialize the token check and unlink so a paused holder cannot remove its successor's marker. 

392 try: 

393 with self._locks.state.acquire(blocking=blocking): 

394 if (read := _read_marker(self._paths.write)) is None: 

395 return 

396 info, _ = read 

397 if info is not None and hmac.compare_digest(info.token, token): 

398 _unlink(self._paths.write) 

399 except Timeout: 

400 # Failed acquisition must return on time; peers can reclaim the unrefreshed writer marker. 

401 return 

402 

403 def _acquire( 

404 self, 

405 mode: _Mode, 

406 timeout: float | None, 

407 *, 

408 blocking: bool | None, 

409 ) -> AcquireReturnProxy: 

410 if self._creator_pid != os.getpid(): # pragma: forked child 

411 msg = f"SoftReadWriteLock on {self.lock_file} was invalidated by fork(); construct a new instance" 

412 raise RuntimeError(msg) 

413 timeout = self.timeout if timeout is None else timeout 

414 blocking = self.blocking if blocking is None else blocking 

415 

416 with self._locks.internal: 

417 if self._closed: 

418 msg = f"SoftReadWriteLock on {self.lock_file} has been closed" 

419 raise RuntimeError(msg) 

420 if self._hold is not None: 

421 return self._validate_reentrant(mode) 

422 

423 start = time.perf_counter() 

424 if not blocking: 

425 acquired = self._locks.transaction.acquire(blocking=False) 

426 elif timeout == -1: 

427 acquired = self._locks.transaction.acquire(blocking=True) 

428 else: 

429 acquired = self._locks.transaction.acquire(blocking=True, timeout=timeout) 

430 if not acquired: 

431 raise Timeout(self.lock_file) from None 

432 try: 

433 return self._do_acquire_inner(mode, timeout, start, blocking=blocking) 

434 finally: 

435 self._locks.transaction.release() 

436 

437 def _do_acquire_inner( 

438 self, 

439 mode: _Mode, 

440 effective_timeout: float, 

441 start: float, 

442 *, 

443 blocking: bool, 

444 ) -> AcquireReturnProxy: 

445 with self._locks.internal: 

446 if self._hold is not None: 

447 return self._validate_reentrant(mode) 

448 deadline = None if effective_timeout == -1 else start + effective_timeout 

449 token = secrets.token_hex(16) 

450 if mode == "write": 

451 marker_name, is_reader = self._acquire_writer_slot(token, deadline=deadline, blocking=blocking) 

452 else: 

453 marker_name, is_reader = self._acquire_reader_slot(token, deadline=deadline, blocking=blocking) 

454 stop_event = threading.Event() 

455 heartbeat = _HeartbeatThread( 

456 refresh=self._refresh_marker, 

457 interval=self.heartbeat_interval, 

458 stop_event=stop_event, 

459 name=f"filelock-heartbeat-{id(self):x}", 

460 ) 

461 # Publish the hold and start its heartbeat under one internal-lock section, so a concurrent release() never 

462 # observes a hold whose thread has not started and joins it. If the OS refuses the thread, clear the hold and 

463 # unlink the marker we claimed: left in place, a peer evicts it as stale and acquires while this instance still 

464 # believes it holds the lock. 

465 start_error: BaseException | None = None 

466 with self._locks.internal: 

467 self._hold = _Hold( 

468 level=1, 

469 mode=mode, 

470 write_thread_id=threading.get_ident() if mode == "write" else None, 

471 marker_name=marker_name, 

472 is_reader=is_reader, 

473 token=token, 

474 heartbeat_thread=heartbeat, 

475 heartbeat_stop=stop_event, 

476 ) 

477 try: 

478 heartbeat.start() 

479 except BaseException as error: # ruff:ignore[blind-except] # clear the slot below and re-raise 

480 self._hold = None 

481 start_error = error 

482 if start_error is not None: 

483 if is_reader: 

484 _unlink(marker_name, dir_fd=self._readers_dir_fd) 

485 else: 

486 self._unlink_writer_marker_if_ours(token, blocking=False) 

487 raise start_error 

488 return AcquireReturnProxy(lock=self) 

489 

490 def _validate_reentrant(self, mode: _Mode) -> AcquireReturnProxy: 

491 hold = self._hold 

492 assert hold is not None # ruff:ignore[assert] # callers dispatch here only inside the self._hold is not None branch 

493 if hold.mode != mode: 

494 opposite = "write" if mode == "read" else "read" 

495 direction = "downgrade" if mode == "read" else "upgrade" 

496 msg = ( 

497 f"Cannot acquire {mode} lock on {self.lock_file} (lock id: {id(self)}): " 

498 f"already holding a {opposite} lock ({direction} not allowed)" 

499 ) 

500 raise RuntimeError(msg) 

501 if mode == "write" and (cur := threading.get_ident()) != hold.write_thread_id: 

502 msg = ( 

503 f"Cannot acquire write lock on {self.lock_file} (lock id: {id(self)}) " 

504 f"from thread {cur} while it is held by thread {hold.write_thread_id}" 

505 ) 

506 raise RuntimeError(msg) 

507 hold.level += 1 

508 return AcquireReturnProxy(lock=self) 

509 

510 def _acquire_writer_slot( 

511 self, 

512 token: str, 

513 *, 

514 deadline: float | None, 

515 blocking: bool, 

516 ) -> tuple[str, bool]: 

517 # Phase 2 scans readers/ via dirfd (where supported), so we need it open even though writers never 

518 # create files inside. 

519 self._open_readers_dir() 

520 

521 def try_claim_writer() -> bool: 

522 return self._claim_writer_marker(token) 

523 

524 def readers_drained_touching() -> bool: 

525 # A paused contender must reclaim its slot before proceeding if a peer replaced its marker. 

526 if not self._touch_writer_marker_if_ours(token) and not self._claim_writer_marker(token): 

527 return False 

528 self._break_stale_readers(time.time()) 

529 return not self._any_readers() 

530 

531 self._wait_for(try_claim_writer, deadline=deadline, blocking=blocking) 

532 try: 

533 self._wait_for(readers_drained_touching, deadline=deadline, blocking=blocking) 

534 except Timeout: 

535 self._unlink_writer_marker_if_ours(token, blocking=False) 

536 raise 

537 return self._paths.write, False 

538 

539 def _claim_writer_marker(self, token: str) -> bool: 

540 # Claim the writer slot for ``token``. Must be called holding ``self._locks.state``. Evicts a 

541 # stale marker first, then refuses to claim while a live ``.write`` exists so a peer holding the 

542 # slot is waited out instead of overwritten. 

543 _break_stale_marker(self._paths.write, stale_threshold=self.stale_threshold, now=time.time()) 

544 if _file_exists(self._paths.write): 

545 return False 

546 try: 

547 _atomic_create_marker(self._paths.write, token) 

548 except FileExistsError: 

549 return False 

550 return True 

551 

552 def _touch_writer_marker_if_ours(self, token: str) -> bool: 

553 # Refresh the writer marker through a single O_NOFOLLOW fd, but only while it still carries our 

554 # token. Returns False when the marker is gone or now belongs to a peer that reclaimed the slot, 

555 # so the caller can re-claim rather than keep a stranger's marker alive. Mirrors _refresh_marker. 

556 fd = _open_marker(self._paths.write) 

557 if fd is None: 

558 return False 

559 try: 

560 try: 

561 data = os.read(fd, _MAX_MARKER_SIZE + 1) 

562 except OSError: # pragma: no cover - e.g. EAGAIN from a hostile FIFO that has a writer attached 

563 return False 

564 info = _parse_marker_bytes(data) 

565 if info is None or not hmac.compare_digest(info.token, token): 

566 return False 

567 with suppress(OSError): 

568 touch(self._paths.write, fd=fd) 

569 return True 

570 finally: 

571 os.close(fd) 

572 

573 def _acquire_reader_slot( 

574 self, 

575 token: str, 

576 *, 

577 deadline: float | None, 

578 blocking: bool, 

579 ) -> tuple[str, bool]: 

580 self._open_readers_dir() 

581 reader_name = f"{uuid.uuid4().hex}.{os.getpid()}" 

582 dir_fd = self._readers_dir_fd 

583 full_reader_path = str(Path(self._paths.readers) / reader_name) 

584 

585 def try_claim_reader() -> bool: 

586 _break_stale_marker(self._paths.write, stale_threshold=self.stale_threshold, now=time.time()) 

587 if _file_exists(self._paths.write): 

588 return False 

589 if dir_fd is not None: # pragma: needs dir-fd 

590 _atomic_create_marker(reader_name, token, dir_fd=dir_fd) 

591 else: # pragma: win32 cover 

592 _atomic_create_marker(full_reader_path, token) 

593 return True 

594 

595 self._wait_for(try_claim_reader, deadline=deadline, blocking=blocking) 

596 return (reader_name if dir_fd is not None else full_reader_path), True 

597 

598 def _wait_for( 

599 self, 

600 predicate: Callable[[], bool], 

601 *, 

602 deadline: float | None, 

603 blocking: bool, 

604 ) -> None: 

605 while True: 

606 # One retry loop owns the deadline, including contention on the state mutex. 

607 try: 

608 with self._locks.state.acquire(blocking=False): 

609 if predicate(): 

610 return 

611 except Timeout: 

612 pass 

613 now = time.perf_counter() 

614 if not blocking: 

615 raise Timeout(self.lock_file) 

616 if deadline is not None and now >= deadline: 

617 raise Timeout(self.lock_file) 

618 sleep_for = self.poll_interval 

619 if deadline is not None: 

620 sleep_for = min(sleep_for, max(deadline - now, 0.0)) 

621 time.sleep(sleep_for) 

622 

623 def _open_readers_dir(self) -> None: 

624 readers_path = Path(self._paths.readers) 

625 with suppress(FileExistsError): 

626 readers_path.mkdir(mode=0o700) 

627 # mkdir has no O_NOFOLLOW, so verify via lstat that we did not land on an attacker-placed symlink 

628 # or a regular file before we open or scan inside. 

629 st = os.lstat(self._paths.readers) 

630 if stat.S_ISLNK(st.st_mode) or not stat.S_ISDIR(st.st_mode): 

631 msg = f"{self._paths.readers} exists but is not a directory or is a symlink; refusing to use it" 

632 raise RuntimeError(msg) 

633 if self._readers_dir_fd is None and _SUPPORTS_DIR_FD: # pragma: needs dir-fd 

634 with _fork_transition(): 

635 fd = os.open(self._paths.readers, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | _O_NOFOLLOW) 

636 try: 

637 token = _register_owned_descriptor(fd) 

638 except BaseException as registration_error: 

639 try: 

640 os.close(fd) 

641 except BaseException as close_error: # ruff:ignore[blind-except] # both errors surface via the group below 

642 _raise_grouped_errors( 

643 "reader directory registration and descriptor close both failed", 

644 registration_error, 

645 close_error, 

646 ) 

647 raise 

648 self._readers_dir_fd = fd 

649 self._readers_dir_fd_token = token 

650 

651 def _any_readers(self) -> bool: 

652 with closing(self._iter_reader_entries()) as entries: 

653 for _ in entries: 

654 return True 

655 return False 

656 

657 def _iter_reader_entries(self) -> Generator[tuple[str, bool]]: 

658 """ 

659 Yield ``(name, dirfd_relative)`` pairs for every live reader marker. 

660 

661 ``dirfd_relative`` is ``True`` when *name* should be passed to ``dir_fd=``-aware syscalls; ``False`` 

662 when *name* is a full path because dirfd-relative I/O is unavailable on this platform. 

663 

664 A consumer that stops early must close this generator: while suspended it holds the ``scandir`` handle open, 

665 and leaving that to the collector surfaces as an unraisable exception inside whatever runs next. 

666 """ 

667 if self._readers_dir_fd is not None: # pragma: needs dir-fd 

668 with os.scandir(self._readers_dir_fd) as it: 

669 for entry in it: 

670 if not _is_housekeeping_name(entry.name): 

671 yield entry.name, True 

672 return 

673 readers_path = Path(self._paths.readers) # pragma: win32 cover 

674 with os.scandir(readers_path) as it: # pragma: win32 cover 

675 for entry in it: # pragma: win32 cover 

676 if not _is_housekeeping_name(entry.name): # pragma: win32 cover 

677 yield str(readers_path / entry.name), False # pragma: win32 cover 

678 

679 def _break_stale_readers(self, now: float) -> None: 

680 names: list[tuple[str, int | None]] = [] 

681 try: 

682 with closing(self._iter_reader_entries()) as entries: 

683 for name, dirfd_relative in entries: 

684 names.append((name, self._readers_dir_fd if dirfd_relative else None)) 

685 except OSError: # pragma: no cover - transient NFS scandir hiccup 

686 return 

687 for name, fd in names: 

688 _break_stale_marker(name, stale_threshold=self.stale_threshold, now=now, dir_fd=fd) 

689 

690 def _refresh_marker(self) -> bool: 

691 with self._locks.internal: 

692 hold = self._hold 

693 if hold is None: # pragma: no cover - race between stop_event.set and join 

694 return False 

695 marker_name = hold.marker_name 

696 token = hold.token 

697 dir_fd = self._readers_dir_fd if hold.is_reader else None 

698 

699 # Open once with O_NOFOLLOW and touch that exact descriptor. Refreshing through the verified fd 

700 # (instead of re-opening by name) closes the window where a peer unlinks our marker and drops a symlink 

701 # or a different file at the path between the read and the touch: utime then lands on the inode we 

702 # verified, or nowhere. Only an unambiguous loss stops the heartbeat: the marker gone, or a peer's token 

703 # in its place. A transient filesystem error (ESTALE / EIO on the NFS-style filesystems this lock targets) 

704 # keeps the heartbeat alive to retry next tick, the way the touch below already does, so one blip does not 

705 # silently drop a held lock. 

706 try: 

707 fd = _open_marker_fd(marker_name, dir_fd=dir_fd) 

708 except FileNotFoundError: 

709 return False 

710 except OSError: 

711 return True 

712 try: 

713 try: 

714 data = _read_marker_fd(fd) 

715 except OSError: # a transient read error or EAGAIN from a hostile FIFO; retry rather than drop the lock 

716 return True 

717 info = _parse_marker_bytes(data) 

718 # Token mismatch means another process already evicted our marker and created its own; stop the 

719 # thread so it does not keep a stranger's file alive. 

720 if info is None or not hmac.compare_digest(info.token, token): 

721 return False 

722 # A transient touch failure (ESTALE / EIO on the NFS-style filesystems this lock targets) must not 

723 # kill the heartbeat thread: the read above just confirmed the marker is still ours, so swallow the 

724 # error and retry on the next tick rather than letting the lease lapse while we still hold the lock. 

725 with suppress(OSError): 

726 touch(marker_name, fd=fd) 

727 return True 

728 finally: 

729 os.close(fd) 

730 

731 def _reset_after_fork_in_child(self) -> None: # pragma: forked child 

732 self._locks = _Locks( 

733 internal=threading.Lock(), 

734 transaction=threading.Lock(), 

735 state=self._locks.state, 

736 ) 

737 self._hold = None 

738 self._readers_dir_fd = None 

739 self._readers_dir_fd_token = None 

740 

741 

742class _HeartbeatThread(threading.Thread): 

743 def __init__( 

744 self, 

745 refresh: Callable[[], bool], 

746 interval: float, 

747 stop_event: threading.Event, 

748 name: str, 

749 ) -> None: 

750 super().__init__(name=name, daemon=True) 

751 self._refresh = refresh 

752 self._interval = interval 

753 self._stop_event = stop_event 

754 

755 def run(self) -> None: 

756 while not self._stop_event.wait(self._interval): 

757 if not self._refresh(): 

758 self._stop_event.set() 

759 return 

760 

761 

762def _read_marker(name: str, *, dir_fd: int | None = None) -> tuple[_MarkerInfo | None, float] | None: 

763 fd = _open_marker(name, dir_fd=dir_fd) 

764 if fd is None: 

765 return None 

766 try: 

767 st = os.fstat(fd) 

768 # A legitimate marker is a regular file, so anything else at the path (a FIFO, say) is reported as a 

769 # malformed marker (its mtime still drives stale eviction) without being read. Reading is where 

770 # platforms diverge: an empty non-blocking read yields 0 bytes on Linux/macOS but EAGAIN on FreeBSD, 

771 # and the EAGAIN used to abort the stale-break and wedge the acquire until timeout (#587). 

772 if not stat.S_ISREG(st.st_mode): # pragma: needs fifo 

773 return None, st.st_mtime 

774 data = os.read(fd, _MAX_MARKER_SIZE + 1) 

775 except OSError: # pragma: no cover - marker vanished or turned unreadable between open and read 

776 return None 

777 finally: 

778 os.close(fd) 

779 return _parse_marker_bytes(data), st.st_mtime 

780 

781 

782def _read_marker_fd(fd: int) -> bytes: 

783 return os.read(fd, _MAX_MARKER_SIZE + 1) 

784 

785 

786def _open_marker_fd(name: str, *, dir_fd: int | None = None) -> int: 

787 # The file is ours; these guard a hostile mid-flight swap. O_NOFOLLOW rejects a symlink; O_NONBLOCK keeps 

788 # a real FIFO from blocking the open forever, so it reads as a malformed marker instead of wedging a peer 

789 # that holds the state lock. 

790 flags = os.O_RDONLY | _O_NOFOLLOW | _O_NONBLOCK 

791 return os.open(name, flags, dir_fd=dir_fd) if _SUPPORTS_DIR_FD and dir_fd is not None else os.open(name, flags) 

792 

793 

794def _open_marker(name: str, *, dir_fd: int | None = None) -> int | None: 

795 try: 

796 return _open_marker_fd(name, dir_fd=dir_fd) 

797 except OSError: 

798 return None 

799 

800 

801def _parse_marker_bytes(data: bytes) -> _MarkerInfo | None: 

802 # Trust nothing about attacker-controlled markers; any deviation returns None so callers fall through 

803 # to stale cleanup. ``re.match`` caches compiled patterns internally, so the regex is built only once 

804 # despite being defined inline. 

805 if not data or len(data) > _MAX_MARKER_SIZE: 

806 return None 

807 try: 

808 text = data.decode("ascii") 

809 except UnicodeDecodeError: 

810 return None 

811 match = re.match( 

812 r""" 

813 \A # start of string 

814 (?P<token> [0-9a-f]{32} ) \n # 128-bit hex token 

815 (?P<pid> [1-9][0-9]{0,9} ) \n # decimal pid: no leading zero, ≤ 10 digits 

816 (?P<hostname> [\x21-\x7e]{1,253}) # printable non-whitespace ASCII (RFC 1123 hostname limit) 

817 \n* # tolerate sloppy writers that append extra newlines 

818 \Z # end of string 

819 """, 

820 text, 

821 re.VERBOSE, 

822 ) 

823 if match is None: 

824 return None 

825 pid = int(match["pid"], 10) 

826 if pid > 2**31 - 1: 

827 return None 

828 return _MarkerInfo(token=match["token"], pid=pid, hostname=match["hostname"]) 

829 

830 

831def _unlink(name: str, *, dir_fd: int | None = None) -> None: 

832 with suppress(FileNotFoundError): 

833 if _SUPPORTS_DIR_FD and dir_fd is not None: # pragma: needs dir-fd 

834 # Path.unlink has no dir_fd support, so we stay on os.unlink for the dirfd path. 

835 os.unlink(name, dir_fd=dir_fd) 

836 else: 

837 Path(name).unlink() 

838 

839 

840def _break_stale_marker( # ruff:ignore[too-many-return-statements] # each return is a distinct abort/commit point in the break protocol 

841 name: str,