R?g0SrSSKrSSKrSSKrSSKJrJr SSKJrJ r SSK J r SSK J r JrJrJrJr SSK r \ R$(a SSK JrJrJr \"S5r/S Qr"S S \5r"S S \5rS\S\ S\\R84SS4Sjr"SS\\5r"SS\\5r"SS\5r "SS\5r!g)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\rSrSrSrSrg)r/z:Raised by `.Queue.get_nowait` when the queue has no items.N__name__ __module__ __qualname____firstlineno____doc____static_attributes__r0/usr/lib/python3/dist-packages/tornado/queues.pyrr/sDr rc\rSrSrSrSrg)r5zBRaised by `.Queue.put_nowait` when a queue is at its maximum size.rNrrr r!rr5sLr rfuturetimeoutreturnc^^^U(aNSU4Sjjn[RR5mTRX5mTR UU4Sj5 gg)Ncz>TR5(d%TR[R"55 ggN)done set_exceptionr TimeoutError)r$sr! on_timeout _set_timeout..on_timeout@s*;;==$$S%5%5%78!r c&>TRT5$r))remove_timeout)_io_looptimeout_handles r!_set_timeout..Fs7+A+A.+Qr r&N)rIOLoopcurrent add_timeoutadd_done_callback)r$r%r-r2r3s` @@r! _set_timeoutr;;sE 9--'') ,,WA  !QRr c2\rSrSrSSjrS\\4SjrSrg)_QueueIteratorIr&NcXlgr)q)selfrAs r!__init___QueueIterator.__init__Jsr c6URR5$r))rAgetrBs r! __anext___QueueIterator.__anext__Msvvzz|r r@)rAz Queue[_T]r&N) rrrrrCr rrHrrr r!r=r=Is9R=r r=c\rSrSrSrSrSS\SS4Sjjr\S\4Sj5r S\4Sjr S\ 4S jr S\ 4S jr SS \S \\\\R&4SS 4SjjrS \SS4SjrSS \\\\R&4S\\4SjjrS\4SjrS SjrSS \\\\R&4S\S4SjjrS\\4SjrS SjrS\4SjrS \SS4SjrS \SS4Sjr S Sjr!S\"4Sjr#S\"4Sjr$S\"4Sjr%Sr&g)!rQaCoordinate 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. Nmaxsizer&c:Uc [S5eUS:a [S5eXlUR5 [R "/5Ul[R "/5UlSUl[5Ul URR5 g)Nzmaxsize can't be Nonerzmaxsize can't be negative) TypeError ValueError_maxsize_init collectionsdeque_getters_putters_unfinished_tasksr _finishedset)rBrLs r!rCQueue.__init__sy ?34 4 Q;89 9  #))"- #))"- !" r cUR$)z%Number of items allowed in the queue.)rPrGs r!rL Queue.maxsizes}}r c,[UR5$)zNumber of items in the queue.)len_queuerGs r!qsize Queue.qsizes4;;r c$UR(+$r)r^rGs r!empty Queue.emptys;;r c^URS:XagUR5UR:$)NrF)rLr_rGs r!full Queue.fulls& <<1 ::<4<</ /r itemr%z Future[None]c[5nURU5 URS5 U$![a+ URR X45 [ X25 U$f=f)aTPut 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_resultrrUappendr;)rBrhr%r$s r!put Queue.putse $ OOD !   d #  * MM $ 0  )  *s01A%$A%c`UR5 UR(aaUR5(dS5eURR5nUR U5 [ X R 55 gUR5(a[eUR U5 g)zkPut 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_expiredrTrcpopleft_Queue__put_internalr_getrfr)rBrhgetters r!rjQueue.put_nowaitsy  ==::<< L!L L<]]**,F    % .vyy{ C YY[[O    %r c[5nURUR55 U$![a* URR U5 [ X!5 U$f=f)aRemove 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). )rrk get_nowaitrrTrlr;)rBr%r$s r!rF Queue.gets\. *   doo/ 0  * MM  (  )  *s-0A! A!cdUR5 UR(aeUR5(dS5eURR5upUR U5 [ US5 UR 5$UR5(aUR 5$[e)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) rprUrfrqrrrrsr_r)rBrhputters r!rwQueue.get_nowaits  ==99;; J J J;==002LD    % .vt <99;  ZZ\\99;  r cURS::a [S5eU=RS-slURS:XaURR5 gg)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)rVrOrWrXrGs r! task_doneQueue.task_donesR  ! !Q &@A A !#  ! !Q & NN    'r c8URRU5$)zBlock until all items in the queue are processed. Returns an awaitable, which raises `tornado.util.TimeoutError` after a timeout. )rWwait)rBr%s r!join Queue.join$s~~""7++r c[U5$r))r=rGs r! __aiter__Queue.__aiter__.s d##r c8[R"5Ulgr))rRrSr^rGs r!rQ Queue._init2s!'') r c6URR5$r))r^rqrGs r!rs Queue._get5s{{""$$r c:URRU5 gr)r^rlrBrhs r!_put Queue._put8 4 r cU=RS- slURR5 URU5 g)Nr})rVrWclearrrs r!__put_internalQueue.__put_internal=s. !#  $r cUR(awURSSR5(aRURR5 UR(a'URSSR5(aMRUR(atURSR5(aQURR5 UR(a%URSR5(aMOgggg)Nrr})rUr*rqrTrGs r!rpQueue._consume_expiredBsmm a 0 3 8 8 : : MM ! ! #mm a 0 3 8 8 : :mm a 0 5 5 7 7 MM ! ! #mm a 0 5 5 7 7m 7mr cS[U5R<S[[U55<SUR 5<S3$)N)typerhexid_formatrGs r!__repr__Queue.__repr__Js'"&t*"5"5s2d8}dllnUUr cZS[U5R<SUR5<S3$)Nrrr)rrrrGs r!__str__ Queue.__str__Ms J//@@r cdSUR<3n[USS5(aUSUR-- nUR(aUS[ UR5-- nUR (aUS[ UR 5-- nUR (aUSUR -- nU$)Nzmaxsize=r^z queue=%rz getters[%s]z putters[%s]z tasks=%s)rLgetattrr^rTr]rUrV)rBresults r!r Queue._formatPs!%/ 44 ( ( kDKK/ /F == ns4=='99 9F == ns4=='99 9F  ! ! kD$:$:: :F r )rWrTrPrUr^rV)rr)r6)'rrrrrr^intrCpropertyrLr_boolrcrfrr rfloatdatetime timedeltarmrjr rFrwr~rr=rrQrsrrrrpstrrrrrrr r!rrQsAJF  D  s t0d0OS!)%x7I7I0I*J!K .&r&d&"EIeX-?-?&? @A 2>B$!&EI,eX-?-?&? @A, 4,$>"-$*%b%!!! 2$ $V#VAA  r rcB\rSrSrSrS SjrS\SS4SjrS\4SjrS r g) ri]abA `.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&Nc/Ulgr)rbrGs r!rQPriorityQueue._initz  r rhcF[R"URU5 gr))heapqheappushr^rs r!rPriorityQueue._put}s t{{D)r cB[R"UR5$r))rheappopr^rGs r!rsPriorityQueue._gets}}T[[))r rbr6 rrrrrrQrrrsrrr r!rr]s+8****b*r rcB\rSrSrSrS SjrS\SS4SjrS\4SjrS r g) riaA `.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&Nc/Ulgr)rbrGs r!rQLifoQueue._initrr rhc:URRU5 gr)rrs r!rLifoQueue._putrr c6URR5$r))r^poprGs r!rsLifoQueue._gets{{  r rbr6rrr r!rrs+4!!!!b!r r)"rrRrrtornadorrtornado.concurrentrr tornado.locksrtypingrr r r r TYPE_CHECKINGr rrr__all__ Exceptionrrrrr;r=rrrrr r!rs  I??  (( T] L    S  S"40B0B#BC S  SWR[IGBKIX$*E$*N"!"!r