+ V@i0nRt^RIt^RIt^RIt^RIHtHt^RIHtH t ^RI H t ^RI H t HtHtHtHt^RI t ] P$'d ^RI HtHtHt]!R4t.ROt!R R ]4t!RR ]4tRRlt!RR]],4t!RR]],4t!RR ]4t!RR ]4tR#)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]tRt^/tRtRtR#)rz: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/sDrc]tRt^5tRtRtR#)rzBRaised by `.Queue.put_nowait` when a queue is at its maximum size.rNrrrrrr5sLrclV^8dQhR\R\R\\P3,RR/#)futuretimeoutNreturn)rrfloatdatetime timedelta)formats"r __annotate__r*;s; S S  S"40B0B#BC S  SrcaaaV'dQRV3Rllp\PP4oSPW4oSP VV3Rl4R#R#)cV^8dQhRR/#r"r%Nr)r)s"rr*"_set_timeout..__annotate__@s 9 9D 9rc<SP4'g'SP\P!44R#R#N)done set_exceptionr TimeoutError)r#sr on_timeout _set_timeout..on_timeout@s*;;==$$S%5%5%78!rc&<SPS4#r0)remove_timeout)_io_looptimeout_handles&r_set_timeout..Fs7+A+A.+QrN)rIOLoopcurrent add_timeoutadd_done_callback)r#r$r4r9r:sf& @@r _set_timeoutrA;sJ 9 9--'') ,,WA  !QRrcDa]tRt^ItoV3RlRltV3RlRltRtVtR#)_QueueIteratorc"<V^8dQhRRRR/#)r"qz Queue[_T]r%Nr)r) __classdict__s"rr*_QueueIterator.__annotate__Js+$rcWnR#r0rE)selfrEs&&r__init___QueueIterator.__init__Jsrc0<V^8dQhRS[S[,/#r"r%)r r)r)rFs"rr*rGMs9R=rc6VPP4#r0)rEgetrJs&r __anext___QueueIterator.__anext__Msvvzz|rrIN)rrrrrKrRr__classdictcell__rFs@rrCrCIsrrCca]tRt^QtoRtRtR,V3RlRllt]V3RlRl4tV3RlRlt V3R lR lt V3R lR lt R-V3R lRllt V3RlRlt R-V3RlRlltV3RlRltV3RlRltR-V3RlRlltV3RlRltV3RlRltV3RlRltV3RlR ltV3R!lR"ltV3R#lR$ltV3R%lR&ltV3R'lR(ltV3R)lR*ltR+tVtR#).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. Nc$<V^8dQhRS[RR/#)r"maxsizer%Nint)r)rFs"rr*Queue.__annotate__s   D rc@Vf \R4hV^8d \R4hWnVP4\P !.4Vn\P !.4Vn^Vn\4Vn VPP4R#)Nzmaxsize can't be Nonezmaxsize can't be negative) TypeError ValueError_maxsize_init collectionsdeque_getters_putters_unfinished_tasksr _finishedset)rJrXs&&rrKQueue.__init__sy ?34 4 Q;89 9  #))"- #))"- !" rc <V^8dQhRS[/#rNrY)r)rFs"rr*r[src VP#)z%Number of items allowed in the queue.)r_rQs&rrX Queue.maxsizes}}rc <V^8dQhRS[/#rNrY)r)rFs"rr*r[s  s rc ,\VP4#)zNumber of items in the queue.)len_queuerQs&rqsize Queue.qsizes4;;rc <V^8dQhRS[/#rNbool)r)rFs"rr*r[strc$VP'*#r0rorQs&rempty Queue.emptys;;rc <V^8dQhRS[/#rNrs)r)rFs"rr*r[s00d0rcbVP^8XdR#VP4VP8#)F)rXrprQs&rfull Queue.fulls& <<1 ::<4<</ /rcd<V^8dQhRS[RS[S[S[S[P 3,,RR/#)r"itemr$r%z Future[None])rr rr&r'r()r)rFs"rr*r[s:!)%x7I7I0I*J!K rc \4pVPV4VPR4V# \d,TPP Y34\ Y24T#i;i)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_resultrrdappendrA)rJrr$r#s&&& rput Queue.putse $ OOD !   d #  * MM $ 0  )  *s02A&%A&c$<V^8dQhRS[RR/#r"rr%Nr)r)rFs"rr*r[s&&r&d&rc jVP4VP'ddVP4'gQR4hVPP4pVP V4\ W P 44R#VP4'd\hVP V4R#)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_expiredrcrwpopleft_Queue__put_internalr_getr|r)rJrgetters&& rrQueue.put_nowaits{  ===::<< L!L L<]]**,F    % .vyy{ C YY[[O    %rcp<V^8dQhRS[S[S[S[P3,,RS[S[,/#)r"r$r%)r rr&r'r(r r)r)rFs"rr*r[s6eX-?-?&? @A 2rc \4pVPVP44V# \d+TPP T4\ Y!4T#i;i)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). )rr get_nowaitrrcrrA)rJr$r#s&& rrP Queue.gets\. *   doo/ 0  * MM  (  )  *s-1A"!A"c <V^8dQhRS[/#rNr)r)rFs"rr*r[sBrc jVP4VP'dgVP4'gQR4hVPP4wrVP V4\ VR4VP 4#VP4'dVP 4#\h)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) rrdr|rrrrrpr)rJrputters& rrQueue.get_nowaits  ===99;; J J J;==002LD    % .vt <99;  ZZ\\99;  rc<V^8dQhRR/#r-r)r)rFs"rr*r[s!!4!rc VP^8:d \R4hV;P^,unVP^8XdVPP4R#R#)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`. z!task_done() called too many timesN)rer^rfrgrQs&r task_doneQueue.task_donesR  ! !Q &@A A !#  ! !Q & NN    'rcn<V^8dQhRS[S[S[S[P3,,RS[R,/#)r"r$r%N)r rr&r'r(r )r)rFs"rr*r[$s6,,eX-?-?&? @A, 4,rc 8VPPV4#)zBlock until all items in the queue are processed. Returns an awaitable, which raises `tornado.util.TimeoutError` after a timeout. )rfwait)rJr$s&&rjoin Queue.join$s~~""7++rc0<V^8dQhRS[S[,/#rN)rCr)r)rFs"rr*r[.s$$>"-$rc\V4#r0)rCrQs&r __aiter__Queue.__aiter__.s d##rc<V^8dQhRR/#r-r)r)rFs"rr*r[2s**t*rc:\P!4VnR#r0)rarbrorQs&rr` Queue._init2s!'') rc <V^8dQhRS[/#rNr)r)rFs"rr*r[5s%%b%rc6VPP4#r0)rorrQs&rr Queue._get5s{{""$$rc$<V^8dQhRS[RR/#rr)r)rFs"rr*r[8!!!!rc<VPPV4R#r0rorrJrs&&r_put Queue._put8 4 rc$<V^8dQhRS[RR/#rr)r)rFs"rr*r[=s2$rcV;P^, unVPP4VPV4R#)N)rerfclearrrs&&r__put_internalQueue.__put_internal=s. !#  $rc<V^8dQhRR/#r-r)r)rFs"rr*r[Bs$$$$rclVP'dKVP^,^,P4'dVPP4K\VP'dFVP^,P4'dVPP4KUR#R#)r{N)rdr1rrcrQs&rrQueue._consume_expiredBsnmmm a 0 3 8 8 : : MM ! ! #mmm a 0 5 5 7 7 MM ! ! #!8mrc <V^8dQhRS[/#rNstr)r)rFs"rr*r[JsNN#NrcR\V4P R\\V44 RVP 4 R2#))typerhexid_formatrQs&r__repr__Queue.__repr__Js74:&&'tC4M?!DLLN;K1MMrc <V^8dQhRS[/#rNr)r)rFs"rr*r[Ms;;;rcVR\V4P RVP4 R2#)rrr)rrrrQs&r__str__ Queue.__str__Ms)4:&&'q(8::rc <V^8dQhRS[/#rNr)r)rFs"rr*r[Ps   rcRVP: 2p\VRR4'dVRVP,, pVP'd$VR\ VP4,, pVP 'd$VR\ VP 4,, pVP 'dVRVP ,, pV#)zmaxsize=roNz queue=%rz getters[%s]z putters[%s]z tasks=%s)rXgetattrrorcrnrdre)rJresults& rr Queue._formatPsDLL+, 44 ( ( kDKK/ /F === ns4=='99 9F === ns4=='99 9F  ! ! ! kD$:$:: :F r)rfrcr_rdrore)r{r0)rrrrrrorKpropertyrXrprwr|rrrPrrrrr`rrrrrrrrrTrUs@rrrQsAJF    00 .&& >$!!$,,$$**%%!!  $$NN;;  rcZa]tRtRtoRtV3RlRltV3RlRltV3RlRltR tVt R #) 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') c<V^8dQhRR/#r-r)r)rFs"rr*PriorityQueue.__annotate__ztrc.VnR#r0rvrQs&rr`PriorityQueue._initz  rc$<V^8dQhRS[RR/#rr)r)rFs"rr*r}s****rcH\P!VPV4R#r0)heapqheappushrors&&rrPriorityQueue._put}s t{{D)rc <V^8dQhRS[/#rNr)r)rFs"rr*rs**b*rcB\P!VP4#r0)rheappoprorQs&rrPriorityQueue._gets}}T[[))rrvN rrrrrr`rrrrTrUs@rrr]s(8****rcZa]tRtRtoRtV3RlRltV3RlRltV3RlRltR tVt R #) 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 c<V^8dQhRR/#r-r)r)rFs"rr*LifoQueue.__annotate__rrc.VnR#r0rvrQs&rr`LifoQueue._initrrc$<V^8dQhRS[RR/#rr)r)rFs"rr*rrrc<VPPV4R#r0rrs&&rrLifoQueue._putrrc <V^8dQhRS[/#rNr)r)rFs"rr*rs!!b!rc6VPP4#r0)ropoprQs&rrLifoQueue._gets{{  rrvNrrUs@rrrs(4!!!!r)rrrrr) rrar'rtornadorrtornado.concurrentrr tornado.locksrtypingrrr r r TYPE_CHECKINGr r rr__all__ ExceptionrrrArCrrrrrrrs  I??  (( T] L    SWR[IGBKIX$*E$*N"!"!r