Pasted

from contextlib import contextmanager
from threading import Condition, Thread
import time
 
class SharedLock:
    def __init__(self):
        self._cond = Condition()
        self._readers = 0
 
    @contextmanager
    def __call__(self, lock_type: str):
        if lock_type == 'shared':
            with self._cond:
                self._readers += 1
            yield self
            with self._cond:
                self._readers -= 1
                if not self._readers:
                    self._cond.notify()
        elif lock_type == 'exclusive':
            with self._cond:
                while self._readers:
                    self._cond.wait()
                yield self
        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()