U /eT@sddlmZddlZddlZddlZddlmZmZddlm Z ddl m Z ddl m Z mZddlmZeeZGdd d ZGd d d ZdS) ) annotationsN) defaultdictdeque)parse_timedelta)Client) TimeoutError log_errors) get_workerc@s4eZdZdZddZed ddZed ddZdS) LockExtensionzAn extension for the scheduler to manage Locks This adds the following routes to the scheduler * lock_acquire * lock_release cCs4||_tt|_t|_|jj|j|j ddS)N) lock_acquire lock_release) schedulerrreventsdictidshandlersupdateacquirerelease)selfr r4/tmp/pip-unpacked-wheel-g426oqom/distributed/lock.py__init__s   zLockExtension.__init__Ncst|trt|}||jkr"d}n||jkrt}|j|||}|dk r`t ||}z8z|IdHWn t k rd}YWqYnXd}W5|j| }||kst Xq"|r||jkst ||j|<|S)NTF) isinstancelisttuplerasyncioEventrappendwaitwait_forpopleftAssertionErrorr)rnameidtimeoutresulteventfutureZevent2rrrr#s,     zLockExtension.acquirecCsdt|trt|}|j||kr*td|j|=|j|rX|jj |j|dj n|j|=dS)Nz#This lock has not yet been acquiredr) rrrrget ValueErrorrr ZloopZ add_callbackset)rr#r$rrrr?s  zLockExtension.release)NNN)NN)__name__ __module__ __qualname____doc__rrrrrrrrr s   r c@s\eZdZdZdddZdddZdd Zd d Zd d ZddZ ddZ ddZ ddZ dS)LockawDistributed Centralized Lock Parameters ---------- name: string (optional) Name of the lock to acquire. Choosing the same name allows two disconnected processes to coordinate a lock. If not given, a random name will be generated. client: Client (optional) Client to use for communication with the scheduler. If not given, the default global client will be used. Examples -------- >>> lock = Lock('x') # doctest: +SKIP >>> lock.acquire(timeout=1) # doctest: +SKIP >>> # do things with protected resource >>> lock.release() # doctest: +SKIP NcCs\z|p t|_Wntk r0tj|_YnX|pBdtj|_tj|_ d|_ dS)Nzlock-F) rcurrentclientr*r uuidZuuid4hexr#r$_locked)rr#r2rrrras z Lock.__init__TcCsHt|}|s |dk rtdd}|jj|jjj|j|j|d}d|_|S)a/Acquire the lock Parameters ---------- blocking : bool, optional If false, don't wait on the lock in the scheduler at all. timeout : string or number or timedelta, optional Seconds to wait on the lock in the scheduler. This does not include local coroutine time, network transfer time, etc.. It is forbidden to specify a timeout when blocking is false. Instead of number of seconds, it is also possible to specify a timedelta in string format, e.g. "200ms". Examples -------- >>> lock = Lock('x') # doctest: +SKIP >>> lock.acquire(timeout="1s") # doctest: +SKIP Returns ------- True or False whether or not it successfully acquired the lock Nz/can't specify a timeout for a non-blocking callr)r#r$r%T) rr*r2syncr r r#r$r5)rblockingr%r&rrrrksz Lock.acquirecCs6|std|jj|jjj|j|jd}d|_|S)z$Release the lock if already acquiredzLock is not yet acquired)r#r$F) lockedr*r2r6r r r#r$r5)rr&rrrrsz Lock.releasecCs|jSN)r5rrrrr8sz Lock.lockedcCs ||Sr9rr:rrr __enter__szLock.__enter__cCs |dSr9rrexc_type exc_value tracebackrrr__exit__sz Lock.__exit__cs|IdH|Sr9r;r:rrr __aenter__szLock.__aenter__cs|IdHdSr9r=r>rrr __aexit__szLock.__aexit__cCs t|jffSr9)r0r#r:rrr __reduce__szLock.__reduce__)NN)TN) r,r-r.r/rrrr8r<rBrCrDrErrrrr0Ls ' r0) __future__rrloggingr3 collectionsrrZ dask.utilsrZdistributed.clientrZdistributed.utilsrrZdistributed.workerr getLoggerr,loggerr r0rrrrs     ;