U /ej@sddlmZddlZddlZddlZddlmZddlmZddl m Z ddl m Z ddl mZmZddlmZeeZGd d d ZGd d d ZdS) ) annotationsN) defaultdict)Hashable)parse_timedelta)Client) TimeoutError log_errors) get_workerc@sPeZdZdZddZdddddd d Zd d ZedddZedddZ d S)MultiLockExtensionaAn extension for the scheduler to manage MultiLocks This adds the following routes to the scheduler * multi_lock_acquire * multi_lock_release The approach is to maintain `self.locks` that maps a lock (unique name given to `MultiLock(names=, ...)` at creation) to a list of users (instances of `MultiLock`) that "requests" the lock. Additionally, `self.requests` maps a user to its requested locks and `self.requests_left` maps a user to the number of locks still need. Every time a user `x` gets to the front in `self.locks[name] = [x, ...]` it means that `x` now holds the lock `name` and when it holds all the requested locks `acquire()` can return. Finally, `self.events` contains all the events users are waiting on to finish. cCs>||_tt|_i|_i|_i|_|jj|j |j ddS)N)multi_lock_acquiremulti_lock_release) schedulerrlistlocksrequests requests_lefteventshandlersupdateacquirerelease)selfr r:/tmp/pip-unpacked-wheel-g426oqom/distributed/multi_lock.py__init__&s  zMultiLockExtension.__init__z list[str]rintbool)rid num_locksreturncs|jkstt|j|<t||kr0|dks4t|j|<t|fddd}t|D]r\}}j||tj|dkrZj|d8<j|dkrZj|t||dd8<dSqZdS) aRequest locks Parameters ---------- locks: List[str] Names of the locks to request. id: Hashable Identifier of the `MultiLock` instance requesting the locks. num_locks: int Number of locks in `locks` requesting Return ------ result: bool Whether `num_locks` requested locks are free immediately or not. rcstj|SN)lenr)xrrrGz3MultiLockExtension._request_locks..)keyNTF) rAssertionErrorsetr!rsorted enumeraterappend)rrrrilockrr#r_request_locks1s "z!MultiLockExtension._request_lockscCst}|D]}|j|d|kr~|j|d|j|r|j|d}|j|d8<|j|dkrd|j|<||n|j||||j|ks tq |j|=|j|=|D]}|jj |j |jqdS)aCancel/release previously requested/acquired locks Parameters ---------- locks: List[str] Names of the locks to refain. id: Hashable Identifier of the `MultiLock` instance refraining the locks. rr'N) r)rpopraddremover(rr ZloopZ add_callbackr)rrrZ waiters_readyr.Z new_firstwaiterrrr _refain_locksRs    z MultiLockExtension._refain_locksNcs||||s||jkstt}||j|<|}|dk rJt||}z:z|IdHWn&tk r|||YW dSXW5|j|=X|j |dkstdS)NFrT) r/rr(asyncioEventwaitwait_forrr4r)rrrtimeoutreventfuturerrrrqs    zMultiLockExtension.acquirecCs||j||dSr )r4r)rrrrrrszMultiLockExtension.release)NNNN)N) __name__ __module__ __qualname____doc__rr/r4rrrrrrrr s ! r c@sdeZdZdZddddddZdd d Zd d Zd dZddZddZ ddZ ddZ ddZ dS) MultiLocka3Distributed Centralized Lock Parameters ---------- names Names of the locks to acquire. Choosing the same name allows two disconnected processes to coordinate a lock. client Client to use for communication with the scheduler. If not given, the default global client will be used. Examples -------- >>> lock = MultiLock(['x', 'y']) # doctest: +SKIP >>> lock.acquire(timeout=1) # doctest: +SKIP >>> # do things with protected resource 'x' and 'y' >>> lock.release() # doctest: +SKIP Nzlist[str] | Nonez Client | None)namesclientcCsRz|p t|_Wntk r0tj|_YnX|p8g|_tj|_ d|_ dS)NF) rcurrentrB ValueErrorr rAuuidZuuid4hexr_locked)rrArBrrrrs  zMultiLock.__init__TcCsTt|}|s |dk rtdd}|jj|jjj|j|j||pDt|jd}d|_ |S)aAcquire 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". num_locks : int, optional Number of locks needed. If None, all locks are needed Examples -------- >>> lock = MultiLock(['x', 'y']) # 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)rrr9rT) rrDrBsyncr r rArr!rG)rblockingr9rresultrrrrs zMultiLock.acquirecCs2|std|jj|jjj|jd}d|_|S)z$Release the lock if already acquiredzLock is not yet acquired)rF)lockedrDrBrHr r rrG)rretrrrrs zMultiLock.releasecCs|jSr )rGr#rrrrKszMultiLock.lockedcCs ||Sr rr#rrr __enter__szMultiLock.__enter__cCs |dSr rrexc_type exc_value tracebackrrr__exit__szMultiLock.__exit__cs|IdH|Sr rMr#rrr __aenter__szMultiLock.__aenter__cs|IdHdSr rOrPrrr __aexit__szMultiLock.__aexit__cCst||jffSr )typerAr#rrr __reduce__szMultiLock.__reduce__)NN)TNN) r<r=r>r?rrrrKrNrTrUrVrXrrrrr@s *r@) __future__rr5loggingrE collectionsrcollections.abcrZ dask.utilsrZdistributed.clientrZdistributed.utilsrrZdistributed.workerr getLoggerr<loggerr r@rrrrs       x