Mh0dZddlZddlZddlZddlmZmZddlmZm Z ddl m Z ddl m Z mZmZmZmZddl Z e jr ddl mZmZmZedZgd ZGd d eZGd d eZdede deejfddfdZGddeeZGddeeZGddeZ GddeZ!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 QueueEmptyceZdZdZdS)rz:Raised by `.Queue.get_nowait` when the queue has no items.N__name__ __module__ __qualname____doc__N/var/www/html/test/jupyter/venv/lib/python3.11/site-packages/tornado/queues.pyrr/sDDDrrceZdZdZdS)rzBRaised by `.Queue.put_nowait` when a queue is at its maximum size.Nrrrrrr5sLLDrrfuturetimeoutreturnc|rUdfd }tj||fddSdS)Nr"cs(tjdSdSN)done set_exceptionr TimeoutError)r sr on_timeoutz _set_timeout..on_timeout@s@;;== 9$$S%5%7%788888 9 9rc.Sr%)remove_timeout)_io_looptimeout_handles rz_set_timeout..Fs7+A+A.+Q+Qrr"N)rIOLoopcurrent add_timeoutadd_done_callback)r r!r)r-r.s` @@r _set_timeoutr5;sS 9 9 9 9 9 9-'')) ,,WjAA  !Q!Q!Q!Q!QRRRRRSSrc.eZdZddZdeefdZdS)_QueueIteratorq Queue[_T]r"Nc||_dSr%)r8)selfr8s r__init__z_QueueIterator.__init__Js rc4|jSr%)r8getr;s r __anext__z_QueueIterator.__anext__Msvzz||r)r8r9r"N)rrrr<r rr@rrrr7r7IsG9R=rr7ceZdZdZdZddeddfdZedefdZdefdZ de fd Z de fd Z dd e d eeeejfdd fdZd e ddfdZ dd eeeejfdee fdZde fdZddZ dd eeeejfdedfdZdee fdZddZde fdZd e ddfdZd e ddfdZddZ de!fdZ"de!fdZ#de!fdZ$dS) raCoordinate 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"cR|td|dkrtd||_|t jg|_t jg|_d|_t|_ |j dS)Nzmaxsize can't be Nonerzmaxsize can't be negative) TypeError ValueError_maxsize_init collectionsdeque_getters_putters_unfinished_tasksr _finishedset)r;rBs rr<zQueue.__init__s ?344 4 Q;;899 9  #)"-- #)"-- !" rc|jS)z%Number of items allowed in the queue.)rFr?s rrBz Queue.maxsizes }rc*t|jS)zNumber of items in the queue.)len_queuer?s rqsizez Queue.qsizes4;rc|j Sr%rRr?s remptyz Queue.emptys ;rcV|jdkrdS||jkS)NrF)rBrSr?s rfullz Queue.fulls) <1  5::<<4</ /ritemr!z Future[None]ct} |||dn<#t$r/|j||ft ||YnwxYw|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_nowait set_resultrrKappendr5)r;rYr!r s rputz Queue.puts $ OOD ! ! !   d # # # #  * * * M $ 0 0 0  ) ) ) ) ) *  s;6A43A4c|||jrp|s Jd|j}||t ||dS|rt||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_expiredrJrVpopleft_Queue__put_internalr_getrXr)r;rYgetters rr[zQueue.put_nowaits  = &::<< L L!L L L L]**,,F    % % % .vtyy{{ C C C C C YY[[ &O    % % % % %rct} ||n:#t$r-|j|t ||YnwxYw|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). )rr\ get_nowaitrrJr]r5)r;r!r s rr>z Queue.gets. *   doo// 0 0 0 0 * * * M  ( ( (  ) ) ) ) ) * s'84A/.A/c|||jrs|s Jd|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) r`rKrXrarbrrcrSr)r;rYputters rrfzQueue.get_nowaits  = 99;; J J J J J J=0022LD&    % % % .vt < < <99;;  ZZ\\ 99;;  rc|jdkrtd|xjdzc_|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)rLrErMrNr?s r task_donezQueue.task_donese  !Q & &@AA A !#  !Q & & N    ' &rc6|j|S)zBlock until all items in the queue are processed. Returns an awaitable, which raises `tornado.util.TimeoutError` after a timeout. )rMwait)r;r!s rjoinz Queue.join$s~""7+++rc t|Sr%)r7r?s r __aiter__zQueue.__aiter__.sd###rc6tj|_dSr%)rHrIrRr?s rrGz Queue._init2s!')) rc4|jSr%)rRrar?s rrcz Queue._get5s{""$$$rc:|j|dSr%rRr]r;rYs r_putz Queue._put8 4     rc|xjdz c_|j||dS)Nrj)rLrMclearrvrus r__put_internalzQueue.__put_internal=s= !#  $rc|jrj|jddrE|j|jr%|jddE|jr`|jdrE|j|jr#|jd?dSdSdSdS)Nrrj)rKr&rarJr?s rr`zQueue._consume_expiredBsm $ a 0 3 8 8 : : $ M ! ! # # #m $ a 0 3 8 8 : : $m $ a 0 5 5 7 7 $ M ! ! # # #m $ a 0 5 5 7 7 $ $ $ $ $ $ $ $ $rcdt|jdtt|d|dS)N)typerhexid_formatr?s r__repr__zQueue.__repr__Js@M4::&MMC4MMMMDLLNNMMMMrc\dt|jd|dS)Nr}r~r)rrrr?s r__str__z Queue.__str__Ms,:4::&::::::rcd|j}t|ddr |d|jzz }|jr|dt |jzz }|jr|dt |jzz }|jr |d|jzz }|S)Nzmaxsize=rRz queue=%rz getters[%s]z putters[%s]z tasks=%s)rBgetattrrRrJrQrKrL)r;results rrz Queue._formatPs,DL,, 44 ( ( 0 kDK/ /F = : ns4='9'99 9F = : ns4='9'99 9F  ! ; kD$:: :F r)rr%r0)%rrrrrRintr<propertyrBrSboolrVrXrr rfloatdatetime timedeltar^r[r r>rfrkrnr7rprGrcrvrbr`strrrrrrrrrQsAAJF   D    X s    t0d0000OS!)%x7I0I*J!K .&r&d&&&&"EIeX-?&? @A 2>B$!!!!&EI,,eX-?&? @A, 4,,,,$>"-$$$$****%b%%%%!!!!!! 2$ $$$$N#NNNN;;;;;       rrc6eZdZdZddZdeddfdZdefdZdS) 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"Ncg|_dSr%rUr?s rrGzPriorityQueue._initz  rrYc:tj|j|dSr%)heapqheappushrRrus rrvzPriorityQueue._put}s t{D)))))rc4tj|jSr%)rheappoprRr?s rrczPriorityQueue._gets}T[)))rr0rrrrrGrrvrcrrrrr]so8*******b******rrc6eZdZdZddZdeddfdZdefdZdS) 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"Ncg|_dSr%rUr?s rrGzLifoQueue._initrrrYc:|j|dSr%rtrus rrvzLifoQueue._putrwrc4|jSr%)rRpopr?s rrczLifoQueue._gets{   rr0rrrrrrso4!!!!!!!b!!!!!!rr)"rrHrrtornadorrtornado.concurrentrr tornado.locksrtypingrr r r r TYPE_CHECKINGr rrr__all__ Exceptionrrrrr5r7rrrrrrrs?   IIIIIIII??????????????  )(((((((((( WT]] L L L                S  S"40B#BC S  S S S SWR[IIIIIGBKIIIX$*$*$*$*$*E$*$*$*N"!"!"!"!"!"!"!"!"!"!r