o

we�C�@s�dd
lZdd
l
Z
dd
lZddlmZmZ
ddlmZmZ
ddl	m
Z
mZmZm
Z
mZ
dd
l	Z	e	jr9ddl	mZmZ
gd�
ZGdd�de�ZGd	d
�d
e�ZGdd�de�ZGd
d�de�ZGdd�de�ZGdd�de�ZGdd�de�Zd
S)�N)�gen�ioloop)�Future�"future_set_result_unless_cancelled)�Union�Optional�Type�Any�	Awaitable)�Deque�Set)�	Condition�Event�	Semaphore�BoundedSemaphore�Lockc@s$eZ
dZd
Zddd�Zddd�ZdS)	�_TimeoutGarbageCollectorz�Base class for objects that periodically clean up timed-out waiters.

    Avoids memory leak in a common pattern like:

        while True:
            yield condition.wait(short_timeout)
            print('looping....')
    �returnNc

Cst�
�|_d
|_dS)Nr
)�collections�deque�_waiters�	_timeouts�
�self�r�F/home/arjun/projects/env/lib/python3.10/site-packages/tornado/locks.py�__init__)s



z!_TimeoutGarbageCollector.__init__c

Cs>|jd
7_|jdkrd|_t
�dd�|jD�
�
|_dSdS)N�
�dr
c
ss�|]	}
|
��s|
V
qdS�
N)
�done)�.0�
wrrr�	<genexpr>2s�z<_TimeoutGarbageCollector._garbage_collect.<locals>.<genexpr>)rrrrrrrr�_garbage_collect-s




�z)_TimeoutGarbageCollector._garbage_collect�rN)�__name__�
__module__�__qualname__�__doc__rr$rrrrrs

	rc@sdeZ
dZd
Zdefdd�Z	ddeeee	j
fdeefdd�Z
dd
eddfdd�Zdd
d�ZdS)r
a�A condition allows one or more coroutines to wait until notified.

    Like a standard `threading.Condition`, but does not need an underlying lock
    that is acquired and released.

    With a `Condition`, coroutines can wait to be notified by other coroutines:

    .. testcode::

        import asyncio
        from tornado import gen
        from tornado.locks import Condition

        condition = Condition()

        async def waiter():
            print("I'll wait right here")
            await condition.wait()
            print("I'm done waiting")

        async def notifier():
            print("About to notify")
            condition.notify()
            print("Done notifying")

        async def runner():
            # Wait for waiter() and notifier() in parallel
            await gen.multi([waiter(), notifier()])

        asyncio.run(runner())

    .. testoutput::

        I'll wait right here
        About to notify
        Done notifying
        I'm done waiting

    `wait` takes an optional ``timeout`` argument, which is either an absolute
    timestamp::

        io_loop = IOLoop.current()

        # Wait up to 1 second for a notification.
        await condition.wait(timeout=io_loop.time() + 1)

    ...or a `datetime.timedelta` for a timeout relative to the current time::

        # Wait up to 1 second.
        await condition.wait(timeout=datetime.timedelta(seconds=1))

    The method returns False if there's no notification before the deadline.

    .. versionchanged:: 5.0
       Previously, waiters could be notified synchronously from within
       `notify`. Now, the notification will always be received on the
       next iteration of the `.IOLoop`.
    rc
Cs.d
|jj
f
}
|jr|
dt|j�
7}
|
dS)Nz<%sz waiters[%s]�
>)�	__class__r&r�len)r�resultrrr�__repr__qs



zCondition.__repr__N�timeoutcsTt���
j
���

|
r(d�
�fdd�}tj�����|
|������fdd��

�S)z�Wait for `.notify`.

        Returns a `.Future` that resolves ``True`` if the condition is notified,
        or ``False`` after a timeout.
        rNcs�
��s	t
�
d
�
���
dS�NF)r rr$r�r�waiterrr�
on_timeout�s



z"Condition.wait.<locals>.on_timeoutc

�
���
�
Sr�
�remove_timeout�
�
_��io_loop�timeout_handlerr�<lambda>��
z Condition.wait.<locals>.<lambda>r%)rr�appendr�IOLoop�current�add_timeout�add_done_callback�rr/r3r�r:rr;r2r�waitws





zCondition.waitr�
ncCsTg}|
r|jr|j�
�}|��s|
d
8}
|�|�

|
r|js|D]}t|d�
q dS)zWake ``n`` waiters.rTN)r�popleftr r>r)rrF�waitersr2rrr�notify�s








�
�zCondition.notifyc

Cs|�t
|j�
�

d
S)zWake all waiters.N)rIr,rrrrr�
notify_all�szCondition.notify_allr�
rr%)r&r'r(r)�strr.rr�float�datetime�	timedeltar
�boolrE�intrIrJrrrrr
5s
;�
�
�r
c@sreZ
dZd
Zddd�Zdefdd�Zdefdd	�Zdd
d�Z	ddd
�Z
	ddeee
ejfdedfdd�ZdS)ra�An event blocks coroutines until its internal flag is set to True.

    Similar to `threading.Event`.

    A coroutine can wait for an event to be set. Once it is set, calls to
    ``yield event.wait()`` will not block unless the event has been cleared:

    .. testcode::

        import asyncio
        from tornado import gen
        from tornado.locks import Event

        event = Event()

        async def waiter():
            print("Waiting for event")
            await event.wait()
            print("Not waiting this time")
            await event.wait()
            print("Done")

        async def setter():
            print("About to set the event")
            event.set()

        async def runner():
            await gen.multi([waiter(), setter()])

        asyncio.run(runner())

    .. testoutput::

        Waiting for event
        About to set the event
        Not waiting this time
        Done
    rNc

Csd
|_t
�|_dSr0)�_value�setrrrrrr�s

zEvent.__init__c

Cs d
|jj
|��rdfSdfS)Nz<%s %s>rS�clear)r+r&�is_setrrrrr.�s



��zEvent.__repr__c


Cs|jS)
z-Return ``True`` if the internal flag is true.�
rRrrrrrU�szEvent.is_setc
Cs2|jsd
|_|j
D]
}
|
��s|
�d�

q	dSdS)z�Set the internal flag to ``True``. All waiters are awakened.

        Calling `.wait` once the flag is set will not block.
        TN)rRrr �
set_result)r�futrrrrS�s




��z	Event.setc

Cs
d
|_dS)zkReset the internal flag to ``False``.

        Calls to `.wait` will block until `.set` is called.
        FNrVrrrrrT�s
zEvent.clearr/csft���
j
r
��d
�

�S�
j���

���
f
dd��

|
d
ur"�St�|
��}|��f
dd��

|S)z�Block until the internal flag is true.

        Returns an awaitable, which raises `tornado.util.TimeoutError` after a
        timeout.
        Nc

s�j�
|�
Sr)r�remove�
rXrrrr<�szEvent.wait.<locals>.<lambda>c

s���s��
�SdSr)r �cancel)
�tfrZrrr<�s)rrRrWr�addrBr�with_timeout)rr/�timeout_futr)rXrrrE�s









�z
Event.waitr%r)r&r'r(r)rrLr.rPrUrSrTrrrMrNrOr
rErrrrr�s

'

�
��rc@sPeZ
dZd
Zdeddfdd�Zddd�Zd	d
deedee	j
ddfd
d�ZdS)�_ReleasingContextManagerz�Releases a Lock or Semaphore at the end of a "with" statement.

    with (yield semaphore.acquire()):
        pass

    # Now semaphore.release() has been called.
    �objrNcCs
|
|_dSr)
�_obj)rrarrrr
s

z!_ReleasingContextManager.__init__c


CsdSrrrrrr�	__enter__
s
z"_ReleasingContextManager.__enter__�exc_type�Optional[Type[BaseException]]�exc_val�exc_tbcCs|j�
�
dSr)rb�release)rrdrfrgrrr�__exit__
sz!_ReleasingContextManager.__exit__r%)r&r'r(r)r	rrcr�
BaseException�types�
TracebackTyperirrrrr`
s

����r`cs�eZ
dZd
Zddeddf�f
dd�
Zdef�f
dd	�Zdd
d�Z	dde	e
eej
fdeefd
d�Zddd�Zddde	ede	ejddfdd�Zddd�Zddde	ede	ejddfdd�Z�ZS)ra�A lock that can be acquired a fixed number of times before blocking.

    A Semaphore manages a counter representing the number of `.release` calls
    minus the number of `.acquire` calls, plus an initial value. The `.acquire`
    method blocks if necessary until it can return without making the counter
    negative.

    Semaphores limit access to a shared resource. To allow access for two
    workers at a time:

    .. testsetup:: semaphore

       from collections import deque

       from tornado import gen
       from tornado.ioloop import IOLoop
       from tornado.concurrent import Future

       inited = False

       async def simulator(futures):
           for f in futures:
               # simulate the asynchronous passage of time
               await gen.sleep(0)
               await gen.sleep(0)
               f.set_result(None)

       def use_some_resource():
           global inited
           global futures_q
           if not inited:
               inited = True
               # Ensure reliable doctest output: resolve Futures one at a time.
               futures_q = deque([Future() for _ in range(3)])
               IOLoop.current().add_callback(simulator, list(futures_q))

           return futures_q.popleft()

    .. testcode:: semaphore

        import asyncio
        from tornado import gen
        from tornado.locks import Semaphore

        sem = Semaphore(2)

        async def worker(worker_id):
            await sem.acquire()
            try:
                print("Worker %d is working" % worker_id)
                await use_some_resource()
            finally:
                print("Worker %d is done" % worker_id)
                sem.release()

        async def runner():
            # Join all workers.
            await gen.multi([worker(i) for i in range(3)])

        asyncio.run(runner())

    .. testoutput:: semaphore

        Worker 0 is working
        Worker 1 is working
        Worker 0 is done
        Worker 2 is working
        Worker 1 is done
        Worker 2 is done

    Workers 0 and 1 are allowed to run concurrently, but worker 2 waits until
    the semaphore has been released once, by worker 0.

    The semaphore can be used as an async context manager::

        async def worker(worker_id):
            async with sem:
                print("Worker %d is working" % worker_id)
                await use_some_resource()

            # Now the semaphore has been released.
            print("Worker %d is done" % worker_id)

    For compatibility with older versions of Python, `.acquire` is a
    context manager, so ``worker`` could also be written as::

        @gen.coroutine
        def worker(worker_id):
            with (yield sem.acquire()):
                print("Worker %d is working" % worker_id)
                yield use_some_resource()

            # Now the semaphore has been released.
            print("Worker %d is done" % worker_id)

    .. versionchanged:: 4.3
       Added ``async with`` support in Python 3.5.

    r�valuerNcs$t��
�
|
d
kr
td�
�
|
|_dS)Nr
z$semaphore initial value must be >= 0)�superr�
ValueErrorrR�rrm�
r+rrr
s




zSemaphore.__init__c
sPt��
�}
|jd
krdnd�|j�
}|jrd�|t|j�
�}d�|
dd�|�S)Nr
�lockedzunlocked,value:{0}z{0},waiters:{1}z<{0} [{1}]>r���)rnr.rR�formatrr,)r�res�extrarqrrr.�
s

�

zSemaphore.__repr__c
CsT|jd
7_|j
r(|j
��}
|
��s#|jd
8_|
�t|�
�

dS|j
s
dSdS)�*Increment the counter and wake one waiter.rN)rRrrGr rWr`r1rrrrh�
s





�zSemaphore.releaser/cs~t���
j
d
kr�
j
d8_
��t�
�
�

�S�
j���

|
r=d	�
�fdd�}tj�����	|
|����
��fdd��

�S)
z�Decrement the counter. Returns an awaitable.

        Block if the counter is zero and wait for a `.release`. The awaitable
        raises `.TimeoutError` after the deadline.
        r
rrNcs"�
��s�
�
t���

���
dSr)r �
set_exceptionr�TimeoutErrorr$rr1rrr3�
s


z%Semaphore.acquire.<locals>.on_timeoutc

r4rr5r7r9rrr<�
r=z#Semaphore.acquire.<locals>.<lambda>r%)rrRrWr`rr>rr?r@rArBrCrrDr�acquire�
s



�




�zSemaphore.acquirec

C�td
�
�
)Nz0Use 'async with' instead of 'with' for Semaphore�
�RuntimeErrorrrrrrc�
�
zSemaphore.__enter__�typre�	tracebackcC�|��
dSr�
rc)rrrmr�rrrri�
�zSemaphore.__exit__c

���
|��IdH
dSr�
rzrrrr�
__aenter__�
��
zSemaphore.__aenter__�tbc���
|��
dSr�
rh�rrrmr�rrr�	__aexit__�
��zSemaphore.__aexit__rKr%r)r&r'r(r)rQrrLr.rhrrrMrNrOr
r`rzrcrjrkrlrir�r��
__classcell__rrrqrr
s>
d
	�
�
�
���
�
����rcs:eZ
dZd
Zd
deddf�f
dd�
Zd�f
dd	�Z�ZS)ra:
A semaphore that prevents release() being called too many times.

    If `.release` would increment the semaphore's value past the initial
    value, it raises `ValueError`. Semaphores are mostly used to guard
    resources with limited capacity, so a semaphore released too many times
    is a sign of a bug.
    rrmrNcst�j
|
d
�

|
|_dS)N�
rm)rnr�_initial_valuerprqrrr�
s


zBoundedSemaphore.__init__c

s"|j|j
kr
td
�
�
t���
dS)rwz!Semaphore released too many timesN)rRr�rornrhrrqrrrh�
s

zBoundedSemaphore.releaserKr%)r&r'r(r)rQrrhr�rrrqrr�
s
rc@s�eZ
dZd
Zddd�Zdefdd�Z	ddeee	e
jfdee
fd	d
�Zddd�Zdd
d�Zdddeedeejddfdd�Zddd�Zdddeedeejddfdd�ZdS)ra�A lock for coroutines.

    A Lock begins unlocked, and `acquire` locks it immediately. While it is
    locked, a coroutine that yields `acquire` waits until another coroutine
    calls `release`.

    Releasing an unlocked lock raises `RuntimeError`.

    A Lock can be used as an async context manager with the ``async
    with`` statement:

    >>> from tornado import locks
    >>> lock = locks.Lock()
    >>>
    >>> async def f():
    ...    async with lock:
    ...        # Do something holding the lock.
    ...        pass
    ...
    ...    # Now the lock is released.

    For compatibility with older versions of Python, the `.acquire`
    method asynchronously returns a regular context manager:

    >>> async def f2():
    ...    with (yield lock.acquire()):
    ...        # Do something holding the lock.
    ...        pass
    ...
    ...    # Now the lock is released.

    .. versionchanged:: 4.3
       Added ``async with`` support in Python 3.5.

    rNc

Cstd
d�
|_
dS)Nrr�)r�_blockrrrrrs
z
Lock.__init__c

Csd
|jj
|jfS)Nz<%s _block=%s>)r+r&r�rrrrr.s
z
Lock.__repr__r/cCs|j�
|
�
S)
z�Attempt to lock. Returns an awaitable.

        Returns an awaitable, which raises `tornado.util.TimeoutError` after a
        timeout.
        )r�rz)rr/rrrrzszLock.acquirec

Cs(z|j�
�
WdSty


td
�
�
w)z�Unlock.

        The first coroutine in line waiting for `acquire` gets the lock.

        If not locked, raise a `RuntimeError`.
        zrelease unlocked lockN)r�rhror}rrrrrhs



�zLock.releasec

Cr{)Nz+Use `async with` instead of `with` for Lockr|rrrrrc(r~zLock.__enter__rrermr�cCr�rr�r�rrrri+r�z
Lock.__exit__c

�r�rr�rrrrr�3r�zLock.__aenter__c�r�rr�r�rrrr�6r�zLock.__aexit__r%r)r&r'r(r)rrLr.rrrMrNrOr
r`rzrhrcrjrkrlrir�r�rrrrr�
s>

$�
�
�


���
�
����r)rrNrk�tornadorr�tornado.concurrentrr�typingrrrr	r
�
TYPE_CHECKINGrr�__all__�objectrr
rr`rrrrrrr�<module>
s$




id: