← back to Atmos Game
assets/gen/.venv/lib/python3.12/site-packages/filelock/_lease.py
330 lines
from __future__ import annotations
import os
import secrets
import time
from contextlib import suppress
from dataclasses import dataclass
from threading import Event, Thread, current_thread, local
from typing import TYPE_CHECKING, Literal
from ._error import LeaseSettingsMismatch
from ._identity import owner_is_stale
from ._marker import MarkerSoftFileLock, OwnerMode, OwnerRecord, parse_marker
from ._soft import _read_lock_file
from ._util import break_lock_file, touch
if TYPE_CHECKING:
import sys
from collections.abc import Callable
from ._api import LockOptions
if sys.version_info >= (3, 11): # pragma: no cover (py311+)
from typing import Unpack
else: # pragma: no cover (<py311)
from typing_extensions import Unpack
CompromiseReason = Literal["marker-missing", "owner-changed", "refresh-failed"]
_RefreshOutcome = Literal["ok", "lost", "transient"]
@dataclass(frozen=True)
class LeaseCompromise:
"""Why a held lease stopped being this process's to hold."""
lock_file: str
token: str
reason: CompromiseReason
error: OSError | None = None
@dataclass(frozen=True)
class _Heartbeat:
"""A running heartbeat and the event that stops it, which only ever exist together."""
thread: Thread
stop: Event
@dataclass
class _LeaseClaim:
"""The state of one claim: its token, its heartbeat, and how that claim was lost."""
token: str | None = None
compromise: LeaseCompromise | None = None
heartbeat: _Heartbeat | None = None
class _LeaseClaimHolder:
"""Holds the claim its owning context acquired."""
# Only the holder is thread-local, mirroring FileLockContext: the claim stays an ordinary object, so the heartbeat
# thread records a compromise where the thread that acquired the lease reads it.
def __init__(self) -> None:
self.claim = _LeaseClaim()
class _ThreadLocalLeaseClaimHolder(_LeaseClaimHolder, local):
"""A thread local version of the ``_LeaseClaimHolder`` class."""
class SoftFileLease(MarkerSoftFileLock):
"""
Existence lock whose claim expires, so a peer may take it while the previous holder still runs.
A lease trades mutual exclusion for progress. The holder publishes a claim and refreshes it every
``heartbeat_interval`` seconds; a contender takes the marker once it is ``lease_duration`` seconds stale. Nothing
stops the expired holder: it keeps running, and it keeps using whatever the lock protects. Treat the lease as a hint
about who *should* be working, not as a guarantee that only one worker is.
To make a protected resource reject a superseded holder, that resource must be linearizable and must fence on a
monotonic generation it controls. :attr:`token` names a claim; it does not fence one. Where overlap is unacceptable,
use :class:`StrictSoftFileLock <filelock.StrictSoftFileLock>` instead.
Every contender for a path must agree on ``lease_duration``. A contender that finds a claim published under a
different duration raises :class:`LeaseSettingsMismatch <filelock.LeaseSettingsMismatch>` rather than apply its own
expiry to a peer that never agreed to it.
Expiry reclaims less on Windows, which refuses to rename or delete a file another process holds open. A peer there
takes an expired claim only once the previous holder's process exits and its handle closes; a holder that lives on
but stops refreshing keeps the marker. Unix reclaims the marker either way.
``on_compromise`` fires from the heartbeat thread when a refresh fails, or when the marker vanishes or names another
owner. The holder should stop touching the protected resource when it runs. Because it runs on that thread, a
``release()`` inside it only takes effect when the lease was built with ``thread_local=False``; the default
thread-local context hides the claim from every thread but the one that acquired it, so the release does nothing.
Signal the acquiring thread instead when the context stays thread-local.
.. versionadded:: 3.30.0
"""
_owner_mode: OwnerMode = "lease"
#: lease_duration replaces the legacy age-based lifetime, so accepting both would give one lock two expiry clocks.
_lifetime_supported: bool = False
_lifetime_unsupported_reason: str = "lease_duration sets when a lease expires"
def __init__(
self,
lock_file: str | os.PathLike[str],
*,
lease_duration: float = 30.0,
heartbeat_interval: float | None = None,
on_compromise: Callable[[LeaseCompromise], None] | None = None,
**kwargs: Unpack[LockOptions],
) -> None:
"""
Create a lease.
:param lease_duration: seconds of marker staleness after which a contender may take the claim. Every contender
for the path must pass the same value.
:param heartbeat_interval: seconds between refreshes. Defaults to a third of ``lease_duration``, leaving room
for two missed refreshes before a peer may take the claim. Must be shorter than ``lease_duration``.
:param on_compromise: called from the heartbeat thread with a :class:`LeaseCompromise` when the claim is lost.
:param kwargs: every other :class:`BaseFileLock <filelock.BaseFileLock>` option, ``timeout`` and ``mode`` among
them. The metaclass passes them all by keyword, and taking them here lets
:class:`AsyncSoftFileLease <filelock.AsyncSoftFileLease>` add the async plumbing a fixed signature would
hide.
"""
if lease_duration <= 0:
msg = f"lease_duration must be positive, got {lease_duration!r}"
raise ValueError(msg)
if heartbeat_interval is None:
heartbeat_interval = lease_duration / 3
if not 0 < heartbeat_interval < lease_duration:
msg = f"heartbeat_interval must be positive and below lease_duration, got {heartbeat_interval!r}"
raise ValueError(msg)
super().__init__(lock_file, **kwargs)
self._lease_duration = lease_duration
self._heartbeat_interval = heartbeat_interval
self._on_compromise = on_compromise
# Sharing one claim across a thread-local lock lets a second thread's failed acquisition stop the heartbeat of
# the thread holding the lease, leaving its marker unrefreshed until a peer reclaims it.
self._claims: _LeaseClaimHolder = (
_ThreadLocalLeaseClaimHolder if self.is_thread_local() else _LeaseClaimHolder
)()
@property
def _claim(self) -> _LeaseClaim:
return self._claims.claim
@property
def lease_duration(self) -> float:
"""The staleness in seconds after which a contender may take this claim."""
return self._lease_duration
@property
def token(self) -> str | None:
"""
The token naming the claim this process published.
:returns: the token while the lease is held, ``None`` otherwise. It identifies a claim; it does not fence one.
"""
return self._claim.token
@property
def compromise(self) -> LeaseCompromise | None:
"""
The loss of claim the heartbeat observed.
:returns: the :class:`LeaseCompromise`, or ``None`` while the claim still holds
"""
return self._claim.compromise
def _acquire(self) -> None:
claim = self._claim
self._stop_heartbeat() # no earlier claim's heartbeat outlives the acquisition of the next one
claim.token = token = secrets.token_hex(16)
claim.compromise = None
super()._acquire()
# The context is thread-local by default, so the heartbeat thread cannot read the descriptor this one just
# published, nor the claim this one owns. Hand it the fd, the inode it verified and the claim instead.
if (fd := self._context.lock_file_fd) is not None and (
identity := self._context.lock_file_fd_identity
) is not None:
self._start_heartbeat(claim, fd, identity, token)
def _release(self) -> None:
self._stop_heartbeat()
self._claim.token = None
super()._release()
def _published_record(self) -> OwnerRecord:
return super()._published_record()._replace(token=self._claim.token, lease_duration=self._lease_duration)
def _try_break_stale_lock(self) -> None:
if (peer := self._read_peer()) is None:
# Not a readable protocol 2 lease record: a partial write, a foreign or legacy protocol 1 marker, or the
# strict sentinel. The base self-heal evicts a genuinely malformed marker once it ages past the grace
# window and leaves a legitimate legacy or strict holder in place, so a corrupt marker no longer wedges
# every lease contender until its own timeout.
super()._try_break_stale_lock()
return
owner, mtime, ino = peer
# Only a peer that published a lease agreed to be superseded by one, so a record stating any other contract is
# never reclaimed by age. Raise the mismatch outside the read so the suppression cannot swallow it.
if owner.mode != "lease":
return
if owner.lease_duration != self._lease_duration:
msg = (
f"{self.lock_file} holds a lease of {owner.lease_duration!r}s but this contender configured "
f"{self._lease_duration!r}s; every contender for a path must agree on lease_duration"
)
raise LeaseSettingsMismatch(msg)
# A break can fail for reasons a contender must ride out rather than raise on: a peer broke the marker first,
# or Windows refuses to rename a file whose holder still has it open. Poll again instead.
with suppress(OSError):
# A dead or recycled owner is reclaimed at once; a live owner past its lease duration is superseded on the
# schedule every contender agreed to.
if owner_is_stale(owner.pid, owner.hostname, owner.start):
break_lock_file(self.lock_file, mtime, ino)
return
if time.time() - mtime >= self._lease_duration:
break_lock_file(self.lock_file, mtime, ino)
def _read_peer(self) -> tuple[OwnerRecord, float, int] | None:
with suppress(OSError, ValueError):
content, mtime, ino = _read_lock_file(self.lock_file)
if (owner := parse_marker(content)) is not None:
return owner, mtime, ino
return None
def _start_heartbeat(self, claim: _LeaseClaim, fd: int, identity: tuple[int, int], token: str) -> None:
# The thread watches the event it was handed rather than whatever the claim names later: a heartbeat that
# outlives its join timeout would otherwise adopt the next acquisition's event and never stop.
stop = Event()
thread = Thread(
target=self._refresh_until_stopped,
args=(claim, fd, identity, token, stop),
name=f"filelock-lease-{os.getpid()}",
daemon=True,
)
claim.heartbeat = _Heartbeat(thread, stop)
thread.start()
def _stop_heartbeat(self) -> None:
claim = self._claim
if (heartbeat := claim.heartbeat) is None:
return
heartbeat.stop.set()
claim.heartbeat = None
# on_compromise runs on the heartbeat thread and may release the lease, which lands back here.
if heartbeat.thread is not current_thread():
heartbeat.thread.join(timeout=self._heartbeat_interval)
def _refresh_until_stopped(
self,
claim: _LeaseClaim,
fd: int,
identity: tuple[int, int],
token: str,
stop: Event,
) -> None:
# The loop ends at the first loss of the claim, so the holder hears about it once. A transient filesystem
# error (ESTALE / EIO on the NFS-style filesystems a lease targets) is not a loss: retry rather than raise a
# false compromise. Report the claim unrefreshable only once failures have run long enough that a contender
# could take it before the next success would land, a margin before the marker actually ages out, the way
# restic declares a lock unrefreshable ahead of its stale time.
last_success = time.monotonic()
while not stop.wait(self._heartbeat_interval):
outcome, error = self._refresh_claim(claim, fd, identity, token)
if outcome == "lost":
return
if outcome == "ok":
last_success = time.monotonic()
elif time.monotonic() - last_success >= self._lease_duration - self._heartbeat_interval:
self._report_compromise(claim, "refresh-failed", error, token)
return
def _refresh_claim(
self,
claim: _LeaseClaim,
fd: int,
identity: tuple[int, int],
token: str,
) -> tuple[_RefreshOutcome, OSError | None]:
try:
st = os.lstat(self.lock_file)
except FileNotFoundError as error:
self._report_compromise(claim, "marker-missing", error, token)
return "lost", None
except OSError as error:
return "transient", error
# A peer that took the expired claim replaced the marker, so the pathname now names its inode, not ours.
if (st.st_dev, st.st_ino) != identity:
self._report_compromise(claim, "owner-changed", None, token)
return "lost", None
try:
touch(self.lock_file, fd=fd)
except OSError as error:
return "transient", error
return "ok", None
def _report_compromise(
self,
claim: _LeaseClaim,
reason: CompromiseReason,
error: OSError | None,
token: str,
) -> None:
# Record it on the claim this heartbeat serves, not on self._claim: a thread-local claim read from the
# heartbeat thread is a different, empty one, so the holder would never see the loss it is being told about.
# The token is the one this thread published, not claim.token, which a release may already have cleared.
claim.compromise = LeaseCompromise(lock_file=self.lock_file, token=token, reason=reason, error=error)
if self._on_compromise is not None:
self._on_compromise(claim.compromise)
__all__ = [
"CompromiseReason",
"LeaseCompromise",
"SoftFileLease",
]