synchronization primitives all lock event condition semaphore boundedsemaphore barrier import collections import enum from import exceptions from import mixins class contextmanagermixin async def aenterself await self acquire we have no use for the as clause in the with statement for locks return none async def aexitself exctype exc tb self release class lockcontextmanagermixin mixins loopboundmixin lock is acquired def initself self waiters none self locked false def reprself res super repr extra locked if self locked else unlocked if self waiters extra f extra waiters lenself waiters return f res1 1 extra def lockedself acquire a lock this method blocks until the lock is unlocked then sets it to locked and returns true finally block should be called before the cancellederror handling as we don t want cancellederror to call wakeupfirst and attempt to wake up itself release a lock when the lock is locked reset it to unlocked and return if any other coroutines are blocked waiting for the lock to become unlocked allow exactly one of them to proceed when invoked on an unlocked lock a runtimeerror is raised there is no return value wake up the first waiter if it isn t done if not self waiters return try fut nextiterself waiters except stopiteration return done necessarily means that a waiter will wake up later on and either take the lock or if it was cancelled and lock wasn t taken already will hit this again and wake up a new waiter if not fut done fut setresulttrue class eventmixins loopboundmixin def initself self waiters collections deque self value false def reprself res super repr extra set if self value else unset if self waiters extra f extra waiters lenself waiters return f res1 1 extra def issetself set the internal flag to true all coroutines waiting for it to become true are awakened coroutine that call wait once the flag is true will not block at all reset the internal flag to false subsequently coroutines calling wait will block until set is called to set the internal flag to true again self value false async def waitself if self value return true fut self getloop createfuture self waiters appendfut try await fut return true finally self waiters removefut class conditioncontextmanagermixin mixins loopboundmixin def initself locknone if lock is none lock lock self lock lock export the lock s locked acquire and release methods self locked lock locked self acquire lock acquire self release lock release self waiters collections deque def reprself res super repr extra locked if self locked else unlocked if self waiters extra f extra waiters lenself waiters return f res1 1 extra async def waitself if not self locked raise runtimeerror cannot wait on unacquired lock self release try fut self getloop createfuture self waiters appendfut try await fut return true finally self waiters removefut finally must reacquire lock even if wait is cancelled cancelled false while true try await self acquire break except exceptions cancellederror cancelled true if cancelled raise exceptions cancellederror async def waitforself predicate result predicate while not result await self wait result predicate return result def notifyself n1 if not self locked raise runtimeerror cannot notify on unacquired lock idx 0 for fut in self waiters if idx n break if not fut done idx 1 fut setresultfalse def notifyallself self notifylenself waiters class semaphorecontextmanagermixin mixins loopboundmixin def initself value1 if value 0 raise valueerrorsemaphore initial value must be 0 self waiters none self value value def reprself res super repr extra locked if self locked else f unlocked value self value if self waiters extra f extra waiters lenself waiters return f res1 1 extra def lockedself acquire a semaphore if the internal counter is larger than zero on entry decrement it by one and return true immediately if it is zero on entry block waiting until some other coroutine has called release to make it larger than 0 and then return true finally block should be called before the cancellederror handling as we don t want cancellederror to call wakeupfirst and attempt to wake up itself release a semaphore incrementing the internal counter by one when it was zero on entry and another coroutine is waiting for it to become larger than zero again wake up that coroutine wake up the first waiter that isn t done if not self waiters return for fut in self waiters if not fut done self value 1 fut setresulttrue return class boundedsemaphoresemaphore def initself value1 self boundvalue value super initvalue def releaseself if self value self boundvalue raise valueerror boundedsemaphore released too many times super release class barrierstateenum enum filling filling draining draining resetting resetting broken broken class barriermixins loopboundmixin def initself parties wait for the barrier reaches the parties number when start draining release and return index of waited task wait for the barrier when the specified number of tasks have started waiting they are all simultaneously awoken returns an unique and individual index number from 0 to parties1 we release the barrier wake up any tasks waiting for barrier to drain block until the barrier is ready for us or raise an exception if it is broken it is draining or resetting wait until done unless a cancellederror occurs see if the barrier is in a broken state release the tasks waiting in the barrier enter draining state next waiting tasks will be blocked until the end of draining wait in the barrier until we are released raise an exception if the barrier is reset or broken wait for end of filling unless a cancellederror occurs if we are the last tasks to exit the barrier signal any tasks waiting for the barrier to drain reset the barrier to the initial state any tasks currently waiting will get the brokenbarrier exception raised reset the barrier waking up tasks place the barrier into a broken state useful in case of error any currently waiting tasks and tasks attempting to wait will have brokenbarriererror raised return the number of tasks required to trip the barrier return self parties property def nwaitingself return true if the barrier is in a broken state return self state is barrierstate broken we have no use for the as clause in the with statement for locks primitive lock objects a primitive lock is a synchronization primitive that is not owned by a particular coroutine when locked a primitive lock is in one of two states locked or unlocked it is created in the unlocked state it has two basic methods acquire and release when the state is unlocked acquire changes the state to locked and returns immediately when the state is locked acquire blocks until a call to release in another coroutine changes it to unlocked then the acquire call resets it to locked and returns the release method should only be called in the locked state it changes the state to unlocked and returns immediately if an attempt is made to release an unlocked lock a runtimeerror will be raised when more than one coroutine is blocked in acquire waiting for the state to turn to unlocked only one coroutine proceeds when a release call resets the state to unlocked first coroutine which is blocked in acquire is being processed acquire is a coroutine and should be called with await locks also support the asynchronous context management protocol async with lock statement should be used usage lock lock await lock acquire try finally lock release context manager usage lock lock async with lock lock objects can be tested for locking state if not lock locked await lock acquire else lock is acquired return true if lock is acquired acquire a lock this method blocks until the lock is unlocked then sets it to locked and returns true finally block should be called before the cancellederror handling as we don t want cancellederror to call _wake_up_first and attempt to wake up itself release a lock when the lock is locked reset it to unlocked and return if any other coroutines are blocked waiting for the lock to become unlocked allow exactly one of them to proceed when invoked on an unlocked lock a runtimeerror is raised there is no return value wake up the first waiter if it isn t done done necessarily means that a waiter will wake up later on and either take the lock or if it was cancelled and lock wasn t taken already will hit this again and wake up a new waiter asynchronous equivalent to threading event class implementing event objects an event manages a flag that can be set to true with the set method and reset to false with the clear method the wait method blocks until the flag is true the flag is initially false return true if and only if the internal flag is true set the internal flag to true all coroutines waiting for it to become true are awakened coroutine that call wait once the flag is true will not block at all reset the internal flag to false subsequently coroutines calling wait will block until set is called to set the internal flag to true again block until the internal flag is true if the internal flag is true on entry return true immediately otherwise block until another coroutine calls set to set the flag to true then return true asynchronous equivalent to threading condition this class implements condition variable objects a condition variable allows one or more coroutines to wait until they are notified by another coroutine a new lock object is created and used as the underlying lock export the lock s locked acquire and release methods wait until notified if the calling coroutine has not acquired the lock when this method is called a runtimeerror is raised this method releases the underlying lock and then blocks until it is awakened by a notify or notify_all call for the same condition variable in another coroutine once awakened it re acquires the lock and returns true must reacquire lock even if wait is cancelled wait until a predicate becomes true the predicate should be a callable which result will be interpreted as a boolean value the final predicate value is the return value by default wake up one coroutine waiting on this condition if any if the calling coroutine has not acquired the lock when this method is called a runtimeerror is raised this method wakes up at most n of the coroutines waiting for the condition variable it is a no op if no coroutines are waiting note an awakened coroutine does not actually return from its wait call until it can reacquire the lock since notify does not release the lock its caller should wake up all threads waiting on this condition this method acts like notify but wakes up all waiting threads instead of one if the calling thread has not acquired the lock when this method is called a runtimeerror is raised a semaphore implementation a semaphore manages an internal counter which is decremented by each acquire call and incremented by each release call the counter can never go below zero when acquire finds that it is zero it blocks waiting until some other thread calls release semaphores also support the context management protocol the optional argument gives the initial value for the internal counter it defaults to 1 if the value given is less than 0 valueerror is raised returns true if semaphore cannot be acquired immediately acquire a semaphore if the internal counter is larger than zero on entry decrement it by one and return true immediately if it is zero on entry block waiting until some other coroutine has called release to make it larger than 0 and then return true finally block should be called before the cancellederror handling as we don t want cancellederror to call _wake_up_first and attempt to wake up itself release a semaphore incrementing the internal counter by one when it was zero on entry and another coroutine is waiting for it to become larger than zero again wake up that coroutine wake up the first waiter that isn t done a bounded semaphore implementation this raises valueerror in release if it would increase the value above the initial value asyncio equivalent to threading barrier implements a barrier primitive useful for synchronizing a fixed number of tasks at known synchronization points tasks block on wait and are simultaneously awoken once they have all made their call create a barrier initialised to parties tasks notify all tasks when state changes count tasks in barrier wait for the barrier reaches the parties number when start draining release and return index of waited task wait for the barrier when the specified number of tasks have started waiting they are all simultaneously awoken returns an unique and individual index number from 0 to parties 1 block while the barrier drains or resets we release the barrier wake up any tasks waiting for barrier to drain block until the barrier is ready for us or raise an exception if it is broken it is draining or resetting wait until done unless a cancellederror occurs see if the barrier is in a broken state release the tasks waiting in the barrier enter draining state next waiting tasks will be blocked until the end of draining wait in the barrier until we are released raise an exception if the barrier is reset or broken wait for end of filling unless a cancellederror occurs if we are the last tasks to exit the barrier signal any tasks waiting for the barrier to drain reset the barrier to the initial state any tasks currently waiting will get the brokenbarrier exception raised reset the barrier waking up tasks place the barrier into a broken state useful in case of error any currently waiting tasks and tasks attempting to wait will have brokenbarriererror raised return the number of tasks required to trip the barrier return the number of tasks currently waiting at the barrier return true if the barrier is in a broken state
__all__ = ('Lock', 'Event', 'Condition', 'Semaphore',
           'BoundedSemaphore', 'Barrier')

import collections
import enum

from . import exceptions
from . import mixins

class _ContextManagerMixin:
    async def __aenter__(self):
        await self.acquire()
        return None

    async def __aexit__(self, exc_type, exc, tb):
        self.release()


class Lock(_ContextManagerMixin, mixins._LoopBoundMixin):
    

    def __init__(self):
        self._waiters = None
        self._locked = False

    def __repr__(self):
        res = super().__repr__()
        extra = 'locked' if self._locked else 'unlocked'
        if self._waiters:
            extra = f'{extra}, waiters:{len(self._waiters)}'
        return f'<{res[1:-1]} [{extra}]>'

    def locked(self):
        
        return self._locked

    async def acquire(self):
        
        if (not self._locked and (self._waiters is None or
                all(w.cancelled() for w in self._waiters))):
            self._locked = True
            return True

        if self._waiters is None:
            self._waiters = collections.deque()
        fut = self._get_loop().create_future()
        self._waiters.append(fut)

        
        
        
        try:
            try:
                await fut
            finally:
                self._waiters.remove(fut)
        except exceptions.CancelledError:
            if not self._locked:
                self._wake_up_first()
            raise

        self._locked = True
        return True

    def release(self):
        
        if self._locked:
            self._locked = False
            self._wake_up_first()
        else:
            raise RuntimeError('Lock is not acquired.')

    def _wake_up_first(self):
        
        if not self._waiters:
            return
        try:
            fut = next(iter(self._waiters))
        except StopIteration:
            return

    
        if not fut.done():
            fut.set_result(True)


class Event(mixins._LoopBoundMixin):
    

    def __init__(self):
        self._waiters = collections.deque()
        self._value = False

    def __repr__(self):
        res = super().__repr__()
        extra = 'set' if self._value else 'unset'
        if self._waiters:
            extra = f'{extra}, waiters:{len(self._waiters)}'
        return f'<{res[1:-1]} [{extra}]>'

    def is_set(self):
        
        return self._value

    def set(self):
        
        if not self._value:
            self._value = True

            for fut in self._waiters:
                if not fut.done():
                    fut.set_result(True)

    def clear(self):
        
        self._value = False

    async def wait(self):
        
        if self._value:
            return True

        fut = self._get_loop().create_future()
        self._waiters.append(fut)
        try:
            await fut
            return True
        finally:
            self._waiters.remove(fut)


class Condition(_ContextManagerMixin, mixins._LoopBoundMixin):
    

    def __init__(self, lock=None):
        if lock is None:
            lock = Lock()

        self._lock = lock
        
        self.locked = lock.locked
        self.acquire = lock.acquire
        self.release = lock.release

        self._waiters = collections.deque()

    def __repr__(self):
        res = super().__repr__()
        extra = 'locked' if self.locked() else 'unlocked'
        if self._waiters:
            extra = f'{extra}, waiters:{len(self._waiters)}'
        return f'<{res[1:-1]} [{extra}]>'

    async def wait(self):
        
        if not self.locked():
            raise RuntimeError('cannot wait on un-acquired lock')

        self.release()
        try:
            fut = self._get_loop().create_future()
            self._waiters.append(fut)
            try:
                await fut
                return True
            finally:
                self._waiters.remove(fut)

        finally:
            
            cancelled = False
            while True:
                try:
                    await self.acquire()
                    break
                except exceptions.CancelledError:
                    cancelled = True

            if cancelled:
                raise exceptions.CancelledError

    async def wait_for(self, predicate):
        
        result = predicate()
        while not result:
            await self.wait()
            result = predicate()
        return result

    def notify(self, n=1):
        
        if not self.locked():
            raise RuntimeError('cannot notify on un-acquired lock')

        idx = 0
        for fut in self._waiters:
            if idx >= n:
                break

            if not fut.done():
                idx += 1
                fut.set_result(False)

    def notify_all(self):
        
        self.notify(len(self._waiters))


class Semaphore(_ContextManagerMixin, mixins._LoopBoundMixin):
    

    def __init__(self, value=1):
        if value < 0:
            raise ValueError("Semaphore initial value must be >= 0")
        self._waiters = None
        self._value = value

    def __repr__(self):
        res = super().__repr__()
        extra = 'locked' if self.locked() else f'unlocked, value:{self._value}'
        if self._waiters:
            extra = f'{extra}, waiters:{len(self._waiters)}'
        return f'<{res[1:-1]} [{extra}]>'

    def locked(self):
        
        return self._value == 0 or (
            any(not w.cancelled() for w in (self._waiters or ())))

    async def acquire(self):
        
        if not self.locked():
            self._value -= 1
            return True

        if self._waiters is None:
            self._waiters = collections.deque()
        fut = self._get_loop().create_future()
        self._waiters.append(fut)

        
        
        
        try:
            try:
                await fut
            finally:
                self._waiters.remove(fut)
        except exceptions.CancelledError:
            if not fut.cancelled():
                self._value += 1
                self._wake_up_next()
            raise

        if self._value > 0:
            self._wake_up_next()
        return True

    def release(self):
        
        self._value += 1
        self._wake_up_next()

    def _wake_up_next(self):
        
        if not self._waiters:
            return

        for fut in self._waiters:
            if not fut.done():
                self._value -= 1
                fut.set_result(True)
                return


class BoundedSemaphore(Semaphore):
    

    def __init__(self, value=1):
        self._bound_value = value
        super().__init__(value)

    def release(self):
        if self._value >= self._bound_value:
            raise ValueError('BoundedSemaphore released too many times')
        super().release()



class _BarrierState(enum.Enum):
    FILLING = 'filling'
    DRAINING = 'draining'
    RESETTING = 'resetting'
    BROKEN = 'broken'


class Barrier(mixins._LoopBoundMixin):
    

    def __init__(self, parties):
        
        if parties < 1:
            raise ValueError('parties must be > 0')

        self._cond = Condition() 

        self._parties = parties
        self._state = _BarrierState.FILLING
        self._count = 0       

    def __repr__(self):
        res = super().__repr__()
        extra = f'{self._state.value}'
        if not self.broken:
            extra += f', waiters:{self.n_waiting}/{self.parties}'
        return f'<{res[1:-1]} [{extra}]>'

    async def __aenter__(self):
        
        
        return await self.wait()

    async def __aexit__(self, *args):
        pass

    async def wait(self):
        
        async with self._cond:
            await self._block() 
            try:
                index = self._count
                self._count += 1
                if index + 1 == self._parties:
                    
                    await self._release()
                else:
                    await self._wait()
                return index
            finally:
                self._count -= 1
                
                self._exit()

    async def _block(self):
        
        
        
        
        
        await self._cond.wait_for(
            lambda: self._state not in (
                _BarrierState.DRAINING, _BarrierState.RESETTING
            )
        )

        
        if self._state is _BarrierState.BROKEN:
            raise exceptions.BrokenBarrierError("Barrier aborted")

    async def _release(self):
        

        
        
        self._state = _BarrierState.DRAINING
        self._cond.notify_all()

    async def _wait(self):
        
        

        
        
        await self._cond.wait_for(lambda: self._state is not _BarrierState.FILLING)

        if self._state in (_BarrierState.BROKEN, _BarrierState.RESETTING):
            raise exceptions.BrokenBarrierError("Abort or reset of barrier")

    def _exit(self):
        
        
        if self._count == 0:
            if self._state in (_BarrierState.RESETTING, _BarrierState.DRAINING):
                self._state = _BarrierState.FILLING
            self._cond.notify_all()

    async def reset(self):
        
        async with self._cond:
            if self._count > 0:
                if self._state is not _BarrierState.RESETTING:
                    
                    self._state = _BarrierState.RESETTING
            else:
                self._state = _BarrierState.FILLING
            self._cond.notify_all()

    async def abort(self):
        
        async with self._cond:
            self._state = _BarrierState.BROKEN
            self._cond.notify_all()

    @property
    def parties(self):
        
        return self._parties

    @property
    def n_waiting(self):
        
        if self._state is _BarrierState.FILLING:
            return self._count
        return 0

    @property
    def broken(self):
        
        return self._state is _BarrierState.BROKEN