From 9a41cba1a9a5e1c4e92734d5f5ff780d2dd95909 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bern=C3=A1t=20G=C3=A1bor?= Date: Wed, 7 Oct 2026 17:52:41 -0700 Subject: [PATCH] =?UTF-8?q?=F0=9F=90=9B=20fix(async):=20release=20from=20f?= =?UTF-8?q?inalizer,=20keep=20I/O=20off=20loop?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A held async lock collected with loop= pointing at a loop in another thread called loop.create_task from the wrong thread, so the release stalled or never ran; the finalizer now uses run_coroutine_threadsafe and keeps the future alive. With no loop left at all, it releases a sync backend directly instead of leaving the marker behind. With run_in_executor=True the lifetime expiry check and the realpath of the lock path ran on the loop thread; both now run in the executor. Docs: loop= only serves the finalizer, async deadlock detection does not see sync holders, the acquire() example awaits, and thread_local defaults to False for async locks. --- docs/changelog/800.bugfix.rst | 2 + docs/concepts.rst | 4 ++ docs/how-to.rst | 11 +--- src/filelock/asyncio.py | 64 +++++++++++++++---- tests/test_async_filelock.py | 64 +++++++++++++++++++ ...est_async_filelock_acquire_cancellation.py | 13 +++- 6 files changed, 135 insertions(+), 23 deletions(-) create mode 100644 docs/changelog/800.bugfix.rst diff --git a/docs/changelog/800.bugfix.rst b/docs/changelog/800.bugfix.rst new file mode 100644 index 00000000..8d0f34974 --- /dev/null +++ b/docs/changelog/800.bugfix.rst @@ -0,0 +1,2 @@ +When the garbage collector finalizes a held :class:`~filelock.BaseAsyncFileLock`, it releases on a ``loop`` running in +another thread, or in place once no loop is left; the ``lifetime`` check and path resolution now run in the executor. diff --git a/docs/concepts.rst b/docs/concepts.rst index 605ac43d..b784bcf9 100644 --- a/docs/concepts.rst +++ b/docs/concepts.rst @@ -102,6 +102,10 @@ filelock detects this and raises ``RuntimeError`` with a message suggesting ``is This detection only applies to blocking acquires (``timeout < 0``) within the same thread. Non-blocking or timed acquires raise :class:`Timeout ` as usual. +Async locks scope the check to the asyncio task instead of the thread, so a sync lock and an async lock do not see each +other's holds. Holding a ``FileLock`` on the event loop thread and then awaiting a blocking ``AsyncFileLock`` acquire of +the same file polls forever instead of raising; give that acquire a timeout. + ************************ How file locking works ************************ diff --git a/docs/how-to.rst b/docs/how-to.rst index 58ecf8e6..300b9e4b 100644 --- a/docs/how-to.rst +++ b/docs/how-to.rst @@ -337,14 +337,9 @@ By default, async locks run blocking I/O in a thread pool. You can customize thi # Or disable executor (only if your filesystem is non-blocking) lock = AsyncFileLock("work.lock", run_in_executor=False) -You can also pass a specific event loop: - -.. code-block:: python - - import asyncio - - loop = asyncio.new_event_loop() - lock = AsyncFileLock("work.lock", loop=loop) +``acquire`` and ``release`` always run on the event loop that awaits them. The ``loop`` argument only matters when the +lock is garbage collected while still held: the finalizer schedules the release on that loop if no loop runs in the +collecting thread. Without any loop, it releases a built-in backend synchronously. Async locks default to ``thread_local=False`` (unlike sync locks which default to ``True``) because the acquiring and releasing threads may differ when using an executor. diff --git a/src/filelock/asyncio.py b/src/filelock/asyncio.py index d6ee4965..37fa157f 100644 --- a/src/filelock/asyncio.py +++ b/src/filelock/asyncio.py @@ -4,6 +4,7 @@ import asyncio import contextlib +import functools import logging import os import time @@ -67,6 +68,7 @@ _ASYNC_CONTEXT_RELEASE_ERRORS: Final[str] = "context body, release cancellation, and backend release failed" _ASYNC_RELEASE_CANCELLATION_MARKER_ATTR: Final[str] = "_filelock_async_release_cancellation" _ASYNC_RELEASE_CANCELLATION_MARKER: Final[list[None]] = [] +_FINALIZER_RELEASES: Final[set[asyncio.Task[None] | futures.Future[None]]] = set() _AT = TypeVar("_AT", bound="BaseAsyncFileLock") @@ -166,7 +168,7 @@ def __init__( # ruff:ignore[too-many-arguments] # public constructor: one para :param mode: file permissions for the lockfile. When not specified, the OS controls permissions via umask and default ACLs, preserving POSIX default ACL inheritance in shared directories. :param thread_local: Whether this object's internal context should be thread local or not. If this is set to - ``False`` then the lock will be reentrant across threads. When ``True`` (the default), **all fields of the + ``False`` (the default) then the lock will be reentrant across threads. When ``True``, **all fields of the lock's internal context are per-thread**, including the configuration values ``poll_interval``, ``timeout``, ``blocking``, ``mode``, and ``lifetime``. Setting one of these properties from one thread does not change the value seen by another thread; threads that did not perform the write continue to see the value supplied @@ -200,7 +202,9 @@ def __init__( # ruff:ignore[too-many-arguments] # public constructor: one para :meth:`~BaseAsyncFileLock.acquire` returns. With ``run_in_executor=True`` (the default) it runs in the backend executor. It must not close or unlock the descriptor; a raise rolls the acquisition back. :class:`AsyncSoftFileLock` rejects it. - :param loop: The event loop to use. If not specified, the running event loop will be used. + :param loop: the event loop a garbage-collected lock schedules its release on when no loop runs in the + collecting thread. :meth:`~BaseAsyncFileLock.acquire` and :meth:`~BaseAsyncFileLock.release` always run + on the loop that awaits them. :param run_in_executor: If this is set to ``True`` then the lock will be acquired in an executor. :param executor: The executor to use. If not specified, the default executor will be used. @@ -296,15 +300,15 @@ async def acquire( # ty: ignore[invalid-method-override] .. code-block:: python # You can use this method in the context manager (recommended) - with lock.acquire(): + async with await lock.acquire(): pass # Or use an equivalent try-finally construct: - lock.acquire() + await lock.acquire() try: pass finally: - lock.release() + await lock.release() """ self._raise_if_inherited() @@ -347,7 +351,7 @@ async def _acquire_with_admission( poll_interval=poll_interval, ): # A canceled provisional acquire must finish rollback before another caller can claim its descriptor. - canonical = _canonical(self.lock_file) + canonical = await self._canonical_lock_file() self._context.lock_counter += 1 self._raise_if_would_deadlock(canonical, timeout=timeout, blocking=blocking) self._context.claim_root = canonical @@ -367,6 +371,17 @@ async def _acquire_with_admission( self._commit_acquire(canonical) return AsyncAcquireReturnProxy(lock=self) + async def _canonical_lock_file(self) -> str: + # realpath stats every parent directory, which blocks the loop as long as any filesystem call on a slow mount. + # A coroutine backend runs its own calls on the loop, so it never starts an executor thread for this either. + if not self.run_in_executor or iscoroutinefunction(self._acquire): + return _canonical(self.lock_file) + future = _run_in_executor( + asyncio.get_running_loop(), self.executor, functools.partial(_canonical, self.lock_file) + ) + await _wait_until_done(future) + return _future_result(future) + async def _async_poll_until_acquired( self, *, @@ -381,7 +396,6 @@ async def _async_poll_until_acquired( while True: self._raise_if_inherited() if not self.is_locked: - self._try_break_expired_lock() _LOGGER.debug("Attempting to acquire lock %s on %s", lock_id, lock_filename) await self._run_acquire_attempt() self._raise_if_inherited() @@ -405,7 +419,7 @@ async def _run_acquire_attempt(self) -> None: acquire_future = self._start_internal_method( self._acquire_with_fork_tracking_async if iscoroutinefunction(self._acquire) - else self._acquire_with_fork_tracking + else self._break_expired_and_acquire ) try: await _wait_until_done(acquire_future) @@ -547,7 +561,13 @@ def _start_tracked_release(self) -> asyncio.Future[_BackendOutcome[None]]: else self._release_with_fork_tracking ) + def _break_expired_and_acquire(self) -> None: + # Runs as one backend call so the lifetime check's lstat and unlink stay off the event loop with the acquire. + self._try_break_expired_lock() + self._acquire_with_fork_tracking() + async def _acquire_with_fork_tracking_async(self) -> None: + self._try_break_expired_lock() with _fork_transition(self): try: await cast("Callable[[], Awaitable[None]]", self._acquire)() @@ -701,16 +721,32 @@ def __del__(self) -> None: if vars(self).get("_creator_pid") != os.getpid(): return # pragma: forked child with contextlib.suppress(Exception): + pending: asyncio.Task[None] | futures.Future[None] try: loop = asyncio.get_running_loop() except RuntimeError: - loop = self._context.loop if self._context.loop and not self._context.loop.is_closed() else None - if loop is None: - return - if not loop.is_running(): # pragma: no cover - loop.run_until_complete(self.release(force=True)) + if (loop := self._context.loop) is None or loop.is_closed(): + self._release_without_loop() + return + if not loop.is_running(): # pragma: no cover + loop.run_until_complete(self.release(force=True)) + return + # The stored loop runs in another thread, and only its own thread may create tasks on it. + pending = asyncio.run_coroutine_threadsafe(self.release(force=True), loop) else: - loop.create_task(self.release(force=True)) + pending = loop.create_task(self.release(force=True)) + # The loop keeps only a weak reference to a task, so hold one until the release finishes. + _FINALIZER_RELEASES.add(pending) + pending.add_done_callback(_FINALIZER_RELEASES.discard) + + def _release_without_loop(self) -> None: + # Nothing can await a coroutine release here (at interpreter exit the loop is long gone), so call a sync backend + # directly, the way BaseFileLock.__del__ does; a coroutine backend has no way to run and keeps its lock. + if self.is_locked and not iscoroutinefunction(self._release): + try: + self._release_with_fork_tracking() + finally: + self._commit_release_if_released() @dataclass diff --git a/tests/test_async_filelock.py b/tests/test_async_filelock.py index 2a2ce245..01bb0a9c 100644 --- a/tests/test_async_filelock.py +++ b/tests/test_async_filelock.py @@ -1302,3 +1302,67 @@ def test_async_del_without_any_loop_returns(tmp_path: Path, mocker: MockerFixtur def test_asyncio_module_dir_lists_only_its_public_api() -> None: assert dir(filelock.asyncio) == sorted(filelock.asyncio.__all__) + + +def test_async_del_without_any_loop_releases_a_sync_backend(tmp_path: Path) -> None: + lock = AsyncSoftFileLock(str(tmp_path / "a")) + asyncio.run(lock.acquire()) + + lock.__del__() # ruff:ignore[unnecessary-dunder-call] # a finalizer test cannot ride on collection timing + + assert (lock.is_locked, lock.lock_counter, (tmp_path / "a").exists()) == (False, 0, False) + + +def test_async_del_schedules_release_on_a_stored_loop_in_another_thread(tmp_path: Path, mocker: MockerFixture) -> None: + loop = asyncio.new_event_loop() + runner = threading.Thread(target=loop.run_forever) + runner.start() + try: + lock = AsyncSoftFileLock(str(tmp_path / "a"), loop=loop) + asyncio.run_coroutine_threadsafe(lock.acquire(), loop).result(timeout=5) + schedule = mocker.spy(asyncio, "run_coroutine_threadsafe") + + lock.__del__() # ruff:ignore[unnecessary-dunder-call] # a finalizer test cannot ride on collection timing + + schedule.spy_return.result(timeout=5) + assert (lock.is_locked, (tmp_path / "a").exists()) == (False, False) + finally: + loop.call_soon_threadsafe(loop.stop) + runner.join(timeout=5) + loop.close() + + +@pytest.mark.asyncio +async def test_async_del_inside_a_running_loop_schedules_release(tmp_path: Path, mocker: MockerFixture) -> None: + lock = AsyncSoftFileLock(str(tmp_path / "a")) + await lock.acquire() + schedule = mocker.spy(asyncio.get_running_loop(), "create_task") + + lock.__del__() # ruff:ignore[unnecessary-dunder-call] # a finalizer test cannot ride on collection timing + + await schedule.spy_return + assert (lock.is_locked, (tmp_path / "a").exists()) == (False, False) + + +@pytest.mark.filterwarnings("ignore::filelock.SoftFileLockLifetimeWarning") +@pytest.mark.asyncio +async def test_executor_acquire_keeps_filesystem_calls_off_the_loop(tmp_path: Path, mocker: MockerFixture) -> None: + lock_path = tmp_path / "a" + lock_path.touch() + os.utime(lock_path, (0, 0)) + lock = AsyncSoftFileLock(str(lock_path), lifetime=1) + real_lstat = os.lstat + lstat_threads: set[int] = set() + + def lstat(path: str, *, dir_fd: int | None = None) -> os.stat_result: + lstat_threads.add(threading.get_ident()) + return real_lstat(path, dir_fd=dir_fd) + + mocker.patch("os.lstat", side_effect=lstat) + await lock.acquire() + mocker.stopall() + await lock.release() + + # Breaking the expired marker lstats it on every platform, so an empty set means the spy saw nothing. + assert lstat_threads + assert threading.get_ident() not in lstat_threads diff --git a/tests/test_async_filelock_acquire_cancellation.py b/tests/test_async_filelock_acquire_cancellation.py index c7f9d09c..eaefdc7c 100644 --- a/tests/test_async_filelock_acquire_cancellation.py +++ b/tests/test_async_filelock_acquire_cancellation.py @@ -134,18 +134,29 @@ async def test_acquire_cancellation_before_executor_start_rolls_back(tmp_path: P executor_started = threading.Event() release_executor = threading.Event() acquired = threading.Event() + second_started = threading.Event() + release_second = threading.Event() with ThreadPoolExecutor(max_workers=1) as executor: blocker = executor.submit(_block_executor, executor_started, release_executor) assert executor_started.wait(timeout=5) lock = AsyncFileLock(tmp_path / "a", executor=executor, on_acquired=lambda _fd: acquired.set()) task = asyncio.create_task(lock.acquire()) await asyncio.sleep(0) + # The acquire resolves the lock path on the executor first, so queue a second blocker behind that job to hold + # the backend attempt that follows it. + second_blocker = executor.submit(_block_executor, second_started, release_second) + release_executor.set() + assert await asyncio.to_thread(second_started.wait, 5) + await asyncio.sleep( + 0.05 + ) # the resolved path wakes the task, which queues the attempt behind the second blocker assert (lock.lock_counter, acquired.is_set()) == (1, False) task.cancel() - release_executor.set() + release_second.set() with pytest.raises(asyncio.CancelledError): await task blocker.result(timeout=5) + second_blocker.result(timeout=5) assert acquired.is_set() assert (lock.is_locked, lock.lock_counter) == (False, 0)