from __future__ import annotations import logging import os import pathlib import shutil import sqlite3 import stat import sys import tempfile import threading import time from contextlib import contextmanager, suppress from typing import TYPE_CHECKING, ClassVar, Final, Literal, TypeAlias, cast from weakref import WeakValueDictionary from ._api import ( AcquireReturnProxy, _canonical, _ensure_current_process, _fork_transition, _raise_chained_errors, _register_fork_class, _register_fork_object, ) from ._error import Timeout if sys.platform == "win32": # pragma: win32 cover from ._windows import _open_non_reparse_fd if TYPE_CHECKING: from collections.abc import Callable, Generator from _typeshed import Unused if sys.version_info >= (3, 11): from typing import Self else: from typing_extensions import Self _LOGGER: Final[logging.Logger] = logging.getLogger("filelock") _GETPID: Final[Callable[[], int]] = os.getpid _IS_PYPY: Final[bool] = sys.implementation.name == "pypy" _NEEDS_CONNECTION_ESCROW: Final[bool] = ( hasattr(os, "register_at_fork") and sys.implementation.name == "cpython" and sys.version_info < (3, 12) ) _ConnectionParameter: TypeAlias = ( str | bytes | os.PathLike[str] | os.PathLike[bytes] | float | int | type[sqlite3.Connection] | None ) _DatabaseIdentity: TypeAlias = tuple[int, int] # sqlite3_busy_timeout() accepts a C int, max 2_147_483_647 on 32-bit. Use a lower value to be safe (~23 days). _MAX_SQLITE_TIMEOUT_MS: Final[int] = 2_000_000_000 - 1 _UNSAFE_FORK_EXIT_STATUS: Final[int] = 70 # O_NONBLOCK keeps an open from blocking on a FIFO planted at the path; the regular-file check then rejects it. _DB_OPEN_FLAGS: Final[int] = os.O_RDWR | os.O_CREAT | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_NONBLOCK", 0) # Linux names an open descriptor under /proc/self/fd, macOS and the BSDs under /dev/fd. SQLite connects through that # name, so it reopens the inode already validated, and journal_mode=MEMORY keeps it from deriving an on-disk journal # name from the synthetic path. _FD_DIR: Final[str] = "/proc/self/fd" if sys.platform == "linux" else "/dev/fd" class _SQLiteTransitionContext(threading.local): depth: int = 0 _SQLITE_TRANSITION_CONTEXT: Final = _SQLiteTransitionContext() class _ConnectionEscrow: def __init__(self) -> None: self._lock = threading.RLock() self._functions: tuple[Callable[[sqlite3.Connection], None], Callable[[sqlite3.Connection], None]] | None = None def functions( self, ) -> tuple[Callable[[sqlite3.Connection], None], Callable[[sqlite3.Connection], None]] | None: if not _NEEDS_CONNECTION_ESCROW: return None # pragma: >=3.12 cover with self._lock: # pragma: <3.12 cover # pragma: needs fork if self._functions is None: import ctypes # ruff:ignore[import-outside-top-level] # keep optional ctypes and its audited dlsym out of ordinary imports function_type = ctypes.PYFUNCTYPE(None, ctypes.py_object) increment_address = ctypes.cast(ctypes.pythonapi.Py_IncRef, ctypes.c_void_p).value decrement_address = ctypes.cast(ctypes.pythonapi.Py_DecRef, ctypes.c_void_p).value if increment_address is None or decrement_address is None: # pragma: no cover - resolved CPython API msg = "CPython reference functions have no address" raise RuntimeError(msg) self._functions = ( cast("Callable[[sqlite3.Connection], None]", function_type(increment_address)), cast("Callable[[sqlite3.Connection], None]", function_type(decrement_address)), ) return self._functions def _reset_after_fork_in_child(self) -> None: # pragma: forked child self._lock = threading.RLock() _CONNECTION_ESCROW: Final = _ConnectionEscrow() class _ForkedDatabaseRegistry: def __init__(self) -> None: self._lock = threading.RLock() self._paths: set[pathlib.Path] = set() self._identities: set[_DatabaseIdentity] = set() self._sqlite_used = False self._all_paths_poisoned = False def raise_if_poisoned(self, path: pathlib.Path) -> None: identity = self.identity(path) with self._lock: all_paths_poisoned = self._all_paths_poisoned poisoned = ( all_paths_poisoned or path in self._paths or (identity is not None and identity in self._identities) ) if poisoned: # pragma: needs fork msg = ( "ReadWriteLock is unavailable in a PyPy fork child; exec or exit before using it" if all_paths_poisoned else f"SQLite database {path!s} was active across fork(); exec or exit before using it in the child" ) raise RuntimeError(msg) def poison_after_fork(self, path: pathlib.Path, identity: _DatabaseIdentity | None) -> None: self._paths.add(path) if identity is not None: self._identities.add(identity) def note_sqlite_use(self) -> None: if _IS_PYPY: with self._lock: self._sqlite_used = True def _reset_after_fork_in_child(self) -> None: # pragma: forked child self._lock = threading.RLock() self._all_paths_poisoned = self._all_paths_poisoned or (_IS_PYPY and self._sqlite_used) self._sqlite_used = False @staticmethod def identity(path: pathlib.Path) -> _DatabaseIdentity | None: try: stat_result = path.stat() except OSError: return None return stat_result.st_dev, stat_result.st_ino _FORKED_DATABASES: Final = _ForkedDatabaseRegistry() class _ForkSafeConnection(sqlite3.Connection): _creator_pid: int _decrement_escrow: Callable[[sqlite3.Connection], None] | None _database_directory: pathlib.Path | None def __new__( cls, *_args: _ConnectionParameter, **_kwargs: _ConnectionParameter, ) -> Self: connection = super().__new__(cls) connection._creator_pid = _GETPID() connection._decrement_escrow = None connection._database_directory = None return connection def close(self) -> None: with _sqlite_transition(): if _GETPID() != self._creator_pid: # pragma: needs fork return with _fork_transition(): sqlite3.Connection.close(self) if (directory := self._database_directory) is not None: # pragma: needs posix-hard-link shutil.rmtree(directory) self._database_directory = None if (decrement := self._decrement_escrow) is not None: # pragma: <3.12 cover # pragma: needs fork self._decrement_escrow = None decrement(self) def retain_database_directory(self, directory: pathlib.Path | None) -> None: """Keep the private link until SQLite no longer checks its path.""" self._database_directory = directory def acquire_escrow( # pragma: <3.12 cover # pragma: needs fork self, functions: tuple[Callable[[sqlite3.Connection], None], Callable[[sqlite3.Connection], None]] | None, ) -> None: # The caller only reaches here holding the escrow functions; it skips the call entirely without them. if functions is not None: # pragma: no branch increment, decrement = functions increment(self) self._decrement_escrow = decrement def __del__(self) -> None: with suppress(sqlite3.Error, RuntimeError, OSError): self.close() class _ReadWriteLockMeta(type): """ Resolve singleton instances for ``is_singleton=True`` construction. This logic lives here rather than in ReadWriteLock.get_lock so ``ReadWriteLock(path)`` returns cached instances without a 2-arg ``super()`` call that type checkers cannot verify. """ _instances: WeakValueDictionary[pathlib.Path, ReadWriteLock] _instances_lock: threading.RLock _instances_pid: int _instances_under_construction: set[pathlib.Path] def __call__( cls, lock_file: str | os.PathLike[str], timeout: float = -1, *, blocking: bool = True, is_singleton: bool = True, ) -> ReadWriteLock: _ensure_current_process() if cls._instances_pid != _GETPID(): cls._reset_class_after_fork() construction_pid = _GETPID() if not is_singleton: instance = super().__call__(lock_file, timeout, blocking=blocking, is_singleton=is_singleton) if _GETPID() != construction_pid: # pragma: forked child msg = "ReadWriteLock construction cannot continue after fork" raise RuntimeError(msg) return instance # Resolve only the parent, so a symlink at the lock path stays its own key instead of aliasing its target. normalized = pathlib.Path(_canonical(lock_file)) with cls._instances_lock: if normalized not in cls._instances: if normalized in cls._instances_under_construction: # pragma: no cover - exercised in an audit callback msg = f"Singleton lock construction is already active for {lock_file!s}" raise RuntimeError(msg) construction_registry = cls._instances_under_construction construction_registry.add(normalized) try: instance = super().__call__(lock_file, timeout, blocking=blocking, is_singleton=is_singleton) finally: if _GETPID() == construction_pid: construction_registry.discard(normalized) if _GETPID() != construction_pid: msg = "ReadWriteLock construction cannot continue after fork" raise RuntimeError(msg) cls._instances[normalized] = instance else: instance = cls._instances[normalized] if instance.timeout != timeout or instance.blocking != blocking: msg = ( f"Singleton lock created with timeout={instance.timeout}, blocking={instance.blocking}," f" cannot be changed to timeout={timeout}, blocking={blocking}" ) raise ValueError(msg) return instance def _reset_class_after_fork(cls) -> None: # pragma: forked child cls._instances = WeakValueDictionary() cls._instances_lock = threading.RLock() cls._instances_pid = _GETPID() cls._instances_under_construction = set() class ReadWriteLock(metaclass=_ReadWriteLockMeta): """ Cross-process read-write lock backed by SQLite. Allows concurrent shared readers or a single exclusive writer. The lock is reentrant within the same mode (multiple ``acquire_read`` calls nest, as do multiple ``acquire_write`` calls from the same thread), but upgrading from read to write or downgrading from write to read raises :class:`RuntimeError`. Write locks are pinned to the thread that acquired them. By default, ``is_singleton=True``: calling ``ReadWriteLock(path)`` with the same resolved path returns the same instance. The path is handed to :func:`sqlite3.connect` as given, so a ``.db`` extension is a convention rather than a requirement; the filesystem must be one the active SQLite VFS supports. :param lock_file: path to the SQLite database file used as the lock :param timeout: maximum wait time in seconds; ``-1`` means block indefinitely :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately when the lock is unavailable :param is_singleton: if ``True``, reuse existing instances for the same resolved path .. versionadded:: 3.21.0 """ _instances: WeakValueDictionary[pathlib.Path, ReadWriteLock] = WeakValueDictionary() _instances_lock = threading.RLock() _instances_pid = _GETPID() _instances_under_construction: ClassVar[set[pathlib.Path]] = set() def __init_subclass__(cls) -> None: super().__init_subclass__() cls._instances = WeakValueDictionary() cls._instances_lock = threading.RLock() cls._instances_pid = _GETPID() cls._instances_under_construction = set() _register_fork_class(cls) @classmethod def get_lock( cls, lock_file: str | os.PathLike[str], timeout: float = -1, *, blocking: bool = True ) -> ReadWriteLock: """ Return the singleton :class:`ReadWriteLock` for *lock_file*. :param lock_file: path to the SQLite database file used as the lock :param timeout: maximum wait time in seconds; ``-1`` means block indefinitely :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately when the lock is unavailable :returns: the singleton lock instance :raises ValueError: if an instance already exists for this path with different *timeout* or *blocking* values """ return cls(lock_file, timeout, blocking=blocking) def __init__( self, lock_file: str | os.PathLike[str], timeout: float = -1, *, blocking: bool = True, is_singleton: bool = True, # ruff:ignore[unused-method-argument] # consumed by _ReadWriteLockMeta.__call__ ) -> None: self.lock_file = os.fspath(lock_file) self._canonical_path = pathlib.Path(_canonical(lock_file)) _FORKED_DATABASES.raise_if_poisoned(self._canonical_path) self.timeout = timeout self.blocking = blocking self._transaction_lock = threading.Lock() # serializes the (possibly blocking) SQLite transaction work self._internal_lock = threading.Lock() # protects _lock_level / _current_mode updates and rollback self._lock_level = 0 self._current_mode: Literal["read", "write"] | None = None self._write_thread_id: int | None = None self._acquisition_thread_ids: set[int] = set() self._con: _ForkSafeConnection | None = None self._connection_transaction_released = True self._connection_identity: _DatabaseIdentity | None = None self._closed = False self._creator_pid = _GETPID() self._fork_invalidated = False _register_fork_object(self) with _fork_transition(), _sqlite_transition(): validation_connection = self._open_connection(sqlite_timeout=5.0) validation_connection.close() def acquire_read(self, timeout: float | None = None, *, blocking: bool | None = None) -> AcquireReturnProxy: """ Acquire a shared read lock. If this instance already holds a read lock, the lock level is incremented (reentrant). Attempting to acquire a read lock while holding a write lock raises :class:`RuntimeError` (downgrade not allowed). :param timeout: seconds to wait; ``None`` uses the instance setting; ``-1`` waits without a limit :param blocking: if ``False``, raise :class:`~filelock.Timeout` on contention; ``None`` uses the instance setting :returns: a proxy that can be used as a context manager to release the lock :raises RuntimeError: if a write lock is already held on this instance :raises Timeout: if the lock cannot be acquired within *timeout* seconds """ return self._acquire( "read", self.timeout if timeout is None else timeout, blocking=self.blocking if blocking is None else blocking, ) def acquire_write(self, timeout: float | None = None, *, blocking: bool | None = None) -> AcquireReturnProxy: """ Acquire an exclusive write lock. If this instance already holds a write lock from the same thread, the lock level is incremented (reentrant). Attempting to acquire a write lock while holding a read lock raises :class:`RuntimeError` (upgrade not allowed). Write locks are pinned to the acquiring thread: a different thread trying to re-enter also raises :class:`RuntimeError`. :param timeout: seconds to wait; ``None`` uses the instance setting; ``-1`` waits without a limit :param blocking: if ``False``, raise :class:`~filelock.Timeout` on contention; ``None`` uses the instance setting :returns: a proxy that can be used as a context manager to release the lock :raises RuntimeError: if a read lock is already held, or a write lock is held by a different thread :raises Timeout: if the lock cannot be acquired within *timeout* seconds """ return self._acquire( "write", self.timeout if timeout is None else timeout, blocking=self.blocking if blocking is None else blocking, ) @contextmanager def read_lock(self, timeout: float | None = None, *, blocking: bool | None = None) -> Generator[None]: """ Context manager that acquires and releases a shared read lock. Falls back to instance defaults for *timeout* and *blocking* when ``None``. :param timeout: maximum wait time in seconds, or ``None`` to use the instance default :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately; ``None`` uses the instance default """ self.acquire_read(timeout, blocking=blocking) try: yield finally: self.release() @contextmanager def write_lock(self, timeout: float | None = None, *, blocking: bool | None = None) -> Generator[None]: """ Context manager that acquires and releases an exclusive write lock. Falls back to instance defaults for *timeout* and *blocking* when ``None``. :param timeout: maximum wait time in seconds, or ``None`` to use the instance default :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately; ``None`` uses the instance default """ self.acquire_write(timeout, blocking=blocking) try: yield finally: self.release() def release(self, *, force: bool = False) -> None: """ Release one level of the current lock. When the lock level reaches zero the underlying SQLite transaction is rolled back, releasing the database lock. :param force: if ``True``, release the lock completely regardless of the current lock level :raises RuntimeError: if no lock is currently held and *force* is ``False``, or if a write lock is held by another thread and *force* is ``False`` """ with _fork_transition(): _ensure_current_process() if self._inherited: # pragma: needs fork return self._raise_if_acquiring("release") self._release(force=force, close=False) def close(self) -> None: """ Release the lock (if held) and close the underlying SQLite connection. After calling this method, the lock instance is no longer usable. """ with _fork_transition(): _ensure_current_process() if self._inherited: # pragma: needs fork return self._raise_if_acquiring("close") self._release(force=True, close=True) def _release(self, *, force: bool, close: bool) -> None: with self._transaction_lock, self._internal_lock: # The connection runs with check_same_thread=False, so a release from another thread would end the owner's # transaction under it and let a second writer in. if not force and self._current_mode == "write" and (cur := threading.get_ident()) != self._write_thread_id: msg = ( f"Cannot release write lock on {self.lock_file} (lock id: {id(self)}) " f"from thread {cur} while it is held by thread {self._write_thread_id}" ) raise RuntimeError(msg) if self._lock_level == 0: if force and self._con is None: if close: self._closed = True return if not force: msg = f"Cannot release a lock on {self.lock_file} (lock id: {id(self)}) that is not held" raise RuntimeError(msg) if not force and self._lock_level > 1: self._lock_level -= 1 return try: self._finish_connection() except sqlite3.Error: if self._connection_transaction_released: self._clear_lock_state() raise self._clear_lock_state() if close: self._closed = True def _clear_lock_state(self) -> None: self._lock_level = 0 self._current_mode = None self._write_thread_id = None def __del__(self) -> None: if _GETPID() == getattr(self, "_creator_pid", None) and (connection := getattr(self, "_con", None)) is not None: with suppress(sqlite3.Error, RuntimeError): connection.close() def _reset_after_fork_in_child(self) -> None: # pragma: forked child if self._con is not None: _FORKED_DATABASES.poison_after_fork(self._canonical_path, self._connection_identity) self._con = None self._connection_transaction_released = True self._connection_identity = None self._transaction_lock = threading.Lock() self._internal_lock = threading.Lock() self._clear_lock_state() self._acquisition_thread_ids = set() self._fork_invalidated = True @property def _inherited(self) -> bool: return self._fork_invalidated or _GETPID() != self._creator_pid def _raise_if_unusable(self) -> None: _ensure_current_process() if self._inherited: # pragma: needs fork msg = f"ReadWriteLock on {self.lock_file} was invalidated by fork(); construct a new instance" raise RuntimeError(msg) if self._closed: msg = "Cannot operate on a closed database." raise sqlite3.ProgrammingError(msg) def _acquire(self, mode: Literal["read", "write"], timeout: float, *, blocking: bool) -> AcquireReturnProxy: with _fork_transition(): self._raise_if_unusable() if blocking and timeout < 0 and timeout != -1: message: Final[str] = "timeout must be a non-negative number or -1" raise ValueError(message) operation_pid = _GETPID() thread_id = threading.get_ident() with self._internal_lock: if self._lock_level > 0: return self._validate_reentrant(mode) if thread_id in self._acquisition_thread_ids: # pragma: no cover - exercised in an audit callback msg = f"Cannot acquire ReadWriteLock on {self.lock_file} while acquisition is active in this thread" raise RuntimeError(msg) self._acquisition_thread_ids.add(thread_id) try: start_time = time.perf_counter() self._acquire_transaction_lock(blocking=blocking, timeout=timeout) try: self._raise_if_unusable() return self._do_acquire_inner( mode, timeout, blocking=blocking, operation_pid=operation_pid, start_time=start_time, ) finally: self._transaction_lock.release() finally: with self._internal_lock: self._acquisition_thread_ids.discard(thread_id) def _do_acquire_inner( self, mode: Literal["read", "write"], timeout: float, *, blocking: bool, operation_pid: int, start_time: float, ) -> AcquireReturnProxy: # Double-check: another thread may have acquired the lock while we waited on _transaction_lock. with self._internal_lock: if self._lock_level > 0: return self._validate_reentrant(mode) if self._con is not None: self._finish_connection() try: self._open_for_acquisition( timeout, blocking=blocking, operation_pid=operation_pid, start_time=start_time, ) self._configure_and_begin( mode, timeout, blocking=blocking, operation=(operation_pid, start_time), ) self._raise_if_process_changed(operation_pid) except BaseException as error: acquisition_error: BaseException if isinstance(error, sqlite3.OperationalError) and "database is locked" in str(error): acquisition_error = Timeout(self.lock_file) else: acquisition_error = error try: self._finish_connection() except sqlite3.Error as cleanup_error: _raise_chained_errors(acquisition_error, cleanup_error) if acquisition_error is not error: raise acquisition_error from None raise with self._internal_lock: self._raise_if_process_changed(operation_pid) self._current_mode = mode self._lock_level = 1 if mode == "write": self._write_thread_id = threading.get_ident() return AcquireReturnProxy(lock=self) def _open_for_acquisition(self, timeout: float, *, blocking: bool, operation_pid: int, start_time: float) -> None: with _sqlite_transition(): sqlite_timeout = ( timeout_for_sqlite( timeout, blocking=blocking, already_waited=time.perf_counter() - start_time, ) / 1000 ) connection = self._open_connection(sqlite_timeout=sqlite_timeout) self._con, self._connection_transaction_released, self._connection_identity = ( connection, False, _FORKED_DATABASES.identity(self._canonical_path), ) self._raise_if_process_changed(operation_pid) def _configure_and_begin( self, mode: Literal["read", "write"], timeout: float, *, blocking: bool, operation: tuple[int, float], ) -> None: with _sqlite_transition(): operation_pid, start_time = operation connection = cast("_ForkSafeConnection", self._con) waited = time.perf_counter() - start_time timeout_ms = timeout_for_sqlite(timeout, blocking=blocking, already_waited=waited) self._raise_if_process_changed(operation_pid) connection.executescript(f"PRAGMA busy_timeout={timeout_ms}; PRAGMA journal_mode=MEMORY;").close() # Use legacy journal mode (not WAL) because WAL does not block readers while a concurrent EXCLUSIVE # write transaction is active, which makes read-write locking impossible without modifying table data. # MEMORY is safe here since no writes happen, so a crash cannot corrupt the DB. # See https://sqlite.org/lang_transaction.html#deferred_immediate_and_exclusive_transactions # # Recompute the remaining timeout after the blocking journal_mode pragma. waited = time.perf_counter() - start_time recomputed = timeout_for_sqlite(timeout, blocking=blocking, already_waited=waited) self._raise_if_process_changed(operation_pid) statements = f"PRAGMA busy_timeout={recomputed}; " if recomputed != timeout_ms else "" statements += "BEGIN EXCLUSIVE TRANSACTION;" if mode == "write" else "BEGIN TRANSACTION;" if mode == "read": # SQLite takes the SHARED lock only when a statement reads; BEGIN alone stays deferred. # https://www.sqlite.org/lockingv3.html#transaction_control # sqlite_schema is an alias added in SQLite 3.33.0; sqlite_master works on every version. statements += " SELECT name FROM sqlite_master LIMIT 1;" connection.executescript(statements).close() def _open_connection(self, *, sqlite_timeout: float) -> _ForkSafeConnection: with _sqlite_transition(): creator_pid = _GETPID() functions = _CONNECTION_ESCROW.functions() if _GETPID() != creator_pid: # pragma: forked child msg = "SQLite connection construction cannot continue after fork" raise RuntimeError(msg) connection = _connect( os.fspath(self._canonical_path), factory=_ForkSafeConnection, timeout=sqlite_timeout, ) if functions is not None: # pragma: <3.12 cover # pragma: needs fork connection.acquire_escrow(functions) if _GETPID() != creator_pid: # pragma: forked child _FORKED_DATABASES.poison_after_fork( self._canonical_path, _FORKED_DATABASES.identity(self._canonical_path), ) msg = "SQLite connection construction cannot continue after fork" raise RuntimeError(msg) return connection def _finish_connection(self) -> None: with _sqlite_transition(): if (connection := self._con) is None: return rollback_error: sqlite3.Error | None = None if not self._connection_transaction_released: if connection.in_transaction: try: connection.rollback() except sqlite3.Error as error: if connection.in_transaction: raise self._connection_transaction_released = True rollback_error = error else: self._connection_transaction_released = True else: self._connection_transaction_released = True try: connection.close() except sqlite3.Error as close_error: if rollback_error is not None: _raise_chained_errors(rollback_error, close_error) raise self._con = None self._connection_transaction_released = True self._connection_identity = None if rollback_error is not None: raise rollback_error def _validate_reentrant(self, mode: Literal["read", "write"]) -> AcquireReturnProxy: if self._current_mode != mode: opposite = "write" if mode == "read" else "read" direction = "downgrade" if mode == "read" else "upgrade" msg = ( f"Cannot acquire {mode} lock on {self.lock_file} (lock id: {id(self)}): " f"already holding a {opposite} lock ({direction} not allowed)" ) raise RuntimeError(msg) if mode == "write" and (cur := threading.get_ident()) != self._write_thread_id: msg = ( f"Cannot acquire write lock on {self.lock_file} (lock id: {id(self)}) " f"from thread {cur} while it is held by thread {self._write_thread_id}" ) raise RuntimeError(msg) self._lock_level += 1 return AcquireReturnProxy(lock=self) def _acquire_transaction_lock(self, *, blocking: bool, timeout: float) -> None: if not blocking: acquired = self._transaction_lock.acquire(blocking=False) elif timeout == -1: acquired = self._transaction_lock.acquire(blocking=True) else: acquired = self._transaction_lock.acquire(blocking=True, timeout=timeout) if not acquired: raise Timeout(self.lock_file) from None def _raise_if_acquiring(self, operation: Literal["acquire", "close", "release"]) -> None: with self._internal_lock: active_in_current_thread = threading.get_ident() in self._acquisition_thread_ids if active_in_current_thread: # pragma: no cover - exercised in an audit callback msg = f"Cannot {operation} ReadWriteLock on {self.lock_file} while acquisition is active in this thread" raise RuntimeError(msg) def _raise_if_process_changed(self, operation_pid: int) -> None: if _GETPID() != operation_pid or self._inherited: # pragma: forked child msg = f"ReadWriteLock on {self.lock_file} was invalidated by fork(); construct a new instance" raise RuntimeError(msg) def _connect(database: str, *, factory: type[_ForkSafeConnection], timeout: float) -> _ForkSafeConnection: _FORKED_DATABASES.note_sqlite_use() # A symlink at the path would make SQLite open, lock, or create its target, so connect through a descriptor that # refuses one. SQLite opens its own descriptor on the file inside connect(), so ours can close after the call. fd: int | None = _open_lock_database(database) target = pathlib.Path(database if sys.platform == "win32" else f"{_FD_DIR}/{fd}") directory: pathlib.Path | None = None try: # NetBSD's static /dev/fd exposes only descriptors 0-63; a private hard link also pins the validated inode. if sys.platform != "win32" and not os.access(target, os.F_OK): # pragma: needs posix-hard-link directory = pathlib.Path(tempfile.mkdtemp(prefix=".filelock-")) target = directory / "lock.db" os.link(database, target, follow_symlinks=False) linked: Final = target.stat(follow_symlinks=False) opened: Final = os.fstat(fd) if (linked.st_dev, linked.st_ino) != (opened.st_dev, opened.st_ino): msg = f"lock database changed while opening: {database!r}" raise OSError(msg) # Closing any descriptor on this inode clears the process's POSIX locks; SQLite must open after this close. os.close(fd) fd = None connection: Final = sqlite3.connect( os.fspath(target), check_same_thread=False, factory=factory, cached_statements=0, timeout=timeout ) connection.retain_database_directory(directory) directory = None return connection finally: if fd is not None: # pragma: needs posix-hard-link os.close(fd) if directory is not None: # pragma: needs posix-hard-link shutil.rmtree(directory) def _open_lock_database(database: str) -> int: if sys.platform == "win32": # pragma: win32 cover # The handle refuses a reparse point and omits delete sharing, so the name cannot be swapped while it is open. if (fd := _open_non_reparse_fd(database, 0o600)) is None: msg = f"lock database {database!r} is being deleted or held without sharing" raise PermissionError(msg) return fd fd = os.open(database, _DB_OPEN_FLAGS, 0o600) # pragma: win32 no cover if not stat.S_ISREG(os.fstat(fd).st_mode): # pragma: win32 no cover os.close(fd) msg = f"refusing a non-regular lock database: {database!r}" raise OSError(msg) return fd # pragma: win32 no cover @contextmanager def _sqlite_transition() -> Generator[None]: _SQLITE_TRANSITION_CONTEXT.depth += 1 try: yield finally: _SQLITE_TRANSITION_CONTEXT.depth -= 1 def _abort_forked_sqlite_transition() -> None: # pragma: forked child if _SQLITE_TRANSITION_CONTEXT.depth: os._exit(_UNSAFE_FORK_EXIT_STATUS) # inherited SQLite handles cannot be used or closed safely def _track_sqlite_use(event: str, _args: Unused) -> None: if event == "sqlite3.connect": _FORKED_DATABASES.note_sqlite_use() def timeout_for_sqlite(timeout: float, *, blocking: bool, already_waited: float) -> int: if blocking is False: return 0 if timeout == -1: return _MAX_SQLITE_TIMEOUT_MS if timeout < 0: msg = "timeout must be a non-negative number or -1" raise ValueError(msg) timeout_ms = int((max(timeout - already_waited, 0) if timeout > 0 else timeout) * 1000) if timeout_ms > _MAX_SQLITE_TIMEOUT_MS or timeout_ms < 0: _LOGGER.warning("timeout %s is too large for SQLite, using %s ms instead", timeout, _MAX_SQLITE_TIMEOUT_MS) return _MAX_SQLITE_TIMEOUT_MS return timeout_ms _register_fork_object(_CONNECTION_ESCROW) _register_fork_object(_FORKED_DATABASES) _register_fork_class(ReadWriteLock) if _IS_PYPY: sys.addaudithook(_track_sqlite_use) # pragma: pypy cover if hasattr(os, "register_at_fork"): # pragma: needs fork os.register_at_fork(after_in_child=_abort_forked_sqlite_transition) __all__ = [ "ReadWriteLock", "timeout_for_sqlite", ]