B h/@s&dZddlZddlZddlZddlmZmZddlmZm Z ddl m Z ddl m Z mZmZmZddl Z e jrddl mZmZmZmZedZd 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 Awaitable)DequeTupleListAny_TQueue PriorityQueue LifoQueue QueueFull QueueEmptyc@seZdZdZdS)rz:Raised by `.Queue.get_nowait` when the queue has no items.N)__name__ __module__ __qualname____doc__rrKC:\Users\sanjo\AppData\Local\Qlobot\Launcher\ext_packages\tornado\queues.pyr/sc@seZdZdZdS)rzBRaised by `.Queue.put_nowait` when a queue is at its maximum size.N)rrrrrrrrr5s)futuretimeoutreturncsD|r@ddfdd }tj||fdddS)N)rcsstdS)N)done set_exceptionr TimeoutErrorr)rrr on_timeout@sz _set_timeout..on_timeoutcs S)N)remove_timeout)_)io_looptimeout_handlerrFz_set_timeout..)rIOLoopcurrent add_timeoutadd_done_callback)rrr!r)rr$r%r _set_timeout;s   r,c@s.eZdZdddddZeedddZdS) _QueueIteratorz Queue[_T]N)qrcCs ||_dS)N)r.)selfr.rrr__init__Jsz_QueueIterator.__init__)rcCs |jS)N)r.get)r/rrr __anext__Msz_QueueIterator.__anext__)rrrr0r rr2rrrrr-Isr-c@sreZdZdZdZd2eddddZeeddd Zedd d Z e dd d Z e dddZ d3e eeejfddddZe ddddZd4eeejfee dddZe dddZddddZd5eeejfeddddZee dd d!Zddd"d#Ze dd$d%Ze ddd&d'Ze ddd(d)Zddd*d+Ze 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:: from tornado import gen 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 gen.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') IOLoop.current().run_sync(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/r3rrrr0s  zQueue.__init__)rcCs|jS)z%Number of items allowed in the queue.)r6)r/rrrr3sz Queue.maxsizecCs t|jS)zNumber of items in the queue.)len_queue)r/rrrqsizesz Queue.qsizecCs|j S)N)r@)r/rrremptysz Queue.emptycCs |jdkrdS||jkSdS)NrF)r3rA)r/rrrfulls z Queue.fullz Future[None])itemrrcCsRt}y||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.put)rDrcCs^||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_expiredr:rBAssertionErrorpopleft_Queue__put_internalr_getrCr)r/rDgetterrrrrEs  zQueue.put_nowait)rrcCsHt}y||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_nowaitrr:rFr,)r/rrrrrr1s 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) rIr;rCrJrKrLrrMrAr)r/rDputterrrrrOs  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<r5r=r>)r/rrr 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|S)N)r-)r/rrr __aiter__*szQueue.__aiter__cCst|_dS)N)r8r9r@)r/rrrr7.sz Queue._initcCs |jS)N)r@rK)r/rrrrM1sz Queue._getcCs|j|dS)N)r@rF)r/rDrrr_put4sz Queue._putcCs&|jd7_|j||dS)NrQ)r<r=clearrV)r/rDrrrZ__put_internal9s zQueue.__put_internalcCsPx&|jr&|jddr&|jqWx"|jrJ|jdrJ|jq*WdS)NrrQ)r;rrKr:)r/rrrrI>szQueue._consume_expiredcCs dt|jtt||fS)Nz <%s at %s %s>)typerhexid_format)r/rrr__repr__FszQueue.__repr__cCsdt|j|fS)Nz<%s %s>)rXrr[)r/rrr__str__Isz 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)r3getattrr@r:r?r;r<)r/resultrrrr[Ls  z Queue._format)r)N)N)N)$rrrrr@intr0propertyr3rAboolrBrCrrfloatdatetime timedeltarHrEr r1rOrRrTr-rUr7rMrVrLrIstrr\r]r[rrrrrQs0B  c@s<eZdZdZddddZeddddZedd d ZdS) raCA `.Queue` that retrieves entries in priority order, lowest first. Entries are typically tuples like ``(priority number, data)``. .. testcode:: from tornado.queues import PriorityQueue q = PriorityQueue() q.put((1, 'medium-priority item')) q.put((0, 'high-priority item')) q.put((10, 'low-priority item')) print(q.get_nowait()) print(q.get_nowait()) print(q.get_nowait()) .. testoutput:: (0, 'high-priority item') (1, 'medium-priority item') (10, 'low-priority item') N)rcCs g|_dS)N)r@)r/rrrr7rszPriorityQueue._init)rDrcCst|j|dS)N)heapqheappushr@)r/rDrrrrVuszPriorityQueue._putcCs t|jS)N)rgheappopr@)r/rrrrMxszPriorityQueue._get)rrrrr7rrVrMrrrrrYsc@s<eZdZdZddddZeddddZedd d ZdS) ra]A `.Queue` that retrieves the most recently put items first. .. testcode:: from tornado.queues import LifoQueue q = LifoQueue() q.put(3) q.put(2) q.put(1) print(q.get_nowait()) print(q.get_nowait()) print(q.get_nowait()) .. testoutput:: 1 2 3 N)rcCs g|_dS)N)r@)r/rrrr7szLifoQueue._init)rDrcCs|j|dS)N)r@rF)r/rDrrrrVszLifoQueue._putcCs |jS)N)r@pop)r/rrrrMszLifoQueue._get)rrrrr7rrVrMrrrrr|s)"rr8rdrgtornadorrtornado.concurrentrrZ tornado.locksrtypingrrr r TYPE_CHECKINGr r r rr__all__ Exceptionrrrcrer,r-rrrrrrrs,   #