rwlock write-priority

from contextlib import contextmanager
from threading import Condition, Thread, Lock
import time

class SharedLock:
    def __init__(self):
        self._lock = Lock()
        self._readcond = Condition(self._lock)
        self._writecond = Condition(self._lock)
        self._readers = 0
        self._writers_waiting = 0

    @contextmanager
    def __call__(self, lock_type: str):
        if lock_type == 'shared':
            with self._lock:
                while self._writers_waiting:
                    self._readcond.wait()
                self._readers += 1
            yield self
            with self._lock:
                self._readers -= 1
                if not self._readers:
                    self._writecond.notify()
        elif lock_type == 'exclusive':
            with self._lock:
                self._writers_waiting += 1
                while self._readers:
                    self._writecond.wait()
                self._writers_waiting -= 1
                yield self
                if self._writers_waiting:
                    self._writecond.notify()
                else:
                    self._readcond.notify_all()
        else:
            raise ValueError(f'Invalid lock_type {lock_type!r}')

if __name__ == '__main__':
    shared_lock = SharedLock()

    def reader():
        with shared_lock('shared'):
            print('start reading', time.time())
            time.sleep(1)
            print('end reading', time.time())

        # Give writer thread a chance to acquire exclusive use:
        #time.sleep(.1)

        #with shared_lock('shared'):
        #    print('start reading', time.time())
        #    time.sleep(1)
        #    print('end reading', time.time())

    def writer():
        # Give reader threads a chance to acquire shared use:
        time.sleep(.1)

        with shared_lock('exclusive'):
            print('start writing', time.time())
            time.sleep(1)
            print('end writing', time.time())

    def test():
        reader_threads = [
            Thread(target=reader) for _ in range(1)
        ]

        writer_threads = [
            Thread(target=writer) for _ in range(2)
        ]

        for t in reader_threads:
            t.start()
        for t in writer_threads:
            t.start()

        for t in reader_threads:
            t.join()
        for t in writer_threads:
            t.join()

    test()