🐛 fix(async): release from the finalizer and keep I/O off the loop

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.

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 <filelock.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
 ************************

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.

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

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

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)