U lu f0@s&dZddlZddlZddlZddlmZmZddlmZm Z ddl m Z ddl m Z mZmZmZmZddl Z e jrddl mZmZmZedZd d d d d gZGdd d eZGdd d eZee deejfddddZGdddeeZGdd d eeZGdd d eZ Gdd d eZ!dS)aAsynchronous queues for coroutines. These classes are very similar to those provided in the standard library's `asyncio package `_. .. 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)genioloop)Future"future_set_result_unless_cancelled)Event)UnionTypeVarGeneric AwaitableOptional)DequeTupleAny_TQueue PriorityQueue LifoQueue QueueFull QueueEmptyc@seZdZdZdS)rz:Raised by `.Queue.get_nowait` when the queue has no items.N__name__ __module__ __qualname____doc__rr2/tmp/pip-unpacked-wheel-bmg6zs32/tornado/queues.pyr/sc@seZdZdZdS)rzBRaised by `.Queue.put_nowait` when a queue is at its maximum size.Nrrrrrr5s)futuretimeoutreturncsD|r@ddfdd }tj||fdddS)NrcsstdSN)doneZ set_exceptionr TimeoutErrorr)rrr on_timeout@sz _set_timeout..on_timeoutcs Sr )Zremove_timeout)_)io_looptimeout_handlerrFz_set_timeout..)rZIOLoopcurrentZ add_timeoutZadd_done_callback)rrr#r)rr%r&r _set_timeout;s   r*c@s.eZdZdddddZeedddZdS) _QueueIteratorz Queue[_T]N)qrcCs ||_dSr )r,)selfr,rrr__init__Jsz_QueueIterator.__init__rcCs |jSr )r,getr-rrr __anext__Msz_QueueIterator.__anext__)rrrr.r rr1rrrrr+Isr+c@s~eZdZdZdZd2eddddZeeddd Zedd d Z e dd d Z e dddZ d3e eeeejfddddZe ddddZd4eeeejfee dddZe dddZddddZd5eeeejfeddddZee dd d!Zddd"d#Ze dd$d%Ze ddd&d'Ze ddd(d)Zddd*d+Z e!dd,d-Z"e!dd.d/Z#e!dd0d1Z$dS)6raCoordinate 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)maxsizercCsb|dkrtd|dkr td||_|tg|_tg|_d|_t |_ |j dS)Nzmaxsize can't be Nonerzmaxsize can't be negative) TypeError ValueError_maxsize_init collectionsdeque_getters_putters_unfinished_tasksr _finishedset)r-r2rrrr.s  zQueue.__init__rcCs|jS)z%Number of items allowed in the queue.)r5r0rrrr2sz Queue.maxsizecCs t|jS)zNumber of items in the queue.)len_queuer0rrrqsizesz Queue.qsizecCs|j Sr r?r0rrremptysz Queue.emptycCs |jdkrdS||jkSdS)NrF)r2r@r0rrrfulls z Queue.fullz Future[None])itemrrcCsRt}z||Wn.tk rB|j||ft||Yn X|d|S)aPut 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_nowaitrr:appendr* set_result)r-rDrrrrrputs  z Queue.putrDrcCs^||jrB|std|j}||t||n|rPt n ||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_expiredr9rBAssertionErrorpopleft_Queue__put_internalr_getrCr)r-rDgetterrrrrEs  zQueue.put_nowait)rrcCsHt}z||Wn*tk rB|j|t||YnX|S)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). )rrG get_nowaitrr9rFr*)r-rrrrrr/s z Queue.getcCs`||jrH|std|j\}}||t|d|S|rX|St dS)zRemove 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) rJr:rCrKrLrMrrNr@r)r-rDZputterrrrrPs  zQueue.get_nowaitcCs8|jdkrtd|jd8_|jdkr4|jdS)aIndicate 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 timesN)r;r4r<r=r0rrr task_dones  zQueue.task_donecCs |j|S)zBlock until all items in the queue are processed. Returns an awaitable, which raises `tornado.util.TimeoutError` after a timeout. )r<wait)r-rrrrjoin$sz Queue.joincCst|Sr )r+r0rrr __aiter__.szQueue.__aiter__cCst|_dSr )r7r8r?r0rrrr62sz Queue._initcCs |jSr )r?rLr0rrrrN5sz Queue._getcCs|j|dSr r?rFr-rDrrr_put8sz Queue._putcCs&|jd7_|j||dS)NrQ)r;r<clearrXrWrrrZ__put_internal=s zQueue.__put_internalcCsH|jr$|jddr$|jq|jrD|jdrD|jq$dS)NrrQ)r:r!rLr9r0rrrrJBs zQueue._consume_expiredcCs dt|jtt||fS)Nz <%s at %s %s>)typerhexid_formatr0rrr__repr__JszQueue.__repr__cCsdt|j|fS)Nz<%s %s>)rZrr]r0rrr__str__Msz Queue.__str__cCsnd|jf}t|ddr&|d|j7}|jr>|dt|j7}|jrV|dt|j7}|jrj|d|j7}|S)Nz maxsize=%rr?z queue=%rz getters[%s]z putters[%s]z tasks=%s)r2getattrr?r9r>r:r;)r-resultrrrr]Ps  z Queue._format)r)N)N)N)%rrrrr?intr.propertyr2r@boolrBrCrr rfloatdatetime timedeltarHrEr r/rPrRrTr+rUr6rNrXrMrJstrr^r_r]rrrrrQsHE   c@s<eZdZdZddddZeddddZedd d ZdS) raA `.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') NrcCs g|_dSr rAr0rrrr6zszPriorityQueue._initrIcCst|j|dSr )heapqheappushr?rWrrrrX}szPriorityQueue._putcCs t|jSr )riheappopr?r0rrrrNszPriorityQueue._getrrrrr6rrXrNrrrrr]sc@s<eZdZdZddddZeddddZedd d ZdS) raA `.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 NrcCs g|_dSr rAr0rrrr6szLifoQueue._initrIcCs|j|dSr rVrWrrrrXszLifoQueue._putcCs |jSr )r?popr0rrrrNszLifoQueue._getrlrrrrrs)"rr7rfriZtornadorrZtornado.concurrentrrZ tornado.locksrtypingrrr r r TYPE_CHECKINGr r rr__all__ Exceptionrrrergr*r+rrrrrrrs0   '