HEX

File: //home/arjun/projects/env/lib/python3.10/site-packages/tornado/__pycache__/queues.cpython-310.pyc
o

we�0�@s$dZddlZddlZddlZddlmZmZddlmZm	Z	ddl
mZddlm
Z
mZmZmZmZddlZejrCddlmZmZmZed�Zgd	�ZGd
d�de�ZGdd
�d
e�Zdede
deejfddfdd�ZGdd�dee�ZGdd�dee�ZGdd�de�Z Gdd�de�Z!dS)a�Asynchronous queues for coroutines. These classes are very similar
to those provided in the standard library's `asyncio package
<https://docs.python.org/3/library/asyncio-queue.html>`_.

.. warning::

   Unlike the standard library's `queue` module, the classes defined here
   are *not* thread-safe. To use these queues from another thread,
   use `.IOLoop.add_callback` to transfer control to the `.IOLoop` thread
   before calling any queue methods.

�N)�gen�ioloop)�Future�"future_set_result_unless_cancelled)�Event)�Union�TypeVar�Generic�	Awaitable�Optional)�Deque�Tuple�Any�_T)�Queue�
PriorityQueue�	LifoQueue�	QueueFull�
QueueEmptyc@�eZdZdZdS)rz:Raised by `.Queue.get_nowait` when the queue has no items.N��__name__�
__module__�__qualname__�__doc__�rr�G/home/arjun/projects/env/lib/python3.10/site-packages/tornado/queues.pyr/�rc@r)rzBRaised by `.Queue.put_nowait` when a queue is at its maximum size.Nrrrrrr5rr�future�timeout�returncsD|r d�fdd�}tj�����||������fdd��dSdS)Nr cs���s
��t���dSdS�N)�done�
set_exceptionr�TimeoutErrorr)rrr�
on_timeout@s�z _set_timeout.<locals>.on_timeoutcs
����Sr!)�remove_timeout)�_)�io_loop�timeout_handlerr�<lambda>Fs
z_set_timeout.<locals>.<lambda>�r N)r�IOLoop�current�add_timeout�add_done_callback)rrr%r)rr(r)r�_set_timeout;s
�r0c@s(eZdZd	dd�Zdeefdd�ZdS)
�_QueueIterator�q�	Queue[_T]r NcCs
||_dSr!)r2)�selfr2rrr�__init__J�
z_QueueIterator.__init__cC�
|j��Sr!)r2�get�r4rrr�	__anext__Mr6z_QueueIterator.__anext__)r2r3r N)rrrr5r
rr:rrrrr1Is
r1c@s�eZdZdZdZd1deddfdd�Zedefdd	��Zdefd
d�Z	de
fdd
�Zde
fdd�Z	d2de
deeeejfddfdd�Zde
ddfdd�Z	d2deeeejfdee
fdd�Zde
fdd�Zd3dd�Z	d2deeeejfdedfdd�Zdee
fdd �Zd3d!d"�Zde
fd#d$�Zde
ddfd%d&�Zde
ddfd'd(�Zd3d)d*�Z de!fd+d,�Z"de!fd-d.�Z#de!fd/d0�Z$dS)4ra�Coordinate producer and consumer coroutines.

    If maxsize is 0 (the default) the queue size is unbounded.

    .. testcode::

        import asyncio
        from tornado.ioloop import IOLoop
        from tornado.queues import Queue

        q = Queue(maxsize=2)

        async def consumer():
            async for item in q:
                try:
                    print('Doing work on %s' % item)
                    await asyncio.sleep(0.01)
                finally:
                    q.task_done()

        async def producer():
            for item in range(5):
                await q.put(item)
                print('Put %s' % item)

        async def main():
            # Start consumer without waiting (since it never finishes).
            IOLoop.current().spawn_callback(consumer)
            await producer()     # Wait for producer to put all tasks.
            await q.join()       # Wait for consumer to finish all tasks.
            print('Done')

        asyncio.run(main())

    .. testoutput::

        Put 0
        Put 1
        Doing work on 0
        Put 2
        Doing work on 1
        Put 3
        Doing work on 2
        Put 4
        Doing work on 3
        Doing work on 4
        Done


    In versions of Python without native coroutines (before 3.5),
    ``consumer()`` could be written as::

        @gen.coroutine
        def consumer():
            while True:
                item = yield q.get()
                try:
                    print('Doing work on %s' % item)
                    yield gen.sleep(0.01)
                finally:
                    q.task_done()

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

    Nr�maxsizer cCsb|durtd��|dkrtd��||_|��t�g�|_t�g�|_d|_t	�|_
|j
��dS)Nzmaxsize can't be Nonerzmaxsize can't be negative)�	TypeError�
ValueError�_maxsize�_init�collections�deque�_getters�_putters�_unfinished_tasksr�	_finished�set)r4r;rrrr5�szQueue.__init__cCs|jS)z%Number of items allowed in the queue.)r>r9rrrr;�sz
Queue.maxsizecCs
t|j�S)zNumber of items in the queue.)�len�_queuer9rrr�qsize�s
zQueue.qsizecCs|jSr!�rHr9rrr�empty��zQueue.emptycCs|jdkrdS|��|jkS)NrF)r;rIr9rrr�full�s
z
Queue.full�itemrzFuture[None]cCsRt�}z|�|�Wnty!|j�||f�t||�Y|Sw|�d�|S)a�Put an item into the queue, perhaps waiting until there is room.

        Returns a Future, which raises `tornado.util.TimeoutError` after a
        timeout.

        ``timeout`` may be a number denoting a time (on the same
        scale as `tornado.ioloop.IOLoop.time`, normally `time.time`), or a
        `datetime.timedelta` object for a deadline relative to the
        current time.
        N)r�
put_nowaitrrC�appendr0�
set_result)r4rNrrrrr�put�s
�
z	Queue.putcCs^|��|jr"|��sJd��|j��}|�|�t||���dS|��r(t�|�|�dS)z{Put an item into the queue without blocking.

        If no free slot is immediately available, raise `QueueFull`.
        z)queue non-empty, why are getters waiting?N)	�_consume_expiredrBrK�popleft�_Queue__put_internalr�_getrMr)r4rN�getterrrrrO�s

zQueue.put_nowaitcCsFt�}z
|�|���W|Sty"|j�|�t||�Y|Sw)a.Remove and return an item from the queue.

        Returns an awaitable which resolves once an item is available, or raises
        `tornado.util.TimeoutError` after a timeout.

        ``timeout`` may be a number denoting a time (on the same
        scale as `tornado.ioloop.IOLoop.time`, normally `time.time`), or a
        `datetime.timedelta` object for a deadline relative to the
        current time.

        .. note::

           The ``timeout`` argument of this method differs from that
           of the standard library's `queue.Queue.get`. That method
           interprets numeric values as relative timeouts; this one
           interprets them as absolute deadlines and requires
           ``timedelta`` objects for relative timeouts (consistent
           with other timeouts in Tornado).

        )rrQ�
get_nowaitrrBrPr0)r4rrrrrr8�s��z	Queue.getcCs\|��|jr$|��sJd��|j��\}}|�|�t|d�|��S|��r,|��St�)z�Remove and return an item from the queue without blocking.

        Return an item if one is immediately available, else raise
        `QueueEmpty`.
        z(queue not full, why are putters waiting?N)	rSrCrMrTrUrrVrIr)r4rN�putterrrrrXs

zQueue.get_nowaitcCs<|jdkr	td��|jd8_|jdkr|j��dSdS)a�Indicate that a formerly enqueued task is complete.

        Used by queue consumers. For each `.get` used to fetch a task, a
        subsequent call to `.task_done` tells the queue that the processing
        on the task is complete.

        If a `.join` is blocking, it resumes when all items have been
        processed; that is, when every `.put` is matched by a `.task_done`.

        Raises `ValueError` if called more times than `.put`.
        rz!task_done() called too many times�N)rDr=rErFr9rrr�	task_dones

�zQueue.task_donecCs|j�|�S)z�Block until all items in the queue are processed.

        Returns an awaitable, which raises `tornado.util.TimeoutError` after a
        timeout.
        )rE�wait)r4rrrr�join$sz
Queue.joincCst|�Sr!)r1r9rrr�	__aiter__.rLzQueue.__aiter__cCst��|_dSr!)r@rArHr9rrrr?2szQueue._initcCr7r!)rHrTr9rrrrV5r6z
Queue._getcC�|j�|�dSr!�rHrP�r4rNrrr�_put8�z
Queue._putcCs&|jd7_|j��|�|�dS)NrZ)rDrE�clearrbrarrr�__put_internal=s
zQueue.__put_internalcCs||jr|jdd��r|j��|jr|jdd��s|jr8|jd��r<|j��|jr:|jd��s'dSdSdSdS)NrrZ)rCr"rTrBr9rrrrSBs
�
$�zQueue._consume_expiredcCs dt|�jtt|��|��fS)Nz
<%s at %s %s>)�typer�hex�id�_formatr9rrr�__repr__Js zQueue.__repr__cCsdt|�j|��fS)Nz<%s %s>)rfrrir9rrr�__str__Msz
Queue.__str__cCsnd|jf}t|dd�r|d|j7}|jr|dt|j�7}|jr+|dt|j�7}|jr5|d|j7}|S)Nz
maxsize=%rrHz	 queue=%rz getters[%s]z putters[%s]z	 tasks=%s)r;�getattrrHrBrGrCrD)r4�resultrrrriPsz
Queue._format)rr!r+)%rrrrrH�intr5�propertyr;rI�boolrKrMrrr�float�datetime�	timedeltarRrOr
r8rXr[r]r1r^r?rVrbrUrS�strrjrkrirrrrrQsRE���
���
�
��
�


rc@�:eZdZdZddd�Zdeddfdd�Zdefd	d
�ZdS)ra�A `.Queue` that retrieves entries in priority order, lowest first.

    Entries are typically tuples like ``(priority number, data)``.

    .. testcode::

        import asyncio
        from tornado.queues import PriorityQueue

        async def main():
            q = PriorityQueue()
            q.put((1, 'medium-priority item'))
            q.put((0, 'high-priority item'))
            q.put((10, 'low-priority item'))

            print(await q.get())
            print(await q.get())
            print(await q.get())

        asyncio.run(main())

    .. testoutput::

        (0, 'high-priority item')
        (1, 'medium-priority item')
        (10, 'low-priority item')
    r NcC�
g|_dSr!rJr9rrrr?zr6zPriorityQueue._initrNcCst�|j|�dSr!)�heapq�heappushrHrarrrrb}szPriorityQueue._putcCst�|j�Sr!)rw�heappoprHr9rrrrV�szPriorityQueue._getr+�rrrrr?rrbrVrrrrr]s

rc@ru)ra�A `.Queue` that retrieves the most recently put items first.

    .. testcode::

        import asyncio
        from tornado.queues import LifoQueue

        async def main():
            q = LifoQueue()
            q.put(3)
            q.put(2)
            q.put(1)

            print(await q.get())
            print(await q.get())
            print(await q.get())

        asyncio.run(main())

    .. testoutput::

        1
        2
        3
    r NcCrvr!rJr9rrrr?�r6zLifoQueue._initrNcCr_r!r`rarrrrb�rczLifoQueue._putcCr7r!)rH�popr9rrrrV�r6zLifoQueue._getr+rzrrrrr�s

r)"rr@rrrw�tornadorr�tornado.concurrentrr�
tornado.locksr�typingrrr	r
r�
TYPE_CHECKINGrr
rr�__all__�	Exceptionrrrqrsr0r1rrrrrrr�<module>s8
��
�'