R?gC4SSKrSSKrSSKrSSKJrJr SSKJrJr SSK J r J r J r J r Jr SSK r \ R(aSSK JrJr /SQr"SS\5r"S S \5r"S S \5r"S S\5r"SS\5r"SS\5r"SS\5rg)N)genioloop)Future"future_set_result_unless_cancelled)UnionOptionalTypeAny Awaitable)DequeSet) ConditionEvent SemaphoreBoundedSemaphoreLockc,\rSrSrSrSSjrSSjrSrg)_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....') NcF[R"5UlSUlg)Nr) collectionsdeque_waiters _timeoutsselfs //usr/lib/python3/dist-packages/tornado/locks.py__init__!_TimeoutGarbageCollector.__init__)s#))+ cU=RS- slURS:a4SUl[R"SUR55Ulgg)Ndrc3R# UHoR5(aMUv M g7fN)done).0ws r <_TimeoutGarbageCollector._garbage_collect..2s-WAffhaas' ')rrrrrs r_garbage_collect)_TimeoutGarbageCollector._garbage_collect-sD ! >>C DN'---W-WWDM r )rrreturnN)__name__ __module__ __qualname____firstlineno____doc__rr+__static_attributes__r rrrsXr rc\rSrSrSrS\4SjrS S\\\ \ R4S\ \ 4SjjrS S\SS4S jjrSS jrS rg)r5aA 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:: import asyncio from tornado import gen 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()]) asyncio.run(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`. r.cSURR<3nUR(aUS[UR5-- nUS-$)N) __class__r/rlen)rresults r__repr__Condition.__repr__qs=..113 == ns4=='99 9F|r Ntimeoutc^^^^[5mTRRT5 U(aNSUU4Sjjn[RR 5mTR X5mTRUU4Sj5 T$)z|Wait for `.notify`. Returns a `.Future` that resolves ``True`` if the condition is notified, or ``False`` after a timeout. ch>TR5(d [TS5 TR5 gNF)r&rr+rwaitersr on_timeout"Condition.wait..on_timeouts${{}}6vuE%%'r c&>TRT5$r%remove_timeout_io_looptimeout_handles r Condition.wait..sw/E/En/Ur r-)rrappendrIOLoopcurrent add_timeoutadd_done_callbackrr@rFrMrNrEs` @@@rwaitCondition.waitwsc V$  ( ( mm++-G$00EN  $ $%U V r nc /nU(apUR(a_URR5nUR5(dUS-nURU5 U(aUR(aM_UHn[ US5 M g)zWake ``n`` waiters.r"TN)rpopleftr&rQr)rrYwaitersrEs rnotifyCondition.notifyseDMM]]**,F;;==Qv& DMMM F .vt <r cLUR[UR55 g)zWake all waiters.N)r]r<rrs r notify_allCondition.notify_alls C &'r r5r%r"r-)r/r0r1r2r3strr>rrfloatdatetime timedeltar boolrWintr]r`r4r5r rrr5sa9v#EIeX-?-?&? @A 4, = =D =(r rc\rSrSrSrS SjrS\4SjrS\4Sjr S Sjr S S jr SS \ \ \\R 4S\S4S jjrS rg)ra6An 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:: import asyncio from tornado import gen 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()]) asyncio.run(runner()) .. testoutput:: Waiting for event About to set the event Not waiting this time Done r.Nc0SUl[5UlgrC)_valuesetrrs rrEvent.__init__s  r cnSURR<SUR5(aSOS<S3$)Nr9 rmclearr:)r;r/is_setrs rr>Event.__repr__s+ NN # #[[]]E /  r cUR$)z-Return ``True`` if the internal flag is true.rlrs rrr Event.is_sets {{r cUR(dCSUlURH+nUR5(aMURS5 M- gg)zsSet the internal flag to ``True``. All waiters are awakened. Calling `.wait` once the flag is set will not block. TN)rlrr& set_result)rfuts rrm Event.sets< {{DK}}xxzzNN4(%r cSUlg)z[Reset the internal flag to ``False``. Calls to `.wait` will block until `.set` is called. FNrurs rrq Event.clears  r r@c(^^[5mTR(aTRS5 T$TRR T5 TR U4Sj5 UcT$[ R"UT5nUR U4Sj5 U$)zxBlock until the internal flag is true. Returns an awaitable, which raises `tornado.util.TimeoutError` after a timeout. Nc:>TRRU5$r%)rremove)ryrs rrOEvent.wait..s$--*>*>s*Cr cR>TR5(dTR5$S$r%)r&cancel)tfrys rrOrssxxzz3::<CtCr )rrlrxraddrUr with_timeout)rr@ timeout_futrys` @rrW Event.waits|h ;; NN4 J # CD ?J**7C8K  ) )C  r )rlrr-r%)r/r0r1r2r3rrcr>rgrrrmrqrrrdrerfr rWr4r5r rrrsi%N #  )EIeX-?-?&? @A 4r rcn\rSrSrSrS\SS4SjrSSjrSS S \\ S \\ RSS4S jr S r g)_ReleasingContextManagerizReleases a Lock or Semaphore at the end of a "with" statement. with (yield semaphore.acquire()): pass # Now semaphore.release() has been called. objr.NcXlgr%_obj)rrs rr!_ReleasingContextManager.__init__ s r cgr%r5rs r __enter__"_ReleasingContextManager.__enter__s r exc_typeOptional[Type[BaseException]]exc_valexc_tbc8URR5 gr%)rrelease)rrrrs r__exit__!_ReleasingContextManager.__exit__s r rr-)r/r0r1r2r3r rrr BaseExceptiontypes TracebackTyperr4r5r rrrsZCD 1-(,,-   r rc$^\rSrSrSrSS\SS4U4SjjjrS\4U4SjjrSSjr SS \ \ \ \ R4S\\4S jjrSS jrS S S\ \S\ \R*SS4SjrSSjrS S S\ \S\ \R*SS4SjrSrU=r$)ria 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 inited = False 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) def use_some_resource(): global inited global futures_q if not inited: inited = True # Ensure reliable doctest output: resolve Futures one at a time. futures_q = deque([Future() for _ in range(3)]) IOLoop.current().add_callback(simulator, list(futures_q)) return futures_q.popleft() .. testcode:: semaphore import asyncio from tornado import gen 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)]) asyncio.run(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. valuer.NcP>[TU]5 US:a [S5eXlg)Nrz$semaphore initial value must be >= 0)superr ValueErrorrlrrr;s rrSemaphore.__init__s&  19CD D r c>[TU]5nURS:XaSOSRUR5nUR(a%SRU[ UR55nSRUSSU5$)Nrlockedzunlocked,value:{0}z{0},waiters:{1}z <{0} [{1}]>r")rr>rlformatrr<)rresextrar;s rr>Semaphore.__repr__srg  q(H.B.I.I$++.V  ==%,,UC 4FGE##C"Iu55r c6U=RS- slUR(asURR5nUR5(d0U=RS-slUR [ U55 gUR(aMrgg)*Increment the counter and wake one waiter.r"N)rlrr[r&rxrrDs rrSemaphore.releasesf q mm]]**,F;;== q !!":4"@Ammmr r@c^^^^[5mTRS:a1T=RS-slTR[T55 T$TRR T5 U(aNSUU4Sjjn[ RR5mTRX5mTRUU4Sj5 T$)zDecrement the counter. Returns an awaitable. Block if the counter is zero and wait for a `.release`. The awaitable raises `.TimeoutError` after the deadline. rr"c>TR5(d$TR[R"55 TR 5 gr%)r& set_exceptionr TimeoutErrorr+rDsrrF%Semaphore.acquire..on_timeouts1!;;==,,S-=-=-?@))+r c&>TRT5$r%rIrKs rrO#Semaphore.acquire..sg44^Dr r-) rrlrxrrrQrrRrSrTrUrVs` @@@racquireSemaphore.acquires ;;? KK1 K   6t< =  MM  (,, !--//1!(!4!4W!I((D r c[S5e)Nz0Use 'async with' instead of 'with' for Semaphore RuntimeErrorrs rrSemaphore.__enter__sMNNr typr tracebackc$UR5 gr%r)rrrrs rrSemaphore.__exit__ r c@# UR5IShvN gN7fr%rrs r __aenter__Semaphore.__aenter__lln tbc,# UR5 g7fr%rrrrrs r __aexit__Semaphore.__aexit__ rurbr-r%)r/r0r1r2r3rhrrcr>rrrrdrerfr rrrrrrrrrr4 __classcell__r;s@rrrsbHc$6#6$EIeX-?-?&? @A + ,8O , &E//0    , & U(( )   r rcL^\rSrSrSrS S\SS4U4SjjjrS U4SjjrSrU=r $) ria&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. rr.Nc,>[TU]US9 Xlg)Nr)rr_initial_valuers rrBoundedSemaphore.__init__s u%#r cl>URUR:a [S5e[TU]5 g)rz!Semaphore released too many timesN)rlrrrr)rr;s rrBoundedSemaphore.releases+ ;;$-- -@A A r )rrbr-) r/r0r1r2r3rhrrr4rrs@rrrs*$c$$$$r rc\rSrSrSrSSjrS\4SjrSS\\ \ \ R4S\ \4SjjrSS jrSS jrS S S \\S\\R(SS4SjrSSjrS S S \\S\\R(SS4SjrSrg)riagA 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. r.Nc [SS9Ulg)Nr"r)r_blockrs rr Lock.__init__ s&Q/ r cTSURR<SUR<S3$)Nr9z _block=r:)r;r/rrs rr> Lock.__repr__s#'>>#:#:DKKHHr r@c8URRU5$)zxAttempt to lock. Returns an awaitable. Returns an awaitable, which raises `tornado.util.TimeoutError` after a timeout. )rr)rr@s rr Lock.acquires{{""7++r clURR5 g![a [S5ef=f)zrUnlock. The first coroutine in line waiting for `acquire` gets the lock. If not locked, raise a `RuntimeError`. zrelease unlocked lockN)rrrrrs rr Lock.releases2 8 KK   ! 867 7 8s3c[S5e)Nz+Use `async with` instead of `with` for Lockrrs rrLock.__enter__(sHIIr rrrrc$UR5 gr%rrs rr Lock.__exit__+rr c@# UR5IShvN gN7fr%rrs rrLock.__aenter__3rrc,# UR5 g7fr%rrs rrLock.__aexit__6rr)rr-r%)r/r0r1r2r3rrcr>rrrdrerfr rrrrrrrrrrr4r5r rrrs"H0I#IEI,eX-?-?&? @A, + ,, 8J , & U(( )    , & U(( )   r r)rrertornadorrtornado.concurrentrrtypingrrr r r TYPE_CHECKINGr r __all__objectrrrrrrrr5r rrs I88  ! IXvX,f((f(RaFaHv0v(vry(U6Ur