Request for Asyncio Support

[oldcai]

It would block other coroutines when acquiring locks now.

Hope it can use unblocking ways to wait for the lock. Such as asyncio.sleep

By the way, in the meantime, I'm using this way to keep other coroutines running.

    acquired_lock = False
    while not acquired_lock:
        try:
            BEFORE_NET_START_LOCK.acquire(timeout=0)
            acquired_lock = True
        except Timeout:
            await asyncio.sleep(1)

Hope this code can help with someone facing the same problem.

[xkortex]

This ought to be implemented with async with: https://www.python.org/dev/peps/pep-0492/#asynchronous-context-managers-and-async-with

This ought to improve performance as well, since it looks like acquire currently uses a busy-wait: https://github.com/benediktschmitt/py-filelock/blob/master/filelock.py#L284

[gaborbernat]

This ought to be implemented with async with: python.org/dev/peps/pep-0492/#asynchronous-context-managers-and-async-with

That would imply an event loop running, which is often not the case. So we cannot add asyncio support straight up, however if someone would make this library sansio type we'd be happy to review that PR 👍

[maxfischer2781]

I've gone over the code and as far as I can tell there is no way to make things nicely async.

Fundamentally, the locking via _acquire requires polling and all that asyncio/trio/... can do is provide an await sleep(poll_interval) instead of the current time.sleep(poll_interval). That is really bad for the event loop at the default poll_interval=0.05; some exponential backoff might help here.
Alternatively, one could reimplement a blocking acquire and use a thread (as suggested on SO). That is ideal if we definitely want to acquire a few locks; it is fatal with timeouts/cancellation and many locks, since the threads will keep on living "forever" in the background in the worst case.

---

If you are fine with having a busy loop, I would suggest (and can provide) an async with context manager that basically uses the while True: + await sleep to poll the lock. There is no way to make it truly sansio, but the sleep function could be parametrized easily.

This could either be a separate class (say AsyncFileLock) that wraps a regular FileLock or merely adding the required magic methods for async with directly to FileLock. Would that be simple enough to add to the library?

[gaborbernat]

My 2c is that we should do this totally in parallel with the existing sync classes 😊 so yeah adding AsyncFileLock class is the way ahead, and no busy-wait just full event loop design.

[axoroll7]

The busy polling (asyncio.wait_for) to acquire the lock is inevitable. But, to lessen the burden, can we watch for the lock file before entering the busy lock ? Using true aio, it would be possible to use inotify with epoll, and the Windows equivalent. I do not know any library to do this, it is just a idea.

[dolamroth]

I have implemented a semi-efficient asynchronous filelock using anyio: https://github.com/dolamroth/starlette-web/blob/main/starlette_web/common/files/filelock.py

It passes some basic tests: https://github.com/dolamroth/starlette-web/blob/main/starlette_web/tests/core/helpers/base_cache_tester.py
If anyone is interested, you may pick it up.

[shughes-uk]

We had implemented async support using

@asynccontextmanager
async def async_lock(lock: BaseFileLock):
    loop = asyncio.get_event_loop()
    await loop.run_in_executor(None, lock.acquire)
    yield
    await loop.run_in_executor(None, lock.release)

However the recent change to make locks thread local broke this and resulted in our code deadlocking. We have moved to using a static ThreadPoolExecutor with one thread, but this is definitely a gotcha.

LOCK_POOL = ThreadPoolExecutor(max_workers=1)


@asynccontextmanager
async def async_lock(lock: BaseFileLock):
    loop = asyncio.get_event_loop()
    await loop.run_in_executor(LOCK_POOL, lock.acquire)
    yield
    await loop.run_in_executor(LOCK_POOL, lock.release)

[CarliJoy]

I ended up with the following implementation

import asyncio
import contextlib
from collections.abc import AsyncGenerator

from filelock import BaseFileLock

@asynccontextmanager
async def _acquire_lock(fl: BaseFileLock) -> AsyncGenerator[None, None]:
    """
    Acquire filelock, async implementation
    """
    for _ in range(int(10 / 0.05)):
        with contextlib.suppress(TimeoutError):
            try:
                fl.acquire(blocking=False)
                # We want to have the exclusive lock
                if fl.lock_counter <= 1:
                    yield
                    break
            finally:
                fl.release()
        await asyncio.sleep(0.05)
    else:
        raise TimeoutError("Could not obtain lock after 10 seconds")

For me the main point is that the sleep is non blocking.
The release and acquire code is fast enough to not block other things of the event loop.
Note also the try: … finally: the will ensure that the file lock is released on exceptions.

For me it was also important, that only one lock is active:
In the documentation the following sync code was given as example why recursive locks should be allowed

def cite1():
    with lock:
        with open(file_path, "a") as f:
            f.write("I hate it when he does that.")


def cite2():
    with lock:
        with open(file_path, "a") as f:
            f.write("You don't want to sell me death sticks.")


# The lock is acquired here.
with lock:
    cite1()
    cite2()
# And released 

However transferring this example to a sync world, show that is not valid for async .

# cite1 and cite2 are implemented async and with multiple writes

async def run():
    with TaskGroup() as tg:
        tg.create_task(cite1())
        tg.create_task(cite2()) # potentially boom as the same file is opened/written to at the same time.

Of course wrapping also the TaskGroup in an async with _acquire_lock(): would deadlock.
However using it differently would be unsafe.

I also wrote a little test that shows my implementation works as expected for me:

Show Test implementation

import asyncio
import time
from pathlib import Path

import pytest


@pytest.mark.asyncio
async def test_async_file_lock(tmp_path: Path) -> None:
    """
    Acquiring the lock is non-blocking
    """
    test_file = tmp_path / "test_file.txt"
    start = time.perf_counter_ns()
    regular_calls: list[float] = []

    def get_elapsed() -> int:
        return time.perf_counter_ns() - start

    def debug(msg: str) -> None:
        print(f"{get_elapsed()/10**9:0.5f}: {msg}")

    async def reader(delay: float) -> tuple[bytes, float]:
        """
        Read file a *delay* seconds
        """
        await asyncio.sleep(delay)
        debug(f"Obtaining lock for reading after {delay}s delay")
        async with pkg_project_mapping._file_lock(test_file):
            debug(f"Got lock for reading after {delay}s delay")
            result = test_file.read_bytes(), get_elapsed() / 1e9
        debug("Released lock for reading after {delay}s delay")
        return result

    async def writer() -> None:
        """
        Write to file 0.1 lines per second and append to it after one second of waiting
        """
        for repeat in range(2):
            debug(f"Obtain write lock {repeat}")
            async with pkg_project_mapping._file_lock(test_file):
                debug(f"Got write lock {repeat}")
                with test_file.open("a") as f:
                    for i in range(10):
                        f.write(f"{i};")
                        debug(f"Wrote {i} ({repeat})")
                        await asyncio.sleep(0.1)
                    f.write("\n")
            debug(f"Release lock for write {repeat}")
            await asyncio.sleep(1)

    async def regular_caller() -> None:
        """
        Simulate something that running regular
        
        this ensures that we can check that  writing/reading/locking is non blocking for
        other async task        
        """
        for i in range(int(5 / 0.05)):  # 20 sec
            await asyncio.sleep(0.05)
            elapsed = get_elapsed() / 1e9
            if regular_calls:
                debug(f"Called regular {i:03} after {elapsed-regular_calls[-1]:0.4f}s")
            else:
                debug(f"Called regular {i:03}")
            regular_calls.append(elapsed)

    async with asyncio.TaskGroup() as tg:
        debug("Print Start Tasks")
        tg.create_task(regular_caller())
        tg.create_task(writer())
        reader1 = tg.create_task(reader(0.02))
        reader2 = tg.create_task(reader(2.5))
        reader3 = tg.create_task(reader(2.6))
        reader4 = tg.create_task(reader(2.7))

    # check that regular task was running non-blocking
    for n in range(len(regular_calls) - 1):
        diff = regular_calls[n + 1] - regular_calls[n]
        assert diff == pytest.approx(diff, abs=0.001)

    expected_str = b"0;1;2;3;4;5;6;7;8;9;\n"
    assert reader1.result()[0] == expected_str
    # expected runtime 1.0s for the delay and 0.05 for the actual writing
    assert reader1.result()[1] == pytest.approx(1.05, abs=0.1)

    assert reader2.result()[0] == expected_str * 2
    # expected runtime 3s for the delay and 0.1 for the actual writing
    assert reader2.result()[1] == pytest.approx(3.1, abs=0.2)

    assert reader3.result()[0] == expected_str * 2
    # expected runtime 3s for the delay and 0.1 for the actual writing
    assert reader3.result()[1] == pytest.approx(3.105, abs=0.2)

    assert reader4.result()[0] == expected_str * 2
    # expected runtime 3s for the delay and 0.1 for the actual writing
    assert reader4.result()[1] == pytest.approx(3.11, abs=0.2)