From 7551eb921f4b81a0283df456b714f94145a9646d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bern=C3=A1t=20G=C3=A1bor?= Date: Thu, 15 Oct 2026 18:40:13 -0700 Subject: [PATCH 01/11] fix(async-rw): stop a backend wait on cancellation AsyncReadWriteLock and AsyncSoftReadWriteLock handed the whole contended wait to one executor call and, when the caller was cancelled, waited that call out before rolling it back. While another process held the lock, an asyncio.timeout() or task.cancel() took effect only once that process let go (with the default timeout=-1, up to SQLite's 23-day busy cap), and with a finite lock timeout the cancellation was replaced by filelock's Timeout, so a task cancelled at shutdown kept retrying. The wrapper now tells the running backend call to stop. ReadWriteLock waits for SQLite and for its in-process lock in short slices on the same connection and gives up between them; SoftReadWriteLock checks between polls. A wait that gave up this way ends in the caller's cancellation. --- src/filelock/_api.py | 20 +++ src/filelock/_async_read_write.py | 15 +- src/filelock/_read_write.py | 122 +++++++++++----- src/filelock/_soft_rw/_async.py | 22 ++- src/filelock/_soft_rw/_sync.py | 42 +++--- tests/soft_rw/test_soft_rw_async.py | 5 +- tests/test_async_read_write_cancel_wait.py | 151 ++++++++++++++++++++ tests/test_async_read_write_cancellation.py | 24 ++++ 8 files changed, 336 insertions(+), 65 deletions(-) create mode 100644 tests/test_async_read_write_cancel_wait.py diff --git a/src/filelock/_api.py b/src/filelock/_api.py index ddc37774..d0b77584 100644 --- a/src/filelock/_api.py +++ b/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", diff --git a/src/filelock/_async_read_write.py b/src/filelock/_async_read_write.py index 555cb808..59204573 100644 --- a/src/filelock/_async_read_write.py +++ b/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: diff --git a/src/filelock/_read_write.py b/src/filelock/_read_write.py index f498d5db..7e066d4d 100644 --- a/src/filelock/_read_write.py +++ b/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,7 +803,11 @@ 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: @@ -791,7 +817,13 @@ def _acquire( 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, so 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: diff --git a/src/filelock/_soft_rw/_async.py b/src/filelock/_soft_rw/_async.py index 02675fc6..a224377a 100644 --- a/src/filelock/_soft_rw/_async.py +++ b/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,23 @@ 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 got the lock before it saw + # the cancellation has committed its claim, set the hold, and started 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. + 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: diff --git a/src/filelock/_soft_rw/_sync.py b/src/filelock/_soft_rw/_sync.py index 48d35831..caa74298 100644 --- a/src/filelock/_soft_rw/_sync.py +++ b/src/filelock/_soft_rw/_sync.py @@ -22,6 +22,7 @@ from filelock._api import ( AcquireReturnProxy, + _acquire_thread_lock, _canonical, _check_timeout_max, _ensure_current_process, @@ -420,10 +421,12 @@ def release(self, *, force: bool = False) -> None: return self._release(threading.get_ident(), force=force) - def _acquire_level_of(self, holder: int, mode: Mode, timeout: float, *, blocking: bool) -> AcquireReturnProxy: + def _acquire_level_of( + self, holder: int, mode: Mode, 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, holder: int) -> None: self._release(holder, force=False) @@ -487,6 +490,7 @@ def _acquire( *, blocking: bool | None, holder: int | None = None, + cancel_check: Callable[[], bool] | None = None, ) -> AcquireReturnProxy: if self._creator_pid != os.getpid(): # pragma: forked child msg = f"SoftReadWriteLock on {self.lock_file} was invalidated by fork(); construct a new instance" @@ -502,18 +506,17 @@ def _acquire( if self._hold is not None: return self._validate_reentrant(mode, holder) - start = time.perf_counter() - if not blocking: - acquired = self._locks.transaction.acquire(blocking=False) - elif timeout == -1: - acquired = self._locks.transaction.acquire(blocking=True) - else: - # Lock.acquire raises OverflowError past TIMEOUT_MAX, which is 49.7 days on Windows. - acquired = self._locks.transaction.acquire(blocking=True, timeout=min(timeout, threading.TIMEOUT_MAX)) - if not acquired: + deadline = None if timeout == -1 else time.perf_counter() + timeout + if not _acquire_thread_lock( + self._locks.transaction, + blocking=blocking, + timeout=timeout, + poll_interval=self.poll_interval, + cancel_check=cancel_check, + ): raise Timeout(self.lock_file) from None try: - return self._do_acquire_inner(mode, timeout, start, holder, blocking=blocking) + return self._do_acquire_inner(mode, deadline, holder, blocking=blocking, cancel_check=cancel_check) finally: self._locks.transaction.release() @@ -523,16 +526,15 @@ def _closed_error(self) -> RuntimeError: def _do_acquire_inner( self, mode: Mode, - effective_timeout: float, - start: float, + deadline: float | None, holder: int, *, blocking: bool, + cancel_check: Callable[[], bool] | None, ) -> AcquireReturnProxy: with self._locks.internal: if self._hold is not None: return self._validate_reentrant(mode, holder) - deadline = None if effective_timeout == -1 else start + effective_timeout participant = Participant( self._files, self._root, @@ -544,7 +546,7 @@ def _do_acquire_inner( ) participant.publish() try: - self._wait_for(participant.advance, deadline=deadline, blocking=blocking) + self._wait_for(participant.advance, deadline=deadline, blocking=blocking, cancel_check=cancel_check) except BaseException: # A contender that gives up must not stay named in the snapshot: a writer left there would block every # reader until a peer waits out the stale threshold. Its own token is all leave() touches, so it cannot @@ -622,15 +624,17 @@ def _wait_for( *, deadline: float | None, blocking: bool, + cancel_check: Callable[[], bool] | None, ) -> None: while not predicate(): # The flag is read without the lock here only to stop waiting early; publishing the hold re-checks it. if self._closed: raise self._closed_error() now = time.perf_counter() - if not blocking: + if not blocking or (deadline is not None and now >= deadline): raise Timeout(self.lock_file) - if deadline is not None and now >= deadline: + # An abandoned acquire gives up like a timed-out one, so the caller's rollback withdraws its claim. + if cancel_check is not None and cancel_check(): raise Timeout(self.lock_file) sleep_for = self.poll_interval if deadline is not None: diff --git a/tests/soft_rw/test_soft_rw_async.py b/tests/soft_rw/test_soft_rw_async.py index 34add7f9..1591a026 100644 --- a/tests/soft_rw/test_soft_rw_async.py +++ b/tests/soft_rw/test_soft_rw_async.py @@ -395,14 +395,15 @@ async def test_async_acquire_cancellation_hands_the_claim_back( @pytest.mark.asyncio @XFAIL_WITHOUT_COROUTINE_CANCELLATION async def test_async_acquire_cancellation_surfaces_a_failed_acquire(tmp_path: Path, mocker: MockerFixture) -> None: - acquire_error = Timeout(str(tmp_path / "foo.lock")) + # A Timeout here means the acquire gave up on the cancellation, so a real failure has to be another error. + acquire_error = OSError("acquire failed") gate = _Gate(mocker, "_acquire_level_of", fail_with=acquire_error) lock = _make(tmp_path) task = asyncio.create_task(lock.acquire_write(timeout=5)) await gate.started.wait() task.cancel("cancel acquire") gate.resume() - with pytest.raises(Timeout) as info: + with pytest.raises(OSError, match="acquire failed") as info: await task assert info.value is acquire_error diff --git a/tests/test_async_read_write_cancel_wait.py b/tests/test_async_read_write_cancel_wait.py new file mode 100644 index 00000000..d7a99bf1 --- /dev/null +++ b/tests/test_async_read_write_cancel_wait.py @@ -0,0 +1,151 @@ +from __future__ import annotations + +import asyncio +import threading +import time +from typing import TYPE_CHECKING, Final, Literal + +import pytest + +pytest.importorskip("sqlite3") + +from collections.abc import Callable + +from filelock import AsyncReadWriteLock, ReadWriteLock, Timeout +from filelock._soft_rw import AsyncSoftReadWriteLock, SoftReadWriteLock +from tests.capability_marks import XFAIL_WITHOUT_COROUTINE_CANCELLATION + +if TYPE_CHECKING: + from collections.abc import Generator + from pathlib import Path + + from pytest_mock import MockerFixture + +# The other process's hold ends this long after the wait starts, so a wait that ignores its deadline ends then. +_HOLD: Final[float] = 3 +# The caller's own deadline, far below _HOLD. +_DEADLINE: Final[float] = 0.2 + +pytestmark = XFAIL_WITHOUT_COROUTINE_CANCELLATION + + +@pytest.fixture(autouse=True) +def _clear_singleton_caches() -> Generator[None]: + ReadWriteLock._instances.clear() + SoftReadWriteLock._instances.clear() + yield + for lock in filter(None, (ref() for ref in list(ReadWriteLock._instances.valuerefs()))): + lock.close() + for soft in filter(None, (ref() for ref in list(SoftReadWriteLock._instances.valuerefs()))): + soft.close() + ReadWriteLock._instances.clear() + SoftReadWriteLock._instances.clear() + + +def _sqlite(path: Path, *, is_singleton: bool) -> tuple[ReadWriteLock, AsyncReadWriteLock]: + return ReadWriteLock(path / "lock.db", is_singleton=is_singleton), AsyncReadWriteLock(path / "lock.db") + + +def _soft(path: Path, *, is_singleton: bool) -> tuple[SoftReadWriteLock, AsyncSoftReadWriteLock]: + options: Final = {"heartbeat_interval": 0.1, "stale_threshold": 0.5, "poll_interval": 0.02} + return SoftReadWriteLock(path / "lock", is_singleton=is_singleton, **options), AsyncSoftReadWriteLock( + path / "lock", **options + ) + + +_Make = Callable[..., tuple[ReadWriteLock | SoftReadWriteLock, AsyncReadWriteLock | AsyncSoftReadWriteLock]] +_FAMILIES: Final = pytest.mark.parametrize("make", [pytest.param(_sqlite, id="sqlite"), pytest.param(_soft, id="soft")]) +_MODES: Final = pytest.mark.parametrize("mode", [pytest.param("read", id="read"), pytest.param("write", id="write")]) + + +async def _acquire_within_deadline( + lock: AsyncReadWriteLock | AsyncSoftReadWriteLock, mode: Literal["read", "write"], timeout: float +) -> float: + start = time.perf_counter() + with pytest.raises(TimeoutError) as info: + async with asyncio.timeout(_DEADLINE): + await (lock.acquire_read if mode == "read" else lock.acquire_write)(timeout) + assert not isinstance(info.value, Timeout) + return time.perf_counter() - start + + +@_FAMILIES +@_MODES +@pytest.mark.parametrize( + "timeout", [pytest.param(-1, id="no-lock-timeout"), pytest.param(_HOLD * 2, id="lock-timeout")] +) +@pytest.mark.asyncio +async def test_async_acquire_behind_another_process_honors_the_callers_deadline( + tmp_path: Path, make: _Make, mode: Literal["read", "write"], timeout: float +) -> None: + holder, lock = make(tmp_path, is_singleton=False) + holder.acquire_write() + ending = asyncio.get_running_loop().call_later(_HOLD, holder.release) + + elapsed = await _acquire_within_deadline(lock, mode, timeout) + + assert elapsed < _HOLD + ending.cancel() + holder.release() + await (lock.acquire_read if mode == "read" else lock.acquire_write)(_HOLD * 2) + await lock.release() + await lock.close() + + +@_FAMILIES +@pytest.mark.asyncio +async def test_async_acquire_behind_a_sync_acquire_of_the_same_instance_honors_the_callers_deadline( + tmp_path: Path, make: _Make +) -> None: + holder, lock = make(tmp_path, is_singleton=False) + shared, _ = make(tmp_path, is_singleton=True) + holder.acquire_write() + + # A thread waiting in the shared instance's acquire keeps that instance busy for the task too. + def write_once() -> None: + with shared.write_lock(timeout=_HOLD * 2): + pass + + waiting = threading.Thread(target=write_once) + waiting.start() + ending = asyncio.get_running_loop().call_later(_HOLD, holder.release) + + elapsed = await _acquire_within_deadline(lock, "write", -1) + + assert elapsed < _HOLD + ending.cancel() + holder.release() + await asyncio.to_thread(waiting.join, _HOLD * 2) + await lock.acquire_write(timeout=_HOLD) + await lock.release() + await lock.close() + + +@_FAMILIES +@pytest.mark.asyncio +async def test_async_acquire_behind_a_sync_acquire_of_the_same_instance_times_out_at_its_own_deadline( + tmp_path: Path, make: _Make, mocker: MockerFixture +) -> None: + # With one thread alive the wait is not sliced, so the thread lock is polled across several intervals. + mocker.patch.object(threading, "active_count", return_value=1) + holder, lock = make(tmp_path, is_singleton=False) + shared, _ = make(tmp_path, is_singleton=True) + holder.acquire_write() + + def write_once() -> None: + with shared.write_lock(timeout=_HOLD * 2): + pass + + waiting = threading.Thread(target=write_once) + waiting.start() + ending = asyncio.get_running_loop().call_later(_HOLD, holder.release) + start = time.perf_counter() + + with pytest.raises(Timeout): + await lock.acquire_write(_DEADLINE) + + assert time.perf_counter() - start < _HOLD + ending.cancel() + holder.release() + await asyncio.to_thread(waiting.join, _HOLD * 2) + await lock.close() diff --git a/tests/test_async_read_write_cancellation.py b/tests/test_async_read_write_cancellation.py index c0e17aca..f35c90d9 100644 --- a/tests/test_async_read_write_cancellation.py +++ b/tests/test_async_read_write_cancellation.py @@ -362,6 +362,30 @@ async def fail_in_context() -> None: assert_read_write_lock_state(lock_file, probe_mode, available=True) +@pytest.mark.asyncio +async def test_read_retried_after_its_select_waited_starts_a_fresh_transaction( + lock_file: str, mocker: MockerFixture +) -> None: + busy_once = True + + def executescript(real_connection: sqlite3.Connection, statement: str) -> sqlite3.Cursor: + nonlocal busy_once + if busy_once and "SELECT name" in statement: + # A writer can take the lock between the deferred BEGIN and the SELECT that needs the shared lock. + busy_once = False + real_connection.execute("BEGIN TRANSACTION") + msg = "database is locked" + raise sqlite3.OperationalError(msg) + return real_connection.executescript(statement) + + _patch_async_connection(mocker, executescript=executescript) + lock = AsyncReadWriteLock(lock_file, is_singleton=False) + + async with lock.read_lock(): + assert_read_write_lock_state(lock_file, "write", available=False) + await lock.close() + + def _patch_async_rollback( mocker: MockerFixture, loop: asyncio.AbstractEventLoop, From bc50355386293b3664022fd98e91945f5ace5aca Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bern=C3=A1t=20G=C3=A1bor?= Date: Thu, 15 Oct 2026 18:41:28 -0700 Subject: [PATCH 02/11] fix(async): give each thread-local context its gate BaseAsyncFileLock created one transition gate per instance, while a thread_local=True lock keeps separate lock state per thread. A thread whose own context was unlocked took the gate for its whole poll loop, and the thread holding the lock then queued on that same gate in release(), so each waited for the other: with the default timeout both hung and the lock file stayed held; with a finite timeout the waiter always ended in Timeout. The gate now lives in the lock context, so a thread-local lock gets one per thread and a shared lock keeps one per instance. --- src/filelock/asyncio.py | 16 +++++++--- ...est_async_filelock_transition_admission.py | 30 ++++++++++++++++++- 2 files changed, 41 insertions(+), 5 deletions(-) diff --git a/src/filelock/asyncio.py b/src/filelock/asyncio.py index 92738894..b25c791a 100644 --- a/src/filelock/asyncio.py +++ b/src/filelock/asyncio.py @@ -22,7 +22,7 @@ import sys import time from concurrent import futures -from dataclasses import dataclass, fields +from dataclasses import dataclass, field, fields from inspect import iscoroutinefunction from threading import RLock from typing import TYPE_CHECKING, Final, NoReturn, TypeVar, cast @@ -231,7 +231,6 @@ def __init__( # ruff:ignore[too-many-arguments] # public constructor: one para self._fallback_to_soft = fallback_to_soft self._preserve_lock_file = preserve_lock_file # already validated by the metaclass self._on_acquired = on_acquired # already validated by the metaclass - self._transition_gate: Final[_AsyncTransitionGate] = _AsyncTransitionGate() # BaseFileLock.__init__ never runs here, and the finalizer releases thread-local holds through these. self._transition_lock = RLock() self._thread_holds = {} @@ -366,7 +365,7 @@ async def _acquire_with_admission( poll_interval: float, start_time: float, ) -> AsyncAcquireReturnProxy: - async with self._transition_gate.hold_for_acquire( + async with self._context.transition_gate.hold_for_acquire( blocking=blocking, cancel_check=cancel_check, deadline=None if timeout < 0 else start_time + timeout, @@ -525,7 +524,7 @@ async def release(self, force: bool = False) -> None: # ty: ignore[invalid-meth """ if self._creator_pid != os.getpid() or not self.is_locked: return - async with self._transition_gate.hold(): + async with self._context.transition_gate.hold(): await self._release_serialized(force=force) async def _release_serialized(self, *, force: bool) -> None: @@ -813,6 +812,11 @@ def _release_without_loop(self) -> None: self._release_thread_holds(exited_only=False) +def _new_transition_gate() -> _AsyncTransitionGate: + # Called per context rather than passed as the factory, so the lazy import resolves on first use, not at import. + return _AsyncTransitionGate() + + @dataclass class AsyncFileLockContext(FileLockContext): """A dataclass which holds the context for a ``BaseAsyncFileLock`` object.""" @@ -826,6 +830,10 @@ class AsyncFileLockContext(FileLockContext): #: The loop loop: asyncio.AbstractEventLoop | None = None + #: Serializes the tasks that transition this context. An acquire keeps it while it polls, so a thread-local lock + #: needs one per thread: a thread waiting for the lock must not keep the thread that holds it from releasing. + transition_gate: _AsyncTransitionGate = field(default_factory=_new_transition_gate) + class AsyncThreadLocalFileContext(AsyncFileLockContext, ThreadLocalFileContext): """A thread local version of the ``FileLockContext`` class.""" diff --git a/tests/test_async_filelock_transition_admission.py b/tests/test_async_filelock_transition_admission.py index 47cabdfb..f7ae1135 100644 --- a/tests/test_async_filelock_transition_admission.py +++ b/tests/test_async_filelock_transition_admission.py @@ -7,7 +7,7 @@ import pytest -from filelock import AsyncFileLock, AsyncSoftFileLock, Timeout +from filelock import AsyncFileLock, AsyncSoftFileLock, BaseAsyncFileLock, Timeout from filelock._async import _AsyncTransitionGate from tests.async_filelock_cancellation_helpers import ( assert_cancellation_message, @@ -187,5 +187,33 @@ async def acquire() -> None: # pragma: win32 no cover return asyncio.create_task(acquire()), started +@pytest.mark.parametrize( + "lock_type", [pytest.param(AsyncFileLock, id="flock"), pytest.param(AsyncSoftFileLock, id="soft")] +) +@pytest.mark.asyncio +async def test_thread_local_lock_releases_while_another_thread_waits_for_it( + tmp_path: Path, lock_type: type[BaseAsyncFileLock] +) -> None: + lock = lock_type(tmp_path / "a", thread_local=True, run_in_executor=False) + await lock.acquire() + polling = threading.Event() + + def note_poll() -> bool: + polling.set() + return False + + async def acquire_and_release() -> None: + await lock.acquire(timeout=5, cancel_check=note_poll, poll_interval=0.01) + await lock.release() + + waiter = asyncio.create_task(asyncio.to_thread(asyncio.run, acquire_and_release())) + assert await asyncio.to_thread(polling.wait, 5) + + await asyncio.wait_for(lock.release(), 5) + + await waiter + assert not lock.is_locked + + class _Interrupt(BaseException): """Stands in for an exception a signal handler raises, without pytest treating it as a KeyboardInterrupt.""" From 41a78966de44bcca6496bc267413f306ffaab454 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bern=C3=A1t=20G=C3=A1bor?= Date: Thu, 15 Oct 2026 18:43:16 -0700 Subject: [PATCH 03/11] fix(async): read the gate from the async context --- src/filelock/asyncio.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/filelock/asyncio.py b/src/filelock/asyncio.py index b25c791a..2318d60b 100644 --- a/src/filelock/asyncio.py +++ b/src/filelock/asyncio.py @@ -365,7 +365,7 @@ async def _acquire_with_admission( poll_interval: float, start_time: float, ) -> AsyncAcquireReturnProxy: - async with self._context.transition_gate.hold_for_acquire( + async with self._async_context.transition_gate.hold_for_acquire( blocking=blocking, cancel_check=cancel_check, deadline=None if timeout < 0 else start_time + timeout, @@ -524,7 +524,7 @@ async def release(self, force: bool = False) -> None: # ty: ignore[invalid-meth """ if self._creator_pid != os.getpid() or not self.is_locked: return - async with self._context.transition_gate.hold(): + async with self._async_context.transition_gate.hold(): await self._release_serialized(force=force) async def _release_serialized(self, *, force: bool) -> None: From b0c74acd8dc22b22da60b2e60bf9d08b994b4a32 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bern=C3=A1t=20G=C3=A1bor?= Date: Thu, 15 Oct 2026 18:43:16 -0700 Subject: [PATCH 04/11] test(async-rw): pass soft lock options by name --- tests/test_async_read_write_cancel_wait.py | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/tests/test_async_read_write_cancel_wait.py b/tests/test_async_read_write_cancel_wait.py index d7a99bf1..af277ee3 100644 --- a/tests/test_async_read_write_cancel_wait.py +++ b/tests/test_async_read_write_cancel_wait.py @@ -47,9 +47,11 @@ def _sqlite(path: Path, *, is_singleton: bool) -> tuple[ReadWriteLock, AsyncRead def _soft(path: Path, *, is_singleton: bool) -> tuple[SoftReadWriteLock, AsyncSoftReadWriteLock]: - options: Final = {"heartbeat_interval": 0.1, "stale_threshold": 0.5, "poll_interval": 0.02} - return SoftReadWriteLock(path / "lock", is_singleton=is_singleton, **options), AsyncSoftReadWriteLock( - path / "lock", **options + return ( + SoftReadWriteLock( + path / "lock", is_singleton=is_singleton, heartbeat_interval=0.1, stale_threshold=0.5, poll_interval=0.02 + ), + AsyncSoftReadWriteLock(path / "lock", heartbeat_interval=0.1, stale_threshold=0.5, poll_interval=0.02), ) From 3b187458e878f2c9f0fe13b9d428df3faf9c929e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bern=C3=A1t=20G=C3=A1bor?= Date: Thu, 15 Oct 2026 18:43:43 -0700 Subject: [PATCH 05/11] docs(changelog): note cancellation and gate fixes --- docs/changelog/844.bugfix.rst | 4 ++++ 1 file changed, 4 insertions(+) create mode 100644 docs/changelog/844.bugfix.rst diff --git a/docs/changelog/844.bugfix.rst b/docs/changelog/844.bugfix.rst new file mode 100644 index 00000000..97b74899 --- /dev/null +++ b/docs/changelog/844.bugfix.rst @@ -0,0 +1,4 @@ +Cancelling an :class:`~filelock.AsyncReadWriteLock` or :class:`~filelock.AsyncSoftReadWriteLock` acquire that waits for +another process now stops the wait and reports the cancellation, instead of waiting until that process releases or +raising :class:`~filelock.Timeout`. A ``thread_local=True`` :class:`~filelock.AsyncFileLock` shared by threads with their +own event loops no longer deadlocks when one thread waits for the lock while another holds it. From 6016d28228cffeaa830f570afe5945e099f3c98d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bern=C3=A1t=20G=C3=A1bor?= Date: Thu, 15 Oct 2026 18:44:47 -0700 Subject: [PATCH 06/11] docs(changelog): use the project's spelling --- docs/changelog/844.bugfix.rst | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/changelog/844.bugfix.rst b/docs/changelog/844.bugfix.rst index 97b74899..926aa345 100644 --- a/docs/changelog/844.bugfix.rst +++ b/docs/changelog/844.bugfix.rst @@ -1,4 +1,4 @@ -Cancelling an :class:`~filelock.AsyncReadWriteLock` or :class:`~filelock.AsyncSoftReadWriteLock` acquire that waits for +Canceling an :class:`~filelock.AsyncReadWriteLock` or :class:`~filelock.AsyncSoftReadWriteLock` acquire that waits for another process now stops the wait and reports the cancellation, instead of waiting until that process releases or raising :class:`~filelock.Timeout`. A ``thread_local=True`` :class:`~filelock.AsyncFileLock` shared by threads with their own event loops no longer deadlocks when one thread waits for the lock while another holds it. From ffc29d5d3c3ce3b06aaa4e6b29908ea3aaf62869 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bern=C3=A1t=20G=C3=A1bor?= Date: Thu, 15 Oct 2026 18:54:39 -0700 Subject: [PATCH 07/11] test(async-rw): skip soft wait cases without hard links --- tests/test_async_read_write_cancel_wait.py | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/tests/test_async_read_write_cancel_wait.py b/tests/test_async_read_write_cancel_wait.py index af277ee3..45c06121 100644 --- a/tests/test_async_read_write_cancel_wait.py +++ b/tests/test_async_read_write_cancel_wait.py @@ -46,7 +46,9 @@ def _sqlite(path: Path, *, is_singleton: bool) -> tuple[ReadWriteLock, AsyncRead return ReadWriteLock(path / "lock.db", is_singleton=is_singleton), AsyncReadWriteLock(path / "lock.db") -def _soft(path: Path, *, is_singleton: bool) -> tuple[SoftReadWriteLock, AsyncSoftReadWriteLock]: +def _soft( # pragma: needs hard-link + path: Path, *, is_singleton: bool +) -> tuple[SoftReadWriteLock, AsyncSoftReadWriteLock]: return ( SoftReadWriteLock( path / "lock", is_singleton=is_singleton, heartbeat_interval=0.1, stale_threshold=0.5, poll_interval=0.02 @@ -56,7 +58,9 @@ def _soft(path: Path, *, is_singleton: bool) -> tuple[SoftReadWriteLock, AsyncSo _Make = Callable[..., tuple[ReadWriteLock | SoftReadWriteLock, AsyncReadWriteLock | AsyncSoftReadWriteLock]] -_FAMILIES: Final = pytest.mark.parametrize("make", [pytest.param(_sqlite, id="sqlite"), pytest.param(_soft, id="soft")]) +_FAMILIES: Final = pytest.mark.parametrize( + "make", [pytest.param(_sqlite, id="sqlite"), pytest.param(_soft, id="soft", marks=pytest.mark.requires_hard_links)] +) _MODES: Final = pytest.mark.parametrize("mode", [pytest.param("read", id="read"), pytest.param("write", id="write")]) From 472da1845a6cc806f701b6120e70ff3881370477 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bern=C3=A1t=20G=C3=A1bor?= Date: Thu, 15 Oct 2026 20:04:47 -0700 Subject: [PATCH 08/11] test(async-rw): close both singleton caches in one loop --- tests/test_async_read_write_cancel_wait.py | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/tests/test_async_read_write_cancel_wait.py b/tests/test_async_read_write_cancel_wait.py index 45c06121..64a737cc 100644 --- a/tests/test_async_read_write_cancel_wait.py +++ b/tests/test_async_read_write_cancel_wait.py @@ -31,15 +31,15 @@ @pytest.fixture(autouse=True) def _clear_singleton_caches() -> Generator[None]: - ReadWriteLock._instances.clear() - SoftReadWriteLock._instances.clear() + caches: Final = (ReadWriteLock._instances, SoftReadWriteLock._instances) + for cache in caches: + cache.clear() yield - for lock in filter(None, (ref() for ref in list(ReadWriteLock._instances.valuerefs()))): + # One loop for both caches, so a platform where only one kind of lock runs still closes through it. + for lock in filter(None, [ref() for cache in caches for ref in list(cache.valuerefs())]): lock.close() - for soft in filter(None, (ref() for ref in list(SoftReadWriteLock._instances.valuerefs()))): - soft.close() - ReadWriteLock._instances.clear() - SoftReadWriteLock._instances.clear() + for cache in caches: + cache.clear() def _sqlite(path: Path, *, is_singleton: bool) -> tuple[ReadWriteLock, AsyncReadWriteLock]: From efc4d93a77c6b73a163f6f139c6a4d3c60b0b2ec Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bern=C3=A1t=20G=C3=A1bor?= Date: Thu, 15 Oct 2026 23:18:40 -0700 Subject: [PATCH 09/11] docs(async): split the cancellation and gate notes One sentence per idea in the sliced-wait, soft rollback and per-context gate comments, and a changelog entry in the same style. --- docs/changelog/844.bugfix.rst | 6 +++--- src/filelock/_read_write.py | 4 ++-- src/filelock/_soft_rw/_async.py | 7 +++---- src/filelock/asyncio.py | 2 +- 4 files changed, 9 insertions(+), 10 deletions(-) diff --git a/docs/changelog/844.bugfix.rst b/docs/changelog/844.bugfix.rst index 926aa345..e3fed168 100644 --- a/docs/changelog/844.bugfix.rst +++ b/docs/changelog/844.bugfix.rst @@ -1,4 +1,4 @@ Canceling an :class:`~filelock.AsyncReadWriteLock` or :class:`~filelock.AsyncSoftReadWriteLock` acquire that waits for -another process now stops the wait and reports the cancellation, instead of waiting until that process releases or -raising :class:`~filelock.Timeout`. A ``thread_local=True`` :class:`~filelock.AsyncFileLock` shared by threads with their -own event loops no longer deadlocks when one thread waits for the lock while another holds it. +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. diff --git a/src/filelock/_read_write.py b/src/filelock/_read_write.py index 7e066d4d..7c2861ef 100644 --- a/src/filelock/_read_write.py +++ b/src/filelock/_read_write.py @@ -953,8 +953,8 @@ def _configure_and_begin( def _execute_waiting( connection: _ForkSafeConnection, statements: str, timeout: float, *, blocking: bool, operation: _Operation ) -> None: - # SQLite's busy wait cannot be interrupted, so an acquire its caller may abandon waits in bounded slices and - # asks between them whether to stop. + # 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( diff --git a/src/filelock/_soft_rw/_async.py b/src/filelock/_soft_rw/_async.py index a224377a..f751d6b9 100644 --- a/src/filelock/_soft_rw/_async.py +++ b/src/filelock/_soft_rw/_async.py @@ -318,10 +318,9 @@ def _leave_backend(self) -> None: 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. A sync acquire that got the lock before it saw - # the cancellation has committed its claim, set the hold, and started 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. + # 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: diff --git a/src/filelock/asyncio.py b/src/filelock/asyncio.py index 2318d60b..8c370373 100644 --- a/src/filelock/asyncio.py +++ b/src/filelock/asyncio.py @@ -831,7 +831,7 @@ class AsyncFileLockContext(FileLockContext): loop: asyncio.AbstractEventLoop | None = None #: Serializes the tasks that transition this context. An acquire keeps it while it polls, so a thread-local lock - #: needs one per thread: a thread waiting for the lock must not keep the thread that holds it from releasing. + #: needs one per thread. A thread waiting for the lock must not keep the thread that holds it from releasing. transition_gate: _AsyncTransitionGate = field(default_factory=_new_transition_gate) From c3887587101f5d7bf713ac78970506fd04bad526 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bern=C3=A1t=20G=C3=A1bor?= Date: Fri, 16 Oct 2026 01:22:44 -0700 Subject: [PATCH 10/11] =?UTF-8?q?=F0=9F=A7=AA=20test(async-rw):=20pin=20th?= =?UTF-8?q?e=20read=20retry=20rollback?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit executescript commits a pending transaction under SQLite's legacy transaction control, so the retry succeeded with or without the rollback and the test could not tell. Record whether the retried statement starts inside a transaction and expect it not to. --- tests/test_async_read_write_cancellation.py | 16 ++++++++++------ 1 file changed, 10 insertions(+), 6 deletions(-) diff --git a/tests/test_async_read_write_cancellation.py b/tests/test_async_read_write_cancellation.py index f35c90d9..ef5aee88 100644 --- a/tests/test_async_read_write_cancellation.py +++ b/tests/test_async_read_write_cancellation.py @@ -367,15 +367,18 @@ async def test_read_retried_after_its_select_waited_starts_a_fresh_transaction( lock_file: str, mocker: MockerFixture ) -> None: busy_once = True + seen_in_transaction: list[bool] = [] def executescript(real_connection: sqlite3.Connection, statement: str) -> sqlite3.Cursor: nonlocal busy_once - if busy_once and "SELECT name" in statement: - # A writer can take the lock between the deferred BEGIN and the SELECT that needs the shared lock. - busy_once = False - real_connection.execute("BEGIN TRANSACTION") - msg = "database is locked" - raise sqlite3.OperationalError(msg) + if "SELECT name" in statement: + if busy_once: + # A writer can take the lock between the deferred BEGIN and the SELECT that needs the shared lock. + busy_once = False + real_connection.execute("BEGIN TRANSACTION") + msg = "database is locked" + raise sqlite3.OperationalError(msg) + seen_in_transaction.append(real_connection.in_transaction) return real_connection.executescript(statement) _patch_async_connection(mocker, executescript=executescript) @@ -384,6 +387,7 @@ def executescript(real_connection: sqlite3.Connection, statement: str) -> sqlite async with lock.read_lock(): assert_read_write_lock_state(lock_file, "write", available=False) await lock.close() + assert seen_in_transaction == [False] def _patch_async_rollback( From fbc5893d37982f6138a4927870fb878d8e67838f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bern=C3=A1t=20G=C3=A1bor?= Date: Fri, 16 Oct 2026 12:38:27 -0700 Subject: [PATCH 11/11] =?UTF-8?q?=F0=9F=90=9B=20fix(rw):=20check=20cancell?= =?UTF-8?q?ation=20between=20busy=20retries?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit With other threads alive, a blocking acquire waits in 50 ms slices so a fork can interleave. The cancel check only runs inside a single wait longer than 100 ms, so no slice ever reached it and a canceled acquire kept retrying until its timeout. The retry loop now asks the cancel check after each timed-out slice and gives up when it is true. The read-retry test mocks every connection from the second on, so the in-place rollback path is still the one exercised. --- src/filelock/_read_write.py | 2 +- tests/test_async_read_write_cancellation.py | 26 ++++++++++++++++++++- 2 files changed, 26 insertions(+), 2 deletions(-) diff --git a/src/filelock/_read_write.py b/src/filelock/_read_write.py index 7c2861ef..5b01da2d 100644 --- a/src/filelock/_read_write.py +++ b/src/filelock/_read_write.py @@ -810,7 +810,7 @@ def _acquire( 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 diff --git a/tests/test_async_read_write_cancellation.py b/tests/test_async_read_write_cancellation.py index ef5aee88..f2f6a314 100644 --- a/tests/test_async_read_write_cancellation.py +++ b/tests/test_async_read_write_cancellation.py @@ -382,6 +382,8 @@ def executescript(real_connection: sqlite3.Connection, statement: str) -> sqlite return real_connection.executescript(statement) _patch_async_connection(mocker, executescript=executescript) + # With other threads alive the acquire waits in short slices that never reach the in-place retry under test. + mocker.patch.object(threading, "active_count", return_value=1) lock = AsyncReadWriteLock(lock_file, is_singleton=False) async with lock.read_lock(): @@ -390,6 +392,28 @@ def executescript(real_connection: sqlite3.Connection, statement: str) -> sqlite assert seen_in_transaction == [False] +@pytest.mark.asyncio +async def test_write_retried_after_its_begin_waited_is_acquired(lock_file: str, mocker: MockerFixture) -> None: + busy_once = True + + def executescript(real_connection: sqlite3.Connection, statement: str) -> sqlite3.Cursor: + nonlocal busy_once + if "BEGIN EXCLUSIVE" in statement and busy_once: + busy_once = False + msg = "database is locked" + raise sqlite3.OperationalError(msg) + return real_connection.executescript(statement) + + _patch_async_connection(mocker, executescript=executescript) + mocker.patch.object(threading, "active_count", return_value=1) + lock = AsyncReadWriteLock(lock_file, is_singleton=False) + + async with lock.write_lock(): + assert_read_write_lock_state(lock_file, "read", available=False) + await lock.close() + assert not busy_once + + def _patch_async_rollback( mocker: MockerFixture, loop: asyncio.AbstractEventLoop, @@ -539,7 +563,7 @@ def connect( timeout=timeout, ) connection_count += 1 - if connection_count != 2: + if connection_count < 2: return real_connection configuration: dict[str, Callable[[str], sqlite3.Cursor] | Callable[[], None]] = {} if executescript is not None: