Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/filelock/_read_write.py: 24%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1from __future__ import annotations
3import logging
4import os
5import pathlib
6import sqlite3
7import sys
8import threading
9import time
10from contextlib import contextmanager, suppress
11from typing import TYPE_CHECKING, ClassVar, Final, Literal, TypeAlias, cast
12from weakref import WeakValueDictionary
14from ._api import (
15 AcquireReturnProxy,
16 _ensure_current_process,
17 _fork_transition,
18 _raise_chained_errors,
19 _register_fork_class,
20 _register_fork_object,
21)
22from ._error import Timeout
24if TYPE_CHECKING:
25 from collections.abc import Callable, Generator
27 if sys.version_info >= (3, 11):
28 from typing import Self
29 else:
30 from typing_extensions import Self
32_LOGGER: Final[logging.Logger] = logging.getLogger("filelock")
33_GETPID: Final[Callable[[], int]] = os.getpid
34_IS_PYPY: Final[bool] = sys.implementation.name == "pypy"
35_NEEDS_CONNECTION_ESCROW: Final[bool] = (
36 hasattr(os, "register_at_fork") and sys.implementation.name == "cpython" and sys.version_info < (3, 12)
37)
38_ConnectionParameter: TypeAlias = (
39 str | bytes | os.PathLike[str] | os.PathLike[bytes] | float | int | type[sqlite3.Connection] | None
40)
41_DatabaseIdentity: TypeAlias = tuple[int, int]
43# sqlite3_busy_timeout() accepts a C int, max 2_147_483_647 on 32-bit. Use a lower value to be safe (~23 days).
44_MAX_SQLITE_TIMEOUT_MS: Final[int] = 2_000_000_000 - 1
45_UNSAFE_FORK_EXIT_STATUS: Final[int] = 70
48class _SQLiteTransitionContext(threading.local):
49 depth: int = 0
52_SQLITE_TRANSITION_CONTEXT: Final = _SQLiteTransitionContext()
55class _ConnectionEscrow:
56 def __init__(self) -> None:
57 self._lock = threading.RLock()
58 self._functions: tuple[Callable[[sqlite3.Connection], None], Callable[[sqlite3.Connection], None]] | None = None
60 def functions(
61 self,
62 ) -> tuple[Callable[[sqlite3.Connection], None], Callable[[sqlite3.Connection], None]] | None:
63 if not _NEEDS_CONNECTION_ESCROW:
64 return None # pragma: >=3.12 cover
65 with self._lock: # pragma: <3.12 cover # pragma: needs fork
66 if self._functions is None:
67 import ctypes # ruff:ignore[import-outside-top-level] # keep optional ctypes and its audited dlsym out of ordinary imports
69 function_type = ctypes.PYFUNCTYPE(None, ctypes.py_object)
70 increment_address = ctypes.cast(ctypes.pythonapi.Py_IncRef, ctypes.c_void_p).value
71 decrement_address = ctypes.cast(ctypes.pythonapi.Py_DecRef, ctypes.c_void_p).value
72 if increment_address is None or decrement_address is None: # pragma: no cover - resolved CPython API
73 msg = "CPython reference functions have no address"
74 raise RuntimeError(msg)
75 self._functions = (
76 cast("Callable[[sqlite3.Connection], None]", function_type(increment_address)),
77 cast("Callable[[sqlite3.Connection], None]", function_type(decrement_address)),
78 )
79 return self._functions
81 def _reset_after_fork_in_child(self) -> None: # pragma: forked child
82 self._lock = threading.RLock()
85_CONNECTION_ESCROW: Final = _ConnectionEscrow()
88class _ForkedDatabaseRegistry:
89 def __init__(self) -> None:
90 self._lock = threading.RLock()
91 self._paths: set[pathlib.Path] = set()
92 self._identities: set[_DatabaseIdentity] = set()
93 self._sqlite_used = False
94 self._all_paths_poisoned = False
96 def raise_if_poisoned(self, path: pathlib.Path) -> None:
97 identity = self.identity(path)
98 with self._lock:
99 all_paths_poisoned = self._all_paths_poisoned
100 poisoned = (
101 all_paths_poisoned or path in self._paths or (identity is not None and identity in self._identities)
102 )
103 if poisoned: # pragma: needs fork
104 msg = (
105 "ReadWriteLock is unavailable in a PyPy fork child; exec or exit before using it"
106 if all_paths_poisoned
107 else f"SQLite database {path!s} was active across fork(); exec or exit before using it in the child"
108 )
109 raise RuntimeError(msg)
111 def poison_after_fork(self, path: pathlib.Path, identity: _DatabaseIdentity | None) -> None:
112 self._paths.add(path)
113 if identity is not None:
114 self._identities.add(identity)
116 def note_sqlite_use(self) -> None:
117 if _IS_PYPY:
118 with self._lock:
119 self._sqlite_used = True
121 def _reset_after_fork_in_child(self) -> None: # pragma: forked child
122 self._lock = threading.RLock()
123 self._all_paths_poisoned = self._all_paths_poisoned or (_IS_PYPY and self._sqlite_used)
124 self._sqlite_used = False
126 @staticmethod
127 def identity(path: pathlib.Path) -> _DatabaseIdentity | None:
128 try:
129 stat_result = path.stat()
130 except OSError:
131 return None
132 return stat_result.st_dev, stat_result.st_ino
135_FORKED_DATABASES: Final = _ForkedDatabaseRegistry()
138class _ForkSafeConnection(sqlite3.Connection):
139 _creator_pid: int
140 _decrement_escrow: Callable[[sqlite3.Connection], None] | None
142 def __new__(
143 cls,
144 *_args: _ConnectionParameter,
145 **_kwargs: _ConnectionParameter,
146 ) -> Self:
147 connection = super().__new__(cls)
148 connection._creator_pid = _GETPID()
149 connection._decrement_escrow = None
150 return connection
152 def close(self) -> None:
153 with _sqlite_transition():
154 if _GETPID() != self._creator_pid: # pragma: needs fork
155 return
156 with _fork_transition():
157 sqlite3.Connection.close(self)
158 if (decrement := self._decrement_escrow) is not None: # pragma: <3.12 cover # pragma: needs fork
159 self._decrement_escrow = None
160 decrement(self)
162 def acquire_escrow( # pragma: <3.12 cover # pragma: needs fork
163 self,
164 functions: tuple[Callable[[sqlite3.Connection], None], Callable[[sqlite3.Connection], None]] | None,
165 ) -> None:
166 # The caller only reaches here holding the escrow functions; it skips the call entirely without them.
167 if functions is not None: # pragma: no branch
168 increment, decrement = functions
169 increment(self)
170 self._decrement_escrow = decrement
172 def __del__(self) -> None:
173 with suppress(sqlite3.Error, RuntimeError):
174 self.close()
177class _ReadWriteLockMeta(type):
178 """
179 Resolve singleton instances for ``is_singleton=True`` construction.
181 This logic lives here rather than in ReadWriteLock.get_lock so ``ReadWriteLock(path)`` returns cached instances
182 without a 2-arg ``super()`` call that type checkers cannot verify.
184 """
186 _instances: WeakValueDictionary[pathlib.Path, ReadWriteLock]
187 _instances_lock: threading.RLock
188 _instances_pid: int
189 _instances_under_construction: set[pathlib.Path]
191 def __call__(
192 cls,
193 lock_file: str | os.PathLike[str],
194 timeout: float = -1,
195 *,
196 blocking: bool = True,
197 is_singleton: bool = True,
198 ) -> ReadWriteLock:
199 _ensure_current_process()
200 if cls._instances_pid != _GETPID():
201 cls._reset_class_after_fork()
202 construction_pid = _GETPID()
203 if not is_singleton:
204 instance = super().__call__(lock_file, timeout, blocking=blocking, is_singleton=is_singleton)
205 if _GETPID() != construction_pid: # pragma: forked child
206 msg = "ReadWriteLock construction cannot continue after fork"
207 raise RuntimeError(msg)
208 return instance
210 normalized = pathlib.Path(lock_file).resolve()
211 with cls._instances_lock:
212 if normalized not in cls._instances:
213 if normalized in cls._instances_under_construction: # pragma: no cover - exercised in an audit callback
214 msg = f"Singleton lock construction is already active for {lock_file!s}"
215 raise RuntimeError(msg)
216 construction_registry = cls._instances_under_construction
217 construction_registry.add(normalized)
218 try:
219 instance = super().__call__(lock_file, timeout, blocking=blocking, is_singleton=is_singleton)
220 finally:
221 if _GETPID() == construction_pid:
222 construction_registry.discard(normalized)
223 if _GETPID() != construction_pid:
224 msg = "ReadWriteLock construction cannot continue after fork"
225 raise RuntimeError(msg)
226 cls._instances[normalized] = instance
227 else:
228 instance = cls._instances[normalized]
230 if instance.timeout != timeout or instance.blocking != blocking:
231 msg = (
232 f"Singleton lock created with timeout={instance.timeout}, blocking={instance.blocking},"
233 f" cannot be changed to timeout={timeout}, blocking={blocking}"
234 )
235 raise ValueError(msg)
236 return instance
238 def _reset_class_after_fork(cls) -> None: # pragma: forked child
239 cls._instances = WeakValueDictionary()
240 cls._instances_lock = threading.RLock()
241 cls._instances_pid = _GETPID()
242 cls._instances_under_construction = set()
245class ReadWriteLock(metaclass=_ReadWriteLockMeta):
246 """
247 Cross-process read-write lock backed by SQLite.
249 Allows concurrent shared readers or a single exclusive writer. The lock is reentrant within the same mode (multiple
250 ``acquire_read`` calls nest, as do multiple ``acquire_write`` calls from the same thread), but upgrading from read
251 to write or downgrading from write to read raises :class:`RuntimeError`. Write locks are pinned to the thread that
252 acquired them.
254 By default, ``is_singleton=True``: calling ``ReadWriteLock(path)`` with the same resolved path returns the same
255 instance. The path is handed to :func:`sqlite3.connect` as given, so a ``.db`` extension is a convention rather
256 than a requirement; the filesystem must be one the active SQLite VFS supports.
258 :param lock_file: path to the SQLite database file used as the lock
259 :param timeout: maximum wait time in seconds; ``-1`` means block indefinitely
260 :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately when the lock is unavailable
261 :param is_singleton: if ``True``, reuse existing instances for the same resolved path
263 .. versionadded:: 3.21.0
265 """
267 _instances: WeakValueDictionary[pathlib.Path, ReadWriteLock] = WeakValueDictionary()
268 _instances_lock = threading.RLock()
269 _instances_pid = _GETPID()
270 _instances_under_construction: ClassVar[set[pathlib.Path]] = set()
272 def __init_subclass__(cls) -> None:
273 super().__init_subclass__()
274 cls._instances = WeakValueDictionary()
275 cls._instances_lock = threading.RLock()
276 cls._instances_pid = _GETPID()
277 cls._instances_under_construction = set()
278 _register_fork_class(cls)
280 @classmethod
281 def get_lock(
282 cls, lock_file: str | os.PathLike[str], timeout: float = -1, *, blocking: bool = True
283 ) -> ReadWriteLock:
284 """
285 Return the singleton :class:`ReadWriteLock` for *lock_file*.
287 :param lock_file: path to the SQLite database file used as the lock
288 :param timeout: maximum wait time in seconds; ``-1`` means block indefinitely
289 :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately when the lock is unavailable
291 :returns: the singleton lock instance
293 :raises ValueError: if an instance already exists for this path with different *timeout* or *blocking* values
295 """
296 return cls(lock_file, timeout, blocking=blocking)
298 def __init__(
299 self,
300 lock_file: str | os.PathLike[str],
301 timeout: float = -1,
302 *,
303 blocking: bool = True,
304 is_singleton: bool = True, # ruff:ignore[unused-method-argument] # consumed by _ReadWriteLockMeta.__call__
305 ) -> None:
306 self.lock_file = os.fspath(lock_file)
307 self._canonical_path = pathlib.Path(lock_file).resolve()
308 _FORKED_DATABASES.raise_if_poisoned(self._canonical_path)
309 self.timeout = timeout
310 self.blocking = blocking
311 self._transaction_lock = threading.Lock() # serializes the (possibly blocking) SQLite transaction work
312 self._internal_lock = threading.Lock() # protects _lock_level / _current_mode updates and rollback
313 self._lock_level = 0
314 self._current_mode: Literal["read", "write"] | None = None
315 self._write_thread_id: int | None = None
316 self._acquisition_thread_ids: set[int] = set()
317 self._con: _ForkSafeConnection | None = None
318 self._connection_transaction_released = True
319 self._connection_identity: _DatabaseIdentity | None = None
320 self._closed = False
321 self._creator_pid = _GETPID()
322 self._fork_invalidated = False
323 _register_fork_object(self)
324 with _fork_transition(), _sqlite_transition():
325 validation_connection = self._open_connection(sqlite_timeout=5.0)
326 validation_connection.close()
328 def acquire_read(self, timeout: float = -1, *, blocking: bool = True) -> AcquireReturnProxy:
329 """
330 Acquire a shared read lock.
332 If this instance already holds a read lock, the lock level is incremented (reentrant). Attempting to acquire a
333 read lock while holding a write lock raises :class:`RuntimeError` (downgrade not allowed).
335 :param timeout: maximum wait time in seconds; ``-1`` means block indefinitely
336 :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately when the lock is unavailable
338 :returns: a proxy that can be used as a context manager to release the lock
340 :raises RuntimeError: if a write lock is already held on this instance
341 :raises Timeout: if the lock cannot be acquired within *timeout* seconds
343 """
344 return self._acquire("read", timeout, blocking=blocking)
346 def acquire_write(self, timeout: float = -1, *, blocking: bool = True) -> AcquireReturnProxy:
347 """
348 Acquire an exclusive write lock.
350 If this instance already holds a write lock from the same thread, the lock level is incremented (reentrant).
351 Attempting to acquire a write lock while holding a read lock raises :class:`RuntimeError` (upgrade not allowed).
352 Write locks are pinned to the acquiring thread: a different thread trying to re-enter also raises
353 :class:`RuntimeError`.
355 :param timeout: maximum wait time in seconds; ``-1`` means block indefinitely
356 :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately when the lock is unavailable
358 :returns: a proxy that can be used as a context manager to release the lock
360 :raises RuntimeError: if a read lock is already held, or a write lock is held by a different thread
361 :raises Timeout: if the lock cannot be acquired within *timeout* seconds
363 """
364 return self._acquire("write", timeout, blocking=blocking)
366 @contextmanager
367 def read_lock(self, timeout: float | None = None, *, blocking: bool | None = None) -> Generator[None]:
368 """
369 Context manager that acquires and releases a shared read lock.
371 Falls back to instance defaults for *timeout* and *blocking* when ``None``.
373 :param timeout: maximum wait time in seconds, or ``None`` to use the instance default
374 :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately; ``None`` uses the instance default
376 """
377 if timeout is None:
378 timeout = self.timeout
379 if blocking is None:
380 blocking = self.blocking
381 self.acquire_read(timeout, blocking=blocking)
382 try:
383 yield
384 finally:
385 self.release()
387 @contextmanager
388 def write_lock(self, timeout: float | None = None, *, blocking: bool | None = None) -> Generator[None]:
389 """
390 Context manager that acquires and releases an exclusive write lock.
392 Falls back to instance defaults for *timeout* and *blocking* when ``None``.
394 :param timeout: maximum wait time in seconds, or ``None`` to use the instance default
395 :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately; ``None`` uses the instance default
397 """
398 if timeout is None:
399 timeout = self.timeout
400 if blocking is None:
401 blocking = self.blocking
402 self.acquire_write(timeout, blocking=blocking)
403 try:
404 yield
405 finally:
406 self.release()
408 def release(self, *, force: bool = False) -> None:
409 """
410 Release one level of the current lock.
412 When the lock level reaches zero the underlying SQLite transaction is rolled back, releasing the database lock.
414 :param force: if ``True``, release the lock completely regardless of the current lock level
416 :raises RuntimeError: if no lock is currently held and *force* is ``False``
418 """
419 with _fork_transition():
420 _ensure_current_process()
421 if self._inherited: # pragma: needs fork
422 return
423 self._raise_if_acquiring("release")
424 self._release(force=force, close=False)
426 def close(self) -> None:
427 """
428 Release the lock (if held) and close the underlying SQLite connection.
430 After calling this method, the lock instance is no longer usable.
432 """
433 with _fork_transition():
434 _ensure_current_process()
435 if self._inherited: # pragma: needs fork
436 return
437 self._raise_if_acquiring("close")
438 self._release(force=True, close=True)
440 def _release(self, *, force: bool, close: bool) -> None:
441 with self._transaction_lock, self._internal_lock:
442 if self._lock_level == 0:
443 if force and self._con is None:
444 if close:
445 self._closed = True
446 return
447 if not force:
448 msg = f"Cannot release a lock on {self.lock_file} (lock id: {id(self)}) that is not held"
449 raise RuntimeError(msg)
450 if not force and self._lock_level > 1:
451 self._lock_level -= 1
452 return
453 try:
454 self._finish_connection()
455 except sqlite3.Error:
456 if self._connection_transaction_released:
457 self._clear_lock_state()
458 raise
459 self._clear_lock_state()
460 if close:
461 self._closed = True
463 def _clear_lock_state(self) -> None:
464 self._lock_level = 0
465 self._current_mode = None
466 self._write_thread_id = None
468 def __del__(self) -> None:
469 if _GETPID() == getattr(self, "_creator_pid", None) and (connection := getattr(self, "_con", None)) is not None:
470 with suppress(sqlite3.Error, RuntimeError):
471 connection.close()
473 def _reset_after_fork_in_child(self) -> None: # pragma: forked child
474 if self._con is not None:
475 _FORKED_DATABASES.poison_after_fork(self._canonical_path, self._connection_identity)
476 self._con = None
477 self._connection_transaction_released = True
478 self._connection_identity = None
479 self._transaction_lock = threading.Lock()
480 self._internal_lock = threading.Lock()
481 self._clear_lock_state()
482 self._acquisition_thread_ids = set()
483 self._fork_invalidated = True
485 @property
486 def _inherited(self) -> bool:
487 return self._fork_invalidated or _GETPID() != self._creator_pid
489 def _raise_if_unusable(self) -> None:
490 _ensure_current_process()
491 if self._inherited: # pragma: needs fork
492 msg = f"ReadWriteLock on {self.lock_file} was invalidated by fork(); construct a new instance"
493 raise RuntimeError(msg)
494 if self._closed:
495 msg = "Cannot operate on a closed database."
496 raise sqlite3.ProgrammingError(msg)
498 def _acquire(self, mode: Literal["read", "write"], timeout: float, *, blocking: bool) -> AcquireReturnProxy:
499 with _fork_transition():
500 self._raise_if_unusable()
501 operation_pid = _GETPID()
502 thread_id = threading.get_ident()
503 with self._internal_lock:
504 if self._lock_level > 0:
505 return self._validate_reentrant(mode)
506 if thread_id in self._acquisition_thread_ids: # pragma: no cover - exercised in an audit callback
507 msg = f"Cannot acquire ReadWriteLock on {self.lock_file} while acquisition is active in this thread"
508 raise RuntimeError(msg)
509 self._acquisition_thread_ids.add(thread_id)
510 try:
511 start_time = time.perf_counter()
512 self._acquire_transaction_lock(blocking=blocking, timeout=timeout)
513 try:
514 self._raise_if_unusable()
515 return self._do_acquire_inner(
516 mode,
517 timeout,
518 blocking=blocking,
519 operation_pid=operation_pid,
520 start_time=start_time,
521 )
522 finally:
523 self._transaction_lock.release()
524 finally:
525 with self._internal_lock:
526 self._acquisition_thread_ids.discard(thread_id)
528 def _do_acquire_inner(
529 self,
530 mode: Literal["read", "write"],
531 timeout: float,
532 *,
533 blocking: bool,
534 operation_pid: int,
535 start_time: float,
536 ) -> AcquireReturnProxy:
537 # Double-check: another thread may have acquired the lock while we waited on _transaction_lock.
538 with self._internal_lock:
539 if self._lock_level > 0:
540 return self._validate_reentrant(mode)
541 if self._con is not None:
542 self._finish_connection()
543 try:
544 self._open_for_acquisition(
545 timeout,
546 blocking=blocking,
547 operation_pid=operation_pid,
548 start_time=start_time,
549 )
550 self._configure_and_begin(
551 mode,
552 timeout,
553 blocking=blocking,
554 operation=(operation_pid, start_time),
555 )
556 self._raise_if_process_changed(operation_pid)
557 except BaseException as error:
558 acquisition_error: BaseException
559 if isinstance(error, sqlite3.OperationalError) and "database is locked" in str(error):
560 acquisition_error = Timeout(self.lock_file)
561 else:
562 acquisition_error = error
563 try:
564 self._finish_connection()
565 except sqlite3.Error as cleanup_error:
566 _raise_chained_errors(acquisition_error, cleanup_error)
567 if acquisition_error is not error:
568 raise acquisition_error from None
569 raise
570 with self._internal_lock:
571 self._raise_if_process_changed(operation_pid)
572 self._current_mode = mode
573 self._lock_level = 1
574 if mode == "write":
575 self._write_thread_id = threading.get_ident()
576 return AcquireReturnProxy(lock=self)
578 def _open_for_acquisition(self, timeout: float, *, blocking: bool, operation_pid: int, start_time: float) -> None:
579 with _sqlite_transition():
580 sqlite_timeout = (
581 timeout_for_sqlite(
582 timeout,
583 blocking=blocking,
584 already_waited=time.perf_counter() - start_time,
585 )
586 / 1000
587 )
588 connection = self._open_connection(sqlite_timeout=sqlite_timeout)
589 self._con, self._connection_transaction_released, self._connection_identity = (
590 connection,
591 False,
592 _FORKED_DATABASES.identity(self._canonical_path),
593 )
594 self._raise_if_process_changed(operation_pid)
596 def _configure_and_begin(
597 self,
598 mode: Literal["read", "write"],
599 timeout: float,
600 *,
601 blocking: bool,
602 operation: tuple[int, float],
603 ) -> None:
604 with _sqlite_transition():
605 operation_pid, start_time = operation
606 connection = cast("_ForkSafeConnection", self._con)
607 waited = time.perf_counter() - start_time
608 timeout_ms = timeout_for_sqlite(timeout, blocking=blocking, already_waited=waited)
609 self._raise_if_process_changed(operation_pid)
610 connection.executescript(f"PRAGMA busy_timeout={timeout_ms}; PRAGMA journal_mode=MEMORY;").close()
611 # Use legacy journal mode (not WAL) because WAL does not block readers while a concurrent EXCLUSIVE
612 # write transaction is active, which makes read-write locking impossible without modifying table data.
613 # MEMORY is safe here since no writes happen, so a crash cannot corrupt the DB.
614 # See https://sqlite.org/lang_transaction.html#deferred_immediate_and_exclusive_transactions
615 #
616 # Recompute the remaining timeout after the blocking journal_mode pragma.
617 waited = time.perf_counter() - start_time
618 recomputed = timeout_for_sqlite(timeout, blocking=blocking, already_waited=waited)
619 self._raise_if_process_changed(operation_pid)
620 statements = f"PRAGMA busy_timeout={recomputed}; " if recomputed != timeout_ms else ""
621 statements += "BEGIN EXCLUSIVE TRANSACTION;" if mode == "write" else "BEGIN TRANSACTION;"
622 if mode == "read":
623 # SQLite takes the SHARED lock only when a statement reads; BEGIN alone stays deferred.
624 # https://www.sqlite.org/lockingv3.html#transaction_control
625 statements += " SELECT name FROM sqlite_schema LIMIT 1;"
626 connection.executescript(statements).close()
628 def _open_connection(self, *, sqlite_timeout: float) -> _ForkSafeConnection:
629 with _sqlite_transition():
630 creator_pid = _GETPID()
631 functions = _CONNECTION_ESCROW.functions()
632 if _GETPID() != creator_pid: # pragma: forked child
633 msg = "SQLite connection construction cannot continue after fork"
634 raise RuntimeError(msg)
635 connection = _connect(
636 os.fspath(self._canonical_path),
637 factory=_ForkSafeConnection,
638 timeout=sqlite_timeout,
639 )
640 if functions is not None: # pragma: <3.12 cover # pragma: needs fork
641 connection.acquire_escrow(functions)
642 if _GETPID() != creator_pid: # pragma: forked child
643 _FORKED_DATABASES.poison_after_fork(
644 self._canonical_path,
645 _FORKED_DATABASES.identity(self._canonical_path),
646 )
647 msg = "SQLite connection construction cannot continue after fork"
648 raise RuntimeError(msg)
649 return connection
651 def _finish_connection(self) -> None:
652 with _sqlite_transition():
653 if (connection := self._con) is None:
654 return
655 rollback_error: sqlite3.Error | None = None
656 if not self._connection_transaction_released:
657 if connection.in_transaction:
658 try:
659 connection.rollback()
660 except sqlite3.Error as error:
661 if connection.in_transaction:
662 raise
663 self._connection_transaction_released = True
664 rollback_error = error
665 else:
666 self._connection_transaction_released = True
667 else:
668 self._connection_transaction_released = True
669 try:
670 connection.close()
671 except sqlite3.Error as close_error:
672 if rollback_error is not None:
673 _raise_chained_errors(rollback_error, close_error)
674 raise
675 self._con = None
676 self._connection_transaction_released = True
677 self._connection_identity = None
678 if rollback_error is not None:
679 raise rollback_error
681 def _validate_reentrant(self, mode: Literal["read", "write"]) -> AcquireReturnProxy:
682 if self._current_mode != mode:
683 opposite = "write" if mode == "read" else "read"
684 direction = "downgrade" if mode == "read" else "upgrade"
685 msg = (
686 f"Cannot acquire {mode} lock on {self.lock_file} (lock id: {id(self)}): "
687 f"already holding a {opposite} lock ({direction} not allowed)"
688 )
689 raise RuntimeError(msg)
690 if mode == "write" and (cur := threading.get_ident()) != self._write_thread_id:
691 msg = (
692 f"Cannot acquire write lock on {self.lock_file} (lock id: {id(self)}) "
693 f"from thread {cur} while it is held by thread {self._write_thread_id}"
694 )
695 raise RuntimeError(msg)
696 self._lock_level += 1
697 return AcquireReturnProxy(lock=self)
699 def _acquire_transaction_lock(self, *, blocking: bool, timeout: float) -> None:
700 if not blocking:
701 acquired = self._transaction_lock.acquire(blocking=False)
702 elif timeout == -1:
703 acquired = self._transaction_lock.acquire(blocking=True)
704 else:
705 acquired = self._transaction_lock.acquire(blocking=True, timeout=timeout)
706 if not acquired:
707 raise Timeout(self.lock_file) from None
709 def _raise_if_acquiring(self, operation: Literal["acquire", "close", "release"]) -> None:
710 with self._internal_lock:
711 active_in_current_thread = threading.get_ident() in self._acquisition_thread_ids
712 if active_in_current_thread: # pragma: no cover - exercised in an audit callback
713 msg = f"Cannot {operation} ReadWriteLock on {self.lock_file} while acquisition is active in this thread"
714 raise RuntimeError(msg)
716 def _raise_if_process_changed(self, operation_pid: int) -> None:
717 if _GETPID() != operation_pid or self._inherited: # pragma: forked child
718 msg = f"ReadWriteLock on {self.lock_file} was invalidated by fork(); construct a new instance"
719 raise RuntimeError(msg)
722def _connect(database: str, *, factory: type[_ForkSafeConnection], timeout: float) -> _ForkSafeConnection:
723 _FORKED_DATABASES.note_sqlite_use()
724 return sqlite3.connect(
725 database,
726 check_same_thread=False,
727 factory=factory,
728 cached_statements=0,
729 timeout=timeout,
730 )
733@contextmanager
734def _sqlite_transition() -> Generator[None]:
735 _SQLITE_TRANSITION_CONTEXT.depth += 1
736 try:
737 yield
738 finally:
739 _SQLITE_TRANSITION_CONTEXT.depth -= 1
742def _abort_forked_sqlite_transition() -> None: # pragma: forked child
743 if _SQLITE_TRANSITION_CONTEXT.depth:
744 os._exit(_UNSAFE_FORK_EXIT_STATUS) # inherited SQLite handles cannot be used or closed safely
747def _track_sqlite_use(event: str, _args: tuple[object, ...]) -> None: # audit payloads are heterogeneous
748 if event == "sqlite3.connect":
749 _FORKED_DATABASES.note_sqlite_use()
752def timeout_for_sqlite(timeout: float, *, blocking: bool, already_waited: float) -> int:
753 if blocking is False:
754 return 0
756 if timeout == -1:
757 return _MAX_SQLITE_TIMEOUT_MS
759 if timeout < 0:
760 msg = "timeout must be a non-negative number or -1"
761 raise ValueError(msg)
763 timeout_ms = int((max(timeout - already_waited, 0) if timeout > 0 else timeout) * 1000)
764 if timeout_ms > _MAX_SQLITE_TIMEOUT_MS or timeout_ms < 0:
765 _LOGGER.warning("timeout %s is too large for SQLite, using %s ms instead", timeout, _MAX_SQLITE_TIMEOUT_MS)
766 return _MAX_SQLITE_TIMEOUT_MS
767 return timeout_ms
770_register_fork_object(_CONNECTION_ESCROW)
771_register_fork_object(_FORKED_DATABASES)
772_register_fork_class(ReadWriteLock)
773if _IS_PYPY:
774 sys.addaudithook(_track_sqlite_use) # pragma: pypy cover
775if hasattr(os, "register_at_fork"): # pragma: needs fork
776 os.register_at_fork(after_in_child=_abort_forked_sqlite_transition)
778__all__ = [
779 "ReadWriteLock",
780 "timeout_for_sqlite",
781]