B hD@sddlZddlmZddlZddlZddlmZmZddlm Z m Z ddl m Z m Z mZmZmZddl Z e jr~ddl mZmZddd d d gZGd d d eZGdddeZGdddeZGdddeZGdd d eZGdd d eZGdd d eZdS)N)CancelledError)genioloop)Future"future_set_result_unless_cancelled)UnionOptionalTypeAny Awaitable)DequeSet ConditionEvent SemaphoreBoundedSemaphoreLockc@s,eZdZdZddddZddddZdS)_TimeoutGarbageCollectorzBase class for objects that periodically clean up timed-out waiters. Avoids memory leak in a common pattern like: while True: yield condition.wait(short_timeout) print('looping....') N)returncCst|_d|_dS)Nr) collectionsdeque_waiters _timeouts)selfrJC:\Users\sanjo\AppData\Local\Qlobot\Launcher\ext_packages\tornado\locks.py__init__*s z!_TimeoutGarbageCollector.__init__cCs:|jd7_|jdkr6d|_tdd|jD|_dS)Ndrcss|]}|s|VqdS)N)done).0wrrr 3sz<_TimeoutGarbageCollector._garbage_collect..)rrrr)rrrr_garbage_collect.s z)_TimeoutGarbageCollector._garbage_collect)__name__ __module__ __qualname____doc__rr#rrrrr srcsteZdZdZddfdd ZedddZdeee j fe e dd d Z dedd d dZddddZZS)raA condition allows one or more coroutines to wait until notified. Like a standard `threading.Condition`, but does not need an underlying lock that is acquired and released. With a `Condition`, coroutines can wait to be notified by other coroutines: .. testcode:: from tornado import gen from tornado.ioloop import IOLoop from tornado.locks import Condition condition = Condition() async def waiter(): print("I'll wait right here") await condition.wait() print("I'm done waiting") async def notifier(): print("About to notify") condition.notify() print("Done notifying") async def runner(): # Wait for waiter() and notifier() in parallel await gen.multi([waiter(), notifier()]) IOLoop.current().run_sync(runner) .. testoutput:: I'll wait right here About to notify Done notifying I'm done waiting `wait` takes an optional ``timeout`` argument, which is either an absolute timestamp:: io_loop = IOLoop.current() # Wait up to 1 second for a notification. await condition.wait(timeout=io_loop.time() + 1) ...or a `datetime.timedelta` for a timeout relative to the current time:: # Wait up to 1 second. await condition.wait(timeout=datetime.timedelta(seconds=1)) The method returns False if there's no notification before the deadline. .. versionchanged:: 5.0 Previously, waiters could be notified synchronously from within `notify`. Now, the notification will always be received on the next iteration of the `.IOLoop`. N)rcstt|tj|_dS)N)superrrrIOLoopcurrentio_loop)r) __class__rrrrszCondition.__init__cCs.d|jjf}|jr&|dt|j7}|dS)Nz<%sz waiters[%s]>)r,r$rlen)rresultrrr__repr__vszCondition.__repr__)timeoutrcsXtj|rTddfdd }tj||fddS)zWait for `.notify`. Returns a `.Future` that resolves ``True`` if the condition is notified, or ``False`` after a timeout. N)rcsstddS)NF)rrr#r)rwaiterrr on_timeouts z"Condition.wait..on_timeoutcs S)N)remove_timeout)_)r+timeout_handlerrz Condition.wait..)rrappendrr)r* add_timeoutadd_done_callback)rr1r3r)r+rr6r2rwait|s   zCondition.waitr)nrcCsTg}x2|r6|jr6|j}|s|d8}||qWx|D]}t|dq>WdS)zWake ``n`` waiters.rTN)rpopleftrr9r)rr=waitersr2rrrnotifys   zCondition.notifycCs|t|jdS)zWake all waiters.N)r@r.r)rrrr notify_allszCondition.notify_all)N)r)r$r%r&r'rstrr0rfloatdatetime timedeltar boolr<intr@rA __classcell__rr)r,rr6s :  c@sveZdZdZddddZedddZeddd Zddd d Z ddd d Z de e e jfeddddZdS)raAn event blocks coroutines until its internal flag is set to True. Similar to `threading.Event`. A coroutine can wait for an event to be set. Once it is set, calls to ``yield event.wait()`` will not block unless the event has been cleared: .. testcode:: from tornado import gen from tornado.ioloop import IOLoop from tornado.locks import Event event = Event() async def waiter(): print("Waiting for event") await event.wait() print("Not waiting this time") await event.wait() print("Done") async def setter(): print("About to set the event") event.set() async def runner(): await gen.multi([waiter(), setter()]) IOLoop.current().run_sync(runner) .. testoutput:: Waiting for event About to set the event Not waiting this time Done N)rcCsd|_t|_dS)NF)_valuesetr)rrrrrszEvent.__init__cCsd|jj|rdndfS)Nz<%s %s>rJclear)r,r$is_set)rrrrr0szEvent.__repr__cCs|jS)z-Return ``True`` if the internal flag is true.)rI)rrrrrLsz Event.is_setcCs2|js.d|_x |jD]}|s|dqWdS)zSet the internal flag to ``True``. All waiters are awakened. Calling `.wait` once the flag is set will not block. TN)rIrr set_result)rfutrrrrJs  z Event.setcCs d|_dS)zkReset the internal flag to ``False``. Calls to `.wait` will block until `.set` is called. FN)rI)rrrrrKsz Event.clear)r1rcsptjrdSjfdd|dkrDStj|tfd}|fdd|SdS)zBlock until the internal flag is true. Returns an awaitable, which raises `tornado.util.TimeoutError` after a timeout. Ncs j|S)N)rremove)rN)rrrr7r8zEvent.wait..)quiet_exceptionscssSdS)N)rcancel)tf)rNrrr7r8) rrIrMraddr;r with_timeoutr)rr1Z timeout_futr)rNrrr<s  z Event.wait)N)r$r%r&r'rrBr0rFrLrJrKrrCrDrEr r<rrrrrs& c@sLeZdZdZeddddZddddZd eeee j dd d d Z dS) _ReleasingContextManagerzReleases a Lock or Semaphore at the end of a "with" statement. with (yield semaphore.acquire()): pass # Now semaphore.release() has been called. N)objrcCs ||_dS)N)_obj)rrVrrrrsz!_ReleasingContextManager.__init__)rcCsdS)Nr)rrrr __enter__sz"_ReleasingContextManager.__enter__zOptional[Type[BaseException]])exc_typeexc_valexc_tbrcCs|jdS)N)rWrelease)rrYrZr[rrr__exit__sz!_ReleasingContextManager.__exit__) r$r%r&r'r rrXr BaseExceptiontypes TracebackTyper]rrrrrUsrUcseZdZdZdeddfdd Zedfdd Zddd d Zde e e j fe ed d dZddddZdeeeejddddZddddZdeeeejddddZZS)raS A lock that can be acquired a fixed number of times before blocking. A Semaphore manages a counter representing the number of `.release` calls minus the number of `.acquire` calls, plus an initial value. The `.acquire` method blocks if necessary until it can return without making the counter negative. Semaphores limit access to a shared resource. To allow access for two workers at a time: .. testsetup:: semaphore from collections import deque from tornado import gen from tornado.ioloop import IOLoop from tornado.concurrent import Future # Ensure reliable doctest output: resolve Futures one at a time. futures_q = deque([Future() for _ in range(3)]) async def simulator(futures): for f in futures: # simulate the asynchronous passage of time await gen.sleep(0) await gen.sleep(0) f.set_result(None) IOLoop.current().add_callback(simulator, list(futures_q)) def use_some_resource(): return futures_q.popleft() .. testcode:: semaphore from tornado import gen from tornado.ioloop import IOLoop from tornado.locks import Semaphore sem = Semaphore(2) async def worker(worker_id): await sem.acquire() try: print("Worker %d is working" % worker_id) await use_some_resource() finally: print("Worker %d is done" % worker_id) sem.release() async def runner(): # Join all workers. await gen.multi([worker(i) for i in range(3)]) IOLoop.current().run_sync(runner) .. testoutput:: semaphore Worker 0 is working Worker 1 is working Worker 0 is done Worker 2 is working Worker 1 is done Worker 2 is done Workers 0 and 1 are allowed to run concurrently, but worker 2 waits until the semaphore has been released once, by worker 0. The semaphore can be used as an async context manager:: async def worker(worker_id): async with sem: print("Worker %d is working" % worker_id) await use_some_resource() # Now the semaphore has been released. print("Worker %d is done" % worker_id) For compatibility with older versions of Python, `.acquire` is a context manager, so ``worker`` could also be written as:: @gen.coroutine def worker(worker_id): with (yield sem.acquire()): print("Worker %d is working" % worker_id) yield use_some_resource() # Now the semaphore has been released. print("Worker %d is done" % worker_id) .. versionchanged:: 4.3 Added ``async with`` support in Python 3.5. rN)valuercs(tt||dkrtd||_dS)Nrz$semaphore initial value must be >= 0)r(rr ValueErrorrI)rra)r,rrr}szSemaphore.__init__)rcsTtt|}|jdkrdn d|j}|jr@d|t|j}d|dd|S)Nrlockedzunlocked,value:{0}z{0},waiters:{1}z <{0} [{1}]>r)r(rr0rIformatrr.)rresextra)r,rrr0s zSemaphore.__repr__cCsN|jd7_x:|jrH|j}|s|jd8_|t|PqWdS)z*Increment the counter and wake one waiter.rN)rIrr>rrMrU)rr2rrrr\s zSemaphore.release)r1rcstjdkr.jd8_tnNj|r|ddfdd }tj || fddS) zDecrement the counter. Returns an awaitable. Block if the counter is zero and wait for a `.release`. The awaitable raises `.TimeoutError` after the deadline. rrN)rcs"stdS)N)r set_exceptionr TimeoutErrorr#r)rr2rrr3sz%Semaphore.acquire..on_timeoutcs S)N)r4)r5)r+r6rrr7r8z#Semaphore.acquire..) rrIrMrUrr9rr)r*r:r;)rr1r3r)r+rr6r2racquires    zSemaphore.acquirecCs tddS)Nz0Use 'async with' instead of 'with' for Semaphore) RuntimeError)rrrrrXszSemaphore.__enter__zOptional[Type[BaseException]])typra tracebackrcCs |dS)N)rX)rrlrarmrrrr]szSemaphore.__exit__cs|IdHdS)N)rj)rrrr __aenter__szSemaphore.__aenter__)rlratbrcs |dS)N)r\)rrlrarorrr __aexit__szSemaphore.__aexit__)r)N)r$r%r&r'rGrrBr0r\rrCrDrEr rUrjrXrr^r_r`r]rnrprHrr)r,rrs ^ cs<eZdZdZd eddfdd Zddfdd ZZS) ra:A semaphore that prevents release() being called too many times. If `.release` would increment the semaphore's value past the initial value, it raises `ValueError`. Semaphores are mostly used to guard resources with limited capacity, so a semaphore released too many times is a sign of a bug. rN)rarcstt|j|d||_dS)N)ra)r(rr_initial_value)rra)r,rrrszBoundedSemaphore.__init__)rcs&|j|jkrtdtt|dS)z*Increment the counter and wake one waiter.z!Semaphore released too many timesN)rIrqrbr(rr\)r)r,rrr\s zBoundedSemaphore.release)r)r$r%r&r'rGrr\rHrr)r,rrsc@seZdZdZddddZedddZdeee j fe e dd d Z ddd d Zddd dZdeeeejddddZddddZdeeeejddddZdS)raA lock for coroutines. A Lock begins unlocked, and `acquire` locks it immediately. While it is locked, a coroutine that yields `acquire` waits until another coroutine calls `release`. Releasing an unlocked lock raises `RuntimeError`. A Lock can be used as an async context manager with the ``async with`` statement: >>> from tornado import locks >>> lock = locks.Lock() >>> >>> async def f(): ... async with lock: ... # Do something holding the lock. ... pass ... ... # Now the lock is released. For compatibility with older versions of Python, the `.acquire` method asynchronously returns a regular context manager: >>> async def f2(): ... with (yield lock.acquire()): ... # Do something holding the lock. ... pass ... ... # Now the lock is released. .. versionchanged:: 4.3 Added ``async with`` support in Python 3.5. N)rcCstdd|_dS)Nr)ra)r_block)rrrrr sz Lock.__init__cCsd|jj|jfS)Nz<%s _block=%s>)r,r$rr)rrrrr0 sz Lock.__repr__)r1rcCs |j|S)zAttempt to lock. Returns an awaitable. Returns an awaitable, which raises `tornado.util.TimeoutError` after a timeout. )rrrj)rr1rrrrjsz Lock.acquirecCs0y|jWntk r*tdYnXdS)zUnlock. The first coroutine in line waiting for `acquire` gets the lock. If not locked, raise a `RuntimeError`. zrelease unlocked lockN)rrr\rbrk)rrrrr\sz Lock.releasecCs tddS)Nz+Use `async with` instead of `with` for Lock)rk)rrrrrX&szLock.__enter__zOptional[Type[BaseException]])rlrarorcCs |dS)N)rX)rrlrarorrrr])sz Lock.__exit__cs|IdHdS)N)rj)rrrrrn1szLock.__aenter__cs |dS)N)r\)rrlrarorrrrp4szLock.__aexit__)N)r$r%r&r'rrBr0rrCrDrEr rUrjr\rXrr^r_r`r]rnrprrrrrs # )rconcurrent.futuresrrDr_tornadorrtornado.concurrentrrtypingrrr r r TYPE_CHECKINGr r __all__objectrrrrUrrrrrrrs$ kd5