o h0@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__rrB/var/www/vscode/kcb/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!)remove_timeout)_)io_looptimeout_handlerrFs z_set_timeout..r N)rIOLoopcurrent add_timeoutadd_done_callback)rrr%r)rr(r)r _set_timeout;s   r0c@s(eZdZd ddZdeefddZdS) _QueueIteratorq Queue[_T]r NcCs ||_dSr!)r2)selfr2rrr__init__J z_QueueIterator.__init__cC |jSr!)r2getr4rrr __anext__Mr6z_QueueIterator.__anext__)r2r3r N)rrrr5r rr:rrrrr1Is r1c@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)r4r;rrrr5s  zQueue.__init__cCs|jS)z%Number of items allowed in the queue.)r>r9rrrr;sz Queue.maxsizecCs t|jS)zNumber of items in the queue.)len_queuer9rrrqsizes z Queue.qsizecCs|j Sr!rHr9rrremptyz Queue.emptycCs|jdkrdS||jkS)NrF)r;rIr9rrrfulls 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_nowaitrrCappendr0 set_result)r4rNrrrrrputs    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_expiredrBrKpopleft_Queue__put_internalr_getrMr)r4rNgetterrrrrOs  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). )rrQ get_nowaitrrBrPr0)r4rrrrrr8s  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) rSrCrMrTrUrrVrIr)r4rNputterrrrrXs  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)rDr=rErFr9rrr 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. )rEwait)r4rrrrjoin$s z Queue.joincCst|Sr!)r1r9rrr __aiter__.rLzQueue.__aiter__cCst|_dSr!)r@rArHr9rrrr?2sz Queue._initcCr7r!)rHrTr9rrrrV5r6z Queue._getcC|j|dSr!rHrPr4rNrrr_put8z Queue._putcCs&|jd7_|j||dS)NrZ)rDrEclearrbrarrr__put_internal=s zQueue.__put_internalcCs||jr|jddr|j|jr|jdds |jr8|jdr<|j|jr:|jds'dSdSdSdS)NrrZ)rCr"rTrBr9rrrrSBs   $zQueue._consume_expiredcCs dt|jtt||fS)Nz <%s at %s %s>)typerhexid_formatr9rrr__repr__Js zQueue.__repr__cCsdt|j|fS)Nz<%s %s>)rfrrir9rrr__str__Msz Queue.__str__cCsnd|jf}t|ddr|d|j7}|jr|dt|j7}|jr+|dt|j7}|jr5|d|j7}|S)Nz maxsize=%rrHz queue=%rz getters[%s]z putters[%s]z tasks=%s)r;getattrrHrBrGrCrD)r4resultrrrriPs  z Queue._format)rr!r+)%rrrrrHintr5propertyr;rIboolrKrMrr rfloatdatetime timedeltarRrOr r8rXr[r]r1r^r?rVrbrUrSstrrjrkrirrrrrQsRE      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!rJr9rrrr?zr6zPriorityQueue._initrNcCst|j|dSr!)heapqheappushrHrarrrrb}szPriorityQueue._putcCs t|jSr!)rwheappoprHr9rrrrVs zPriorityQueue._getr+rrrrr?rrbrVrrrrr]s  rc@ru) 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 NcCrvr!rJr9rrrr?r6zLifoQueue._initrNcCr_r!r`rarrrrbrczLifoQueue._putcCr7r!)rHpopr9rrrrVr6zLifoQueue._getr+rzrrrrrs  r)"rr@rrrwtornadorrtornado.concurrentrr tornado.locksrtypingrrr r r TYPE_CHECKINGr r rr__all__ Exceptionrrrqrsr0r1rrrrrrrs8   '