aio-libs/aiorwlock

Read/Write Lock - synchronization primitive for asyncio

aiorwlock

Read write lock for asyncio . A RWLock maintains a pair of associated locks, one for read-only operations and one for writing. The read lock may be held simultaneously by multiple reader tasks, so long as there are no writers. The write lock is exclusive.

Whether or not a read-write lock will improve performance over the use of a mutual exclusion lock depends on the frequency that the data is read compared to being modified. For example, a collection that is initially populated with data and thereafter infrequently modified, while being frequently searched is an ideal candidate for the use of a read-write lock. However, if updates become frequent then the data spends most of its time being exclusively locked and there is little, if any increase in concurrency.

Implementation is almost direct port from this patch.

Example

import asyncio
import aiorwlock


async def go():
    rwlock = aiorwlock.RWLock()

    # acquire reader lock, multiple coroutines allowed to hold the lock
    async with rwlock.reader_lock:
        print('inside reader lock')
        await asyncio.sleep(0.1)

    # acquire writer lock, only one coroutine can hold the lock
    async with rwlock.writer_lock:
        print('inside writer lock')
        await asyncio.sleep(0.1)


asyncio.run(go())

Fast path

By default RWLock switches context on lock acquiring. That allows to other waiting tasks get the lock even if task that holds the lock doesn't contain context switches (await fut statements).

The default behavior can be switched off by fast argument: RWLock(fast=True).

Long story short: lock is safe by default, but if you sure you have context switches (await, async with, async for or yield from statements) inside locked code you may want to use fast=True for minor speedup.

TLA+ Specification

TLA+ specification of aiorwlock provided in this repository.

License

aiorwlock is offered under the Apache 2 license.

aiorwlock's Issues

Lock in illegal state after asyncio.exceptions.CancelledError is raised

Scenario

1. Reader holds the reader lock
2. Writer waits for the lock
3. asyncio.exceptions.CancelledError is raised
4. No one can acquire the lock again

My motivation was to use "try lock" with timeout by using asyncio.wait_for function (as Python Synchronization Primitives recommendations).
This issue is not only to get "try lock" option, but anyone that call my code can use asyncio.wait_for (somewhere in stack chain) or cancel my task and break the lock.

Root Cause

1. self._w_state increased only in _wake_up function (if not own already).
2. But on cancellation we decrease it without increase it before(due to the CancelledError).
   And now self._w_state is -1 !
   For some reason in the CancelledError catch we assume that we increased it.
3. In the next wake-up the if self._r_state == 0 and self._w_state == 0 is false because self._w_state == -1

This lock can't recover from this illegal state and no one can acquire the lock anymore.

Test example

This test stuck

@pytest.mark.asyncio()
async def test_rw_lock_illegal_state_on_cancellation() -> None:

    async def reader_worker(t_id: int, lock: Union[ReadLock, _ReaderLock], acquired_event: Optional[Event] = None, release_event: Optional[Event] = None) -> None:
        print(f"reader-{t_id} start")
        async with lock:
            if acquired_event:
                acquired_event.set()

            if release_event:
                await release_event.wait()

        print(f"reader-{t_id} end")

    async def writer_worker(t_id: int, lock: Union[WriteLock, _WriterLock]) -> None:
        print(f"writer-{t_id} start")
        async with lock:
            print(f"writer-{t_id} acquired lock")
        print(f"writer-{t_id} end")

    rw_lock = RWLock()
    el = asyncio.get_event_loop()

    reader_acquired_event = Event()
    reader_release_event = Event()
    el.create_task(reader_worker(1, rw_lock.reader, acquired_event=reader_acquired_event, release_event=reader_release_event))
    await reader_acquired_event.wait()
    
    with pytest.raises(TimeoutError):   # Expect to fail on timeout because reader is holding the lock
        await asyncio.wait_for(writer_worker(1, lock=rw_lock.writer), 2)

    reader_release_event.set()  # This will cause the reader complete his work and release the lock

    print("after failure attempt to acquire...")
    async with rw_lock.reader:
        print("after failure => read lock acquired")

    async with rw_lock.writer:
        print("after failure => write lock acquired")

    print("done successfully")

Missing State Check After Future Waits: Concurrent Writers Possible

There is no check for the current state awakening for await fut in either the acquire_read or acquire_write methods.
The code for acquire_writer is below.

fut = self._loop.create_future()
self._write_waiters.append(fut)
try:
    await fut
    self._w_state += 1
    self._owning.append((me, self._WL))
    return True

Because of this, it is possible to end up in a bad state. Specifically, I noticed a problem with concurrent writers being able to write at the same time.
Here is some code to demonstrate.

import asyncio

import aiorwlock


async def work(lock):
	print(asyncio.current_task().get_name(), 'acquiring read lock')
	async with lock.reader:
		print(asyncio.current_task().get_name(), 'reading...')
		await asyncio.sleep(5)
		print(asyncio.current_task().get_name(), 'reading...done')
	print(asyncio.current_task().get_name(), 'acquiring write lock')
	async with lock.writer:
		print(asyncio.current_task().get_name(), 'writing...')
		await asyncio.sleep(5)
		print(asyncio.current_task().get_name(), 'writing...done')


async def main():
	lock = aiorwlock.RWLock(fast=True)

	tasks = [
		asyncio.create_task(work(lock)),
		asyncio.create_task(work(lock)),
	]
	await asyncio.wait(tasks)


asyncio.run(main())

And the output...

Task-2 acquiring read lock
Task-2 reading...
Task-3 acquiring read lock
Task-3 reading...
Task-2 reading...done
Task-2 acquiring write lock
Task-3 reading...done      
Task-3 acquiring write lock
Task-3 writing...
Task-2 writing...
Task-3 writing...done
Task-2 writing...done

A very simple solution is a while loop in both methods. For example, in acquire_write ...

async def acquire_write(self) -> bool:
        me = _current_task(loop=self._loop)

        while True:

            if (me, self._WL) in self._owning:
                self._w_state += 1
                self._owning.append((me, self._WL))
                await self._yield_after_acquire(self._WL)
                return True
            elif (me, self._RL) in self._owning:
                if self._r_state > 0:
                    raise RuntimeError('Cannot upgrade RWLock from read to write')

            if self._r_state == 0 and self._w_state == 0:
                self._w_state += 1
                self._owning.append((me, self._WL))
                await self._yield_after_acquire(self._WL)
                return True

            fut = self._loop.create_future()
            self._write_waiters.append(fut)
            try:
                await fut
                continue

            except asyncio.CancelledError:
                self._wake_up()
                raise

            finally:
                self._write_waiters.remove(fut)

And then running the same test code, I get the following output...

Task-2 acquiring read lock
Task-2 reading...
Task-3 acquiring read lock
Task-3 reading...
Task-2 reading...done
Task-2 acquiring write lock
Task-3 reading...done      
Task-3 acquiring write lock
Task-3 writing...
Task-3 writing...done
Task-2 writing...
Task-2 writing...done

Multiple coroutines acquire locks and request locks

async with rwlock.writer

When I get the lock in the above way, when a coroutine does not release the lock, the other coroutines also get the lock.

Errors caused by no running event loop

I've been using the library for a while in various projects, and after recent python updates I'm now getting the same error in all of them:
"env/lib/python3.10/site-packages/aiorwlock/__init__.py", line 242, in __init__ loop = asyncio.get_running_loop() RuntimeError: no running event loop

I tested it on the example in the README of this repo and it gives a deprecation warning of the same thing:
python main.py main.py:19: DeprecationWarning: There is no current event loop

I'm using python v3.10.7, and aiorwlock v1.3.0 straight from pip

Thanks!

Dependabot couldn't authenticate with https://pypi.python.org/simple/

Dependabot couldn't authenticate with https://pypi.python.org/simple/.

You can provide authentication details in your Dependabot dashboard by clicking into the account menu (in the top right) and selecting 'Config variables'.

View the update logs.

Drop python 3.4 support and fix minimal version python3.5.3

Task should be accomplished with 3 PRs

Port tests to async/await
Port full library to async/await
Update changes.rst and release new version and update README, docs, travisci

Lock not properly acquired when using `asyncio.wait_for()`

When trying to use asyncio.wait_for() to add a timeout to acquire(), seems that it is not being properly acquired for the current task.

lock = RWLock()
try:
    await asyncio.wait_for(lock.writer_lock.acquire(), timeout=5)
    lock.writer_lock.release()
except asyncio.TimeoutError:
    pass

Traceback

Traceback (most recent call last):
  File "/home/marc/test/rwlock/venv/lib/python3.10/site-packages/aiorwlock/__init__.py", line 147, in _release
    self._owning.remove((me, lock_type))
ValueError: list.remove(x): x not in list

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/home/marc/test/rwlock/test.py", line 61, in <module>
    asyncio.run(main())
  File "/usr/lib/python3.10/asyncio/runners.py", line 44, in run
    return loop.run_until_complete(main)
  File "/usr/lib/python3.10/asyncio/base_events.py", line 649, in run_until_complete
    return future.result()
  File "/home/marc/test/rwlock/test.py", line 52, in main
    await asyncio.gather(
  File "/home/marc/test/rwlock/test.py", line 26, in access_object
    lock.reader_lock.release()
  File "/home/marc/test/rwlock/venv/lib/python3.10/site-packages/aiorwlock/__init__.py", line 217, in release
    self._lock.release_read()
  File "/home/marc/test/rwlock/venv/lib/python3.10/site-packages/aiorwlock/__init__.py", line 136, in release_read
    self._release(self._RL)
  File "/home/marc/test/rwlock/venv/lib/python3.10/site-packages/aiorwlock/__init__.py", line 149, in _release
    raise RuntimeError('Cannot release an un-acquired lock')
RuntimeError: Cannot release an un-acquired lock

This same script using python built-in async lock works

lock = asyncio.Lock()
try:
    await asyncio.wait_for(lock.acquire(), timeout=5)
    lock.release()
except asyncio.TimeoutError:
    pass

Use loop.create_future instead of asyncio.Future

Add uvloop to the test suite

Feature request: Add try_lock()

Pretty please, add try_lock() to read and write locks

Let's put `aiorwlock` under the `aio-libs` org on PyPI

See the details @ https://github.com/orgs/aio-libs/discussions/26.

For this, we'll need somebody with Owner privileges to either give me access on PyPI (the username there is the same — webknjaz) or be invited to the org by me (for that I'd need the username of such an individual).

As a bonus, this will also allow us to set up secretless publishing from GHA to PyPI and get rid of the in-repo secrets.

cc @jettify

aiorwlock tla+ spec is missing cfg file

Please add aiorwlock.cfg file for the aiorwlock.tla file

$ tlc aiorwlock.tla 
TLC2 Version 2.14 of 10 July 2019 (rev: 0cae24f)
...
The exception was a tlc2.tool.ConfigFileException
: TLC encountered the following error when trying to read the configuration file aiorwlock.cfg:
File not found.
...

Add type anotations

Update all `secret: ` values in `.travis.yml`

Do like this:

REPO_NAME={{ YOUR_REPO_NAME }}
travis encrypt -r "aio-libs/${REPO_NAME}" --api-endpoint 'https://api.travis-ci.com/'

Ref: https://github.com/orgs/aio-libs/teams/admins/discussions/9