o )Zh0@s$dZddlZddlZddlZddlmZmZddlmZm Z ddl m Z ddl m Z mZmZmZmZddl Z e jrCddl mZmZmZedZgd ZGd d d eZGd d d eZdede deejfddfddZGdddeeZGdddeeZGdddeZ GdddeZ!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_T)Queue PriorityQueue LifoQueue QueueFull QueueEmptyc@eZdZdZdS)rz:Raised by `.Queue.get_nowait` when the queue has no items.N__name__ __module__ __qualname____doc__rrE/var/www/html/lang_env/lib/python3.10/site-packages/tornado/queues.pyr/rc@r)rzBRaised by `.Queue.put_nowait` when a queue is at its maximum size.Nrrrrrr5rrfuturetimeoutreturncsD|r dfdd }tj||fdddSdS)Nr css tdSdSN)doneZ set_exceptionr TimeoutErrorr)rrr on_timeout@sz _set_timeout..on_timeoutcs Sr!)Zremove_timeout)_)io_looptimeout_handlerrFs z_set_timeout..r N)rZIOLoopcurrentZ add_timeoutZadd_done_callback)rrr$r)rr&r'r _set_timeout;s   r+c@s(eZdZd ddZdeefddZdS) _QueueIteratorq Queue[_T]r NcCs ||_dSr!)r-)selfr-rrr__init__J z_QueueIterator.__init__cC |jSr!)r-getr/rrr __anext__Mr1z_QueueIterator.__anext__)r-r.r N)rrrr0r rr5rrrrr,Is r,c@seZdZdZdZd1deddfddZedefdd Zdefd d Z de fd d Z de fddZ d2de deeeejfddfddZde ddfddZ d2deeeejfdee fddZde fddZd3ddZ d2deeeejfdedfddZdee 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/d0Z$dS)4raCoordinate 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. Nrmaxsizer cCsb|durtd|dkrtd||_|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/r6rrrr0s  zQueue.__init__cCs|jS)z%Number of items allowed in the queue.)r9r4rrrr6sz Queue.maxsizecCs t|jS)zNumber of items in the queue.)len_queuer4rrrqsizes z Queue.qsizecCs|j Sr!rCr4rrremptyz Queue.emptycCs|jdkrdS||jkS)NrF)r6rDr4rrrfulls z Queue.fullitemrz Future[None]cCsRt}z||Wnty!|j||ft||Y|Sw|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/rIrrrrrputs    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_expiredr=rFpopleft_Queue__put_internalr_getrHr)r/rIgetterrrrrJs  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). )rrL get_nowaitrr=rKr+)r/rrrrrr3s  z Queue.getcCs\||jr$|sJd|j\}}||t|d|S|r,|St)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) rNr>rHrOrPrrQrDr)r/rIZputterrrrrSs  zQueue.get_nowaitcCs<|jdkr td|jd8_|jdkr|jdSdS)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?r8r@rAr4rrr 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$s z Queue.joincCst|Sr!)r,r4rrr __aiter__.rGzQueue.__aiter__cCst|_dSr!)r;r<rCr4rrrr:2sz Queue._initcCr2r!)rCrOr4rrrrQ5r1z Queue._getcC|j|dSr!rCrKr/rIrrr_put8z Queue._putcCs&|jd7_|j||dS)NrT)r?r@clearr\r[rrrZ__put_internal=s zQueue.__put_internalcCs||jr|jddr|j|jr|jdds |jr8|jdr<|j|jr:|jds'dSdSdSdS)NrrT)r>r"rOr=r4rrrrNBs   $zQueue._consume_expiredcCs dt|jtt||fS)Nz <%s at %s %s>)typerhexid_formatr4rrr__repr__Js zQueue.__repr__cCsdt|j|fS)Nz<%s %s>)r_rrbr4rrr__str__Msz Queue.__str__cCsnd|jf}t|ddr|d|j7}|jr|dt|j7}|jr+|dt|j7}|jr5|d|j7}|S)Nz maxsize=%rrCz queue=%rz getters[%s]z putters[%s]z tasks=%s)r6getattrrCr=rBr>r?)r/resultrrrrbPs  z Queue._format)rr!r))%rrrrrCintr0propertyr6rDboolrFrHrr rfloatdatetime timedeltarMrJr r3rSrUrWr,rXr:rQr\rPrNstrrcrdrbrrrrrQsRE      rc@:eZdZdZd ddZdeddfddZdefd 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') r NcC g|_dSr!rEr4rrrr:zr1zPriorityQueue._initrIcCst|j|dSr!)heapqheappushrCr[rrrr\}szPriorityQueue._putcCs t|jSr!)rpheappoprCr4rrrrQs zPriorityQueue._getr)rrrrr:rr\rQrrrrr]s  rc@rn) 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 r NcCror!rEr4rrrr:r1zLifoQueue._initrIcCrYr!rZr[rrrr\r]zLifoQueue._putcCr2r!)rCpopr4rrrrQr1zLifoQueue._getr)rsrrrrrs  r)"rr;rkrptornadorrZtornado.concurrentrrZ tornado.locksrtypingrrr r r TYPE_CHECKINGr r rr__all__ Exceptionrrrjrlr+r,rrrrrrrs8   '