🐛 fix(async): honor cancellation and per-thread gates

docs/changelog/844.bugfix.rst

@@ -0,0 +1,4 @@
+Canceling an :class:`~filelock.AsyncReadWriteLock` or :class:`~filelock.AsyncSoftReadWriteLock` acquire that waits for
+another process now stops the wait and raises the cancellation. It used to wait until that process released, or raise
+:class:`~filelock.Timeout`. A ``thread_local=True`` :class:`~filelock.AsyncFileLock` that threads with their own event
+loops share no longer deadlocks when one thread waits for the lock while another holds it.

src/filelock/_api.py

@@ -59,6 +59,7 @@
 
 if TYPE_CHECKING:
     from collections.abc import Generator, Mapping
+    from threading import Lock
     from types import FrameType, TracebackType
     from typing import Protocol
 
@@ -1719,6 +1720,24 @@ def _resolve_read_write_timeout(timeout: float, *, blocking: bool) -> float:
     return timeout
 
 
+def _acquire_thread_lock(
+    lock: Lock, *, blocking: bool, timeout: float, poll_interval: float, cancel_check: Callable[[], bool] | None
+) -> bool:
+    """Take a read-write lock's in-process lock; with *cancel_check*, ask it every *poll_interval* while waiting."""
+    if not blocking:
+        return lock.acquire(blocking=False)
+    if cancel_check is None:
+        # Lock.acquire raises OverflowError past TIMEOUT_MAX, which is 49.7 days on Windows.
+        return lock.acquire() if timeout == -1 else lock.acquire(timeout=min(timeout, TIMEOUT_MAX))
+    deadline = None if timeout == -1 else time.perf_counter() + timeout
+    while not lock.acquire(
+        timeout=poll_interval if deadline is None else min(poll_interval, max(deadline - time.perf_counter(), 0))
+    ):
+        if cancel_check() or (deadline is not None and time.perf_counter() >= deadline):
+            return False
+    return True
+
+
 def _resolve_timeout(timeout: float, *, blocking: bool) -> float:
     """Validate an acquire ``timeout``; every negative value waits without a limit."""
     if math.isnan(timeout := _seconds("timeout", timeout)) and blocking:
@@ -2232,6 +2251,7 @@ def _audit_fork_safety(  # pragma: no cover - CPython disables tracing while Pyt
     "FileLockMeta",
     "LockOptions",
     "ThreadLocalFileContext",
+    "_acquire_thread_lock",
     "_append_exception_context",
     "_canonical",
     "_check_timeout_max",

src/filelock/_async_read_write.py

@@ -2,12 +2,13 @@
 
 from __future__ import annotations
 
-__lazy_modules__ = ["asyncio", "concurrent.futures", "filelock._async", "functools"]
+__lazy_modules__ = ["asyncio", "concurrent.futures", "filelock._async", "filelock._error", "functools", "threading"]
 
 import asyncio
 import functools
 import os
 import sqlite3
+import threading
 from concurrent.futures import ThreadPoolExecutor
 from concurrent.futures import thread as executor_threads
 from contextlib import asynccontextmanager
@@ -29,10 +30,10 @@
     _task_owners_for,
     _wait_until_done,
 )
+from filelock._error import Timeout
 from filelock._read_write import ReadWriteLock
 
 if TYPE_CHECKING:
-    import threading
     from collections.abc import AsyncGenerator, Callable, MutableMapping
     from concurrent import futures
     from types import TracebackType
@@ -311,13 +312,19 @@ async def close(self) -> None:
         if self._owns_executor:
             await asyncio.to_thread(functools.partial(self._executor.shutdown, wait=True))
 
-    async def _run_acquire(self, acquire: Callable[[], AcquireReturnProxy]) -> None:
-        acquire_future = self._submit(acquire)
+    async def _run_acquire(self, acquire: Callable[..., AcquireReturnProxy]) -> None:
+        abandoned = threading.Event()
+        acquire_future = self._submit(acquire, cancel_check=abandoned.is_set)
         try:
             await _wait_until_done(acquire_future)
         except asyncio.CancelledError as cancellation:
+            # Stops a backend wait on another process's hold, which would otherwise outlast the caller by as long.
+            abandoned.set()
             try:
                 await _drain_future(acquire_future)
+            except Timeout:
+                # Nothing was taken, so the caller's cancellation is the whole outcome.
+                raise cancellation from None
             except BaseException as error:  # ruff:ignore[blind-except]  # reported with the cancellation below
                 _raise_cancelled_error(cancellation, error)
             try:

src/filelock/_read_write.py

@@ -32,6 +32,7 @@
 
 from filelock._api import (
     AcquireReturnProxy,
+    _acquire_thread_lock,
     _canonical,
     _ensure_current_process,
     _fork_transition,
@@ -70,6 +71,8 @@
 
 # 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
+# How long a wait that its caller may abandon runs before it checks whether it should stop.
+_CANCEL_POLL_INTERVAL: Final[float] = 0.1
 _UNSAFE_FORK_EXIT_STATUS: Final[int] = 70
 #: The longest a contended acquisition waits for the database inside one fork transition, in seconds.
 _MAX_TRANSITION_WAIT: Final[float] = 0.05
@@ -262,6 +265,13 @@ def _reset_after_fork_in_child(self) -> None:  # pragma: forked child
         self._releases = deque()
 
 
+class _Operation(NamedTuple):
+    pid: int
+    start_time: float
+    #: Asked between bounded waits, so an abandoned acquire stops waiting; ``None`` waits in one call.
+    cancel_check: Callable[[], bool] | None
+
+
 class _Probe(NamedTuple):
     identity: _DatabaseIdentity
     fd: int
@@ -643,11 +653,17 @@ def close(self) -> None:
             self._release(force=True, close=True)
 
     def _acquire_level_of(
-        self, holder: int, mode: Literal["read", "write"], timeout: float, *, blocking: bool
+        self,
+        holder: int,
+        mode: Literal["read", "write"],
+        timeout: float,
+        *,
+        blocking: bool,
+        cancel_check: Callable[[], bool],
     ) -> AcquireReturnProxy:
         # Records the level under *holder* instead of the calling thread: an async wrapper's executor thread does not
-        # identify the tasks it holds for.
-        return self._acquire(mode, timeout, blocking=blocking, holder=holder)
+        # identify the tasks it holds for. Gives up with Timeout once *cancel_check* is true.
+        return self._acquire(mode, timeout, blocking=blocking, holder=holder, cancel_check=cancel_check)
 
     def _release_level_of(self, thread_id: int) -> None:
         with _fork_transition():
@@ -763,7 +779,13 @@ def _raise_if_unusable(self) -> None:
             raise sqlite3.ProgrammingError(msg)
 
     def _acquire(
-        self, mode: Literal["read", "write"], timeout: float, *, blocking: bool, holder: int | None = None
+        self,
+        mode: Literal["read", "write"],
+        timeout: float,
+        *,
+        blocking: bool,
+        holder: int | None = None,
+        cancel_check: Callable[[], bool] | None = None,
     ) -> AcquireReturnProxy:
         timeout = _resolve_read_write_timeout(timeout, blocking=blocking)
         start_time: Final = time.monotonic()
@@ -781,17 +803,27 @@ def _acquire(
             )
             try:
                 return self._attempt_acquire(
-                    mode, remaining if final else _MAX_TRANSITION_WAIT, blocking=blocking, holder=holder
+                    mode,
+                    remaining if final else _MAX_TRANSITION_WAIT,
+                    blocking=blocking,
+                    holder=holder,
+                    cancel_check=cancel_check,
                 )
             except Timeout:
-                if final:
+                if final or (cancel_check is not None and cancel_check()):
                     raise
             # Only a runtime with fork handlers waits in more than one attempt.
             if timeout != -1:  # pragma: needs register-at-fork
                 remaining = max(timeout - (time.monotonic() - start_time), 0)
 
     def _attempt_acquire(
-        self, mode: Literal["read", "write"], timeout: float, *, blocking: bool, holder: int | None = None
+        self,
+        mode: Literal["read", "write"],
+        timeout: float,
+        *,
+        blocking: bool,
+        holder: int | None = None,
+        cancel_check: Callable[[], bool] | None = None,
     ) -> AcquireReturnProxy:
         with _fork_transition():
             self._raise_if_unusable()
@@ -807,11 +839,15 @@ def _attempt_acquire(
                 self._acquisition_thread_ids.add(thread_id)
             try:
                 start_time = time.perf_counter()
-                self._acquire_transaction_lock(blocking=blocking, timeout=timeout)
+                self._acquire_transaction_lock(blocking=blocking, timeout=timeout, cancel_check=cancel_check)
                 try:
                     self._raise_if_unusable()
                     return self._do_acquire_inner(
-                        mode, timeout, holder, blocking=blocking, operation=(operation_pid, start_time)
+                        mode,
+                        timeout,
+                        holder,
+                        blocking=blocking,
+                        operation=_Operation(operation_pid, start_time, cancel_check),
                     )
                 finally:
                     self._transaction_lock.release()
@@ -826,9 +862,9 @@ def _do_acquire_inner(
         holder: int,
         *,
         blocking: bool,
-        operation: tuple[int, float],
+        operation: _Operation,
     ) -> AcquireReturnProxy:
-        operation_pid, start_time = operation
+        operation_pid, start_time, _ = operation
         # Double-check: another thread may have acquired the lock while we waited on _transaction_lock.
         with self._internal_lock:
             if self._holds_level():
@@ -891,34 +927,53 @@ def _configure_and_begin(
         timeout: float,
         *,
         blocking: bool,
-        operation: tuple[int, float],
+        operation: _Operation,
     ) -> 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()
+            self._raise_if_process_changed(operation.pid)
             # 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;"
+            self._execute_waiting(
+                connection, "PRAGMA journal_mode=MEMORY;", timeout, blocking=blocking, operation=operation
+            )
+            self._raise_if_process_changed(operation.pid)
+            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()
+            self._execute_waiting(connection, statements, timeout, blocking=blocking, operation=operation)
             self._raise_if_database_replaced(connection)
 
+    @staticmethod
+    def _execute_waiting(
+        connection: _ForkSafeConnection, statements: str, timeout: float, *, blocking: bool, operation: _Operation
+    ) -> None:
+        # SQLite's busy wait cannot be interrupted. An acquire its caller may abandon waits in bounded slices and asks
+        # between them whether to stop.
+        slice_ms = int(_CANCEL_POLL_INTERVAL * 1000)
+        while True:
+            busy_ms = timeout_for_sqlite(
+                timeout, blocking=blocking, already_waited=time.perf_counter() - operation.start_time
+            )
+            cancel_check = operation.cancel_check if busy_ms > slice_ms else None
+            try:
+                connection.executescript(
+                    f"PRAGMA busy_timeout={busy_ms if cancel_check is None else slice_ms}; {statements}"
+                ).close()
+            except sqlite3.OperationalError as error:
+                if cancel_check is None or "database is locked" not in str(error) or cancel_check():
+                    raise
+                # A read's BEGIN succeeds before its SELECT waits, so the retry has to start a fresh transaction.
+                if connection.in_transaction:
+                    connection.rollback()
+            else:
+                return
+
     def _raise_if_database_replaced(self, connection: _ForkSafeConnection) -> None:
         # Windows pins the name: SQLite opens the database without delete sharing, so it cannot be unlinked or renamed.
         if (identity := connection.database_identity) is None:  # pragma: win32 cover
@@ -1015,15 +1070,16 @@ def _validate_reentrant(self, mode: Literal["read", "write"], cur: int) -> Acqui
         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:
-            # Lock.acquire raises OverflowError past TIMEOUT_MAX, which is 49.7 days on Windows.
-            acquired = self._transaction_lock.acquire(blocking=True, timeout=min(timeout, threading.TIMEOUT_MAX))
-        if not acquired:
+    def _acquire_transaction_lock(
+        self, *, blocking: bool, timeout: float, cancel_check: Callable[[], bool] | None
+    ) -> None:
+        if not _acquire_thread_lock(
+            self._transaction_lock,
+            blocking=blocking,
+            timeout=timeout,
+            poll_interval=_CANCEL_POLL_INTERVAL,
+            cancel_check=cancel_check,
+        ):
             raise Timeout(self.lock_file) from None
 
     def _raise_if_acquiring(self, operation: Literal["acquire", "close", "release"]) -> None:

src/filelock/_soft_rw/_async.py

@@ -2,11 +2,12 @@
 
 from __future__ import annotations
 
-__lazy_modules__ = ["asyncio", "filelock._async", "functools"]
+__lazy_modules__ = ["asyncio", "filelock._async", "filelock._error", "functools", "threading"]
 
 import asyncio
 import functools
 import os
+import threading
 from contextlib import asynccontextmanager
 from typing import TYPE_CHECKING, Final, Literal, ParamSpec, TypeVar
 
@@ -20,6 +21,7 @@
     _task_owners_for,
     _wait_until_done,
 )
+from filelock._error import Timeout
 from filelock._soft_rw._sync import SoftReadWriteLock, _lost_hold_error
 
 if TYPE_CHECKING:
@@ -315,17 +317,22 @@ def _leave_backend(self) -> None:
         # Only the tasks' level: a sync user of the same singleton keeps its own.
         self._lock._release_level_of(_TASKS_THREAD_ID)  # ruff:ignore[private-member-access]  # the wrapper owns this backend hold
 
-    async def _run_acquire(self, acquire: Callable[[], AcquireReturnProxy]) -> None:
-        # run_in_executor cannot recall work the pool already started, so canceling the caller does not stop the sync
-        # acquire: it still commits its claim, sets the hold, and starts the heartbeat, which keeps the record fresh
-        # forever so no peer on any host can evict it as stale. Wait the submitted call out and hand the claim back,
-        # the way AsyncReadWriteLock does.
-        acquire_future = self._submit(acquire)
+    async def _run_acquire(self, acquire: Callable[..., AcquireReturnProxy]) -> None:
+        # run_in_executor cannot recall work the pool already started. A sync acquire that won the lock before it saw
+        # the cancellation has published its claim and started the heartbeat, which keeps the record fresh so no peer
+        # evicts it. Wait the call out and hand the claim back, as AsyncReadWriteLock does.
+        abandoned = threading.Event()
+        acquire_future = self._submit(acquire, cancel_check=abandoned.is_set)
         try:
             await _wait_until_done(acquire_future)
         except asyncio.CancelledError as cancellation:
+            # Stops a backend wait on another process's hold, which would otherwise outlast the caller by as long.
+            abandoned.set()
             try:
                 await _drain_future(acquire_future)
+            except Timeout:
+                # Nothing was taken, so the caller's cancellation is the whole outcome.
+                raise cancellation from None
             except BaseException as error:  # ruff:ignore[blind-except]  # reported with the cancellation below
                 _raise_cancelled_error(cancellation, error)
             try: