o tf0@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__rrW/var/www/html/software/conda/envs/catlas/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)done set_exceptionr TimeoutErrorr)rrr on_timeout@sz _set_timeout..on_timeoutcs Sr!)Zremove_timeout)_)io_looptimeout_handlerrFs z_set_timeout..r N)rZIOLoopcurrentZ add_timeoutadd_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/getr1rrr __anext__Mr3z_QueueIterator.__anext__)r/r0r N)rrrr2r rr7rrrrr.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)r1r8rrrr2s  zQueue.__init__cCs|jS)z%Number of items allowed in the queue.)r;r6rrrr8sz Queue.maxsizecCs t|jS)zNumber of items in the queue.)len_queuer6rrrqsizes z Queue.qsizecCs|j Sr!rEr6rrremptyz Queue.emptycCs|jdkrdS||jkS)NrF)r8rFr6rrrfulls 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)r1rKrrrrrputs    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?rHpopleft_Queue__put_internalr_getrJr)r1rKgetterrrrrLs  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). )rrN get_nowaitrr?rMr-)r1rrrrrr5s  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) rPr@rJrQrRrrSrFr)r1rKputterrrrrUs  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)rAr:rBrCr6rrr 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. )rBwait)r1rrrrjoin$s z Queue.joincCst|Sr!)r.r6rrr __aiter__.rIzQueue.__aiter__cCst|_dSr!)r=r>rEr6rrrr<2sz Queue._initcCr4r!)rErQr6rrrrS5r3z Queue._getcC|j|dSr!rErMr1rKrrr_put8z Queue._putcCs&|jd7_|j||dS)NrW)rArBclearr_r^rrrZ__put_internal=s zQueue.__put_internalcCs||jr|jddr|j|jr|jdds |jr8|jdr<|j|jr:|jds'dSdSdSdS)NrrW)r@r"rQr?r6rrrrPBs   $zQueue._consume_expiredcCs dt|jtt||fS)Nz <%s at %s %s>)typerhexid_formatr6rrr__repr__Js zQueue.__repr__cCsdt|j|fS)Nz<%s %s>)rbrrer6rrr__str__Msz Queue.__str__cCsnd|jf}t|ddr|d|j7}|jr|dt|j7}|jr+|dt|j7}|jr5|d|j7}|S)Nz maxsize=%rrEz queue=%rz getters[%s]z putters[%s]z tasks=%s)r8getattrrEr?rDr@rA)r1resultrrrrePs  z Queue._format)rr!r*)%rrrrrEintr2propertyr8rFboolrHrJrr rfloatdatetime timedeltarOrLr r5rUrXrZr.r[r<rSr_rRrPstrrfrgrerrrrrQsRE      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!rGr6rrrr<zr3zPriorityQueue._initrKcCst|j|dSr!)heapqheappushrEr^rrrr_}szPriorityQueue._putcCs t|jSr!)rsheappoprEr6rrrrSs zPriorityQueue._getr*rrrrr<rr_rSrrrrr]s  rc@rq) 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 NcCrrr!rGr6rrrr<r3zLifoQueue._initrKcCr\r!r]r^rrrr_r`zLifoQueue._putcCr4r!)rEpopr6rrrrSr3zLifoQueue._getr*rvrrrrrs  r)"rr=rnrstornadorrZtornado.concurrentrrZ tornado.locksrtypingrrr r r TYPE_CHECKINGr r rr__all__ Exceptionrrrmror-r.rrrrrrrs8   '