[Async-sig] "read-write" synchronization

[Chris Jerdonek]

On Mon, Jun 26, 2017 at 12:37 PM, Dima Tisnek <dimaqq at gmail.com> wrote:
> Chris, here's a simple RWLock implementation and analysis:
> ...
> Obv., this code could be nicer:
> * separate context managers for read and write cases
> * .unlock can be automatic (if self.writer: unlock_for_write()) at the
> cost of opening doors wide open to bugs
> * policy can be introduced if `.lock` identified itself (by an
> object(), since there's no thread id) in shared state
> * notifyAll() makes real life use O(N^2) for N being number of
> simultaneous write lock requests
>
> Feel free to use it :)

Thanks, Dima. However, as I said in my earlier posts, I'm actually
more interested in exploring approaches to synchronizing readers and
writers in async code that don't require locking on reads. (This is
also why I've always been saying RW "synchronization" instead of RW
"locking.")

I'm interested in this because I think the single-threadedness of the
event loop might be what makes this simplification possible over the
traditional multi-threaded approach (along the lines Guido was
mentioning). It also makes the "fast path" faster. Lastly, the API for
the callers is just to call read() or write(), so there is no need for
a general RWLock construct or to work through RWLock semantics of the
sort Nathaniel mentioned.

I coded up a working version of the pseudo-code I included in an
earlier email so people can see how it works. I included it at the
bottom of this email and also in this gist:
https://gist.github.com/cjerdonek/858e1467f768ee045849ea81ddb47901

--Chris


import asyncio
import random


NO_READERS_EVENT = asyncio.Event()
NO_WRITERS_EVENT = asyncio.Event()
WRITE_LOCK = asyncio.Lock()


class State:
    reader_count = 0
    mock_file_data = 'initial'


async def read_file():
    data = State.mock_file_data
    print(f'read: {data}')


async def write_file(data):
    print(f'writing: {data}')
    State.mock_file_data = data
    await asyncio.sleep(0.5)


async def write(data):
    async with WRITE_LOCK:
        NO_WRITERS_EVENT.clear()
        # Wait for the readers to finish.
        await NO_READERS_EVENT.wait()
        # Do the file write.
        await write_file(data)

    # Awaken waiting readers.
    NO_WRITERS_EVENT.set()


async def read():
    while True:
        await NO_WRITERS_EVENT.wait()
        # Check the writer_lock again in case a new writer has
        # started writing.
        if WRITE_LOCK.locked():
            print(f'cannot read: still writing: {State.mock_file_data!r}')
        else:
            # Otherwise, we can do the read.
            break

    State.reader_count += 1
    if State.reader_count == 1:
        NO_READERS_EVENT.clear()
    # Do the file read.
    await read_file()
    State.reader_count -= 1
    if State.reader_count == 0:
        # Awaken any waiting writer.
        NO_READERS_EVENT.set()


async def delayed(coro):
    await asyncio.sleep(random.random())
    await coro


async def test_synchronization():
    NO_READERS_EVENT.set()
    NO_WRITERS_EVENT.set()

    coros = [
        read(),
        read(),
        read(),
        read(),
        read(),
        read(),
        write('apple'),
        write('banana'),
    ]
    # Add a delay before each coroutine for variety.
    coros = [delayed(coro) for coro in coros]
    await asyncio.gather(*coros)


if __name__ == '__main__':
    asyncio.get_event_loop().run_until_complete(test_synchronization())

# Sample output:
#
# read: initial
# read: initial
# read: initial
# read: initial
# writing: banana
# writing: apple
# cannot read: still writing: 'apple'
# cannot read: still writing: 'apple'
# read: apple
# read: apple