Buckets:
MisterAI/LocalAI_Demo_backends / cpu-diffusers /venv /lib /python3.10 /site-packages /filelock /_lease.py
| 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"] | |
| class LeaseCompromise: | |
| """Why a held lease stopped being this process's to hold.""" | |
| lock_file: str | |
| token: str | |
| reason: CompromiseReason | |
| error: OSError | None = None | |
| class _Heartbeat: | |
| """A running heartbeat and the event that stops it, which only ever exist together.""" | |
| thread: Thread | |
| stop: Event | |
| 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 | |
| )() | |
| def _claim(self) -> _LeaseClaim: | |
| return self._claims.claim | |
| def lease_duration(self) -> float: | |
| """The staleness in seconds after which a contender may take this claim.""" | |
| return self._lease_duration | |
| 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 | |
| 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, | |
| ) | |
| # Record the heartbeat before starting the thread so a release racing this acquire on a shared, | |
| # non-thread-local claim always sees it and sets the stop event; the thread then exits at its first wait | |
| # instead of outliving the release. A start that raises leaves the unstarted thread for _stop_heartbeat. | |
| 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 | |
| # thread.ident is None until start() runs: a heartbeat recorded before its thread started (a start that | |
| # raised, or a release racing acquire on a shared claim) has nothing to join, and the stop above makes it | |
| # exit at once. on_compromise runs on the heartbeat thread and may release the lease, landing back here. | |
| if heartbeat.thread.ident is not None and 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", | |
| ] | |
Xet Storage Details
- Size:
- 15.2 kB
- Xet hash:
- f496c606712debddb0f95b98ed98dfac830e8bb8071a50b2fd6e46069240709f
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.