U /ebL@sddlmZddlZddlZddlZddlZddlmZddlmZm Z ddl Z ddl m Z ddl mZddlmZddlmZmZmZdd lmZdd lmZmZeeZGd d d ZGd ddeZdS)) annotationsN) TimeoutError) defaultdictdeque)parse_timedelta)PeriodicCallback)time)DeadlineSyncMethodMixin log_errors)retry_operation) get_client get_workerc@seZdZdZddZdddZdddZedd d Zd d Z d dZ edddZ edddZ ddZ ddZedddZdS)SemaphoreExtensionzAn extension for the scheduler to manage Semaphores This adds the following routes to the scheduler * semaphore_acquire * semaphore_release * semaphore_close * semaphore_refresh_leases * semaphore_register c Cs||_ttj|_t|_tt|_|jj |j |j |j |j |j|jdttttttttd|_ttjddd}t|j|d|jjd<}|ttjddd|_dS) N)semaphore_registersemaphore_acquiresemaphore_releasesemaphore_closesemaphore_refresh_leasessemaphore_value) acquire_total release_totalaverage_pending_lease_timependingz5distributed.scheduler.locks.lease-validation-intervalsdefaultzsemaphore-lease-timeout)distributed.scheduler.locks.lease-timeout) schedulerrasyncioEventeventsdict max_leasesleaseshandlersupdatecreateacquirereleasecloserefresh_leases get_valueintfloatmetricsrdaskconfiggetr_check_lease_timeoutZperiodic_callbacksstart lease_timeout)selfrZvalidation_callback_timepcr99/tmp/pip-unpacked-wheel-g426oqom/distributed/semaphore.py__init__"sF     zSemaphoreExtension.__init__NcCst|j|SN)lenr%r7namer9r9r:r-MszSemaphoreExtension.get_valuecCsP||jkr(t|tst|||j|<n$||j|krLtd||j|fdS)Nz)Inconsistent max leases: %s, expected: %s)r$ isinstancer.AssertionError ValueError)r7r?r$r9r9r:r(Qs   zSemaphoreExtension.createcCsXt}td||||D]8}||j|krDtd|d|d||j||<qdS)Nz'Refresh leases for %s with ids %s at %szRefreshing an unknown lease ID z for z. This might be due to leases timing out and may cause overbooking of the semaphore!This is often caused by long-running GIL-holding in the task which acquired the lease.)rloggerdebugr%critical)r7r? lease_idsnowZid_r9r9r:r,^sz!SemaphoreExtension.refresh_leasescCsnd}||j|ks*t|j||j|krft}td|||||j||<|jd|d7<nd}|S)NTzAcquire lease %s for %s at %srF)r%r=r$rrCrDr0)r7r?lease_idresultrGr9r9r: _get_leaseks zSemaphoreExtension._get_leasecCs||jkrdSdS)NFT)r$r>r9r9r:_semaphore_exists{s z$SemaphoreExtension._semaphore_existscs||std|dt|tr,t|}t|}|jd|d7<t d|||j |j | | ||}|stj|j ||j d}z|IdHWqLWntk rd}YnXt d||||j|jd ||jd |jd |<|jd|d8<|S) Nz Semaphore `z` not known or already closed.rrHz1Trying to acquire %s for %s with %s seconds left.timeoutFz;Acquisition of lease %s for %s is %s after waiting for %ss.r)rL RuntimeErrorr@listtupler afterr0rCrD remainingr"clearrKr wait_forwaitrelapsed)r7r?rNrIdeadlinerJfuturer9r9r:r)sH        zSemaphoreExtension.acquirecCst||s td|ddSt|tr2t|}||jkrX||j|krX|||ntd|d|ddS)NzTried to release semaphore `z(` but it is not known or already closed.z=Tried to release semaphore but it was already released: name=z , lease_id=z4. This can happen if the semaphore timed out before.)rLrCwarningr@rQrRr%_release_valuer7r?rIr9r9r:r*s   zSemaphoreExtension.releasecCsBtd|||j||=|j||jd|d7<dS)NzReleasing %s for %srrH)rCrDr%r"setr0r]r9r9r:r\s z!SemaphoreExtension._release_valuecCst}t|j}|D]l}t|j|}td|||j||D]>}||j||}||jkrDtd||||j||dqDqdS)Nz7Validating leases for %s at time %s. Currently known %sz$Lease %s for %s timed out after %ss.)r?rI)rrQr%keysrCrDr6r\)r7rGZsemaphore_namesr?idsZ_idZtime_since_refreshr9r9r:r4s( z'SemaphoreExtension._check_lease_timeoutcCs||sdS|j|=||jkr(|j|=||jkrf|j|r^td|dt|j|t|j|=||jdkr|jd|rtd|dt|j D]\}}||kr||=qdS)zJHard close the semaphore without warning clients which still hold a lease.NzClosing semaphore z$ but there remain unreleased leases rz but there remain pending leases) rLr$r"r%warningswarnsortedRuntimeWarningr0items)r7r?_Z metric_dictr9r9r:r+s*     zSemaphoreExtension.close)N)NN)NN)NNN)NN)N)__name__ __module__ __qualname____doc__r;r-r(r r,rKrLr)r*r\r4r+r9r9r9r:rs  +    0 rc@seZdZdZd)ddZddZd d Zd d Zd dZd*ddZ d+ddZ ddZ ddZ ddZ ddZddZddZdd Zd!d"Zd#d$Zd%d&Zd'd(ZdS), SemaphoreaSemaphore This `semaphore `_ will track leases on the scheduler which can be acquired and released by an instance of this class. If the maximum amount of leases are already acquired, it is not possible to acquire more and the caller waits until another lease has been released. The lifetime or leases are controlled using a timeout. This timeout is refreshed in regular intervals by the ``Client`` of this instance and provides protection from deadlocks or resource starvation in case of worker failure. The timeout can be controlled using the configuration option ``distributed.scheduler.locks.lease-timeout`` and the interval in which the scheduler verifies the timeout is set using the option ``distributed.scheduler.locks.lease-validation-interval``. A noticeable difference to the Semaphore of the python standard library is that this implementation does not allow to release more often than it was acquired. If this happens, a warning is emitted but the internal state is not modified. .. warning:: This implementation is susceptible to lease overbooking in case of lease timeouts. It is advised to monitor log information and adjust above configuration options to suitable values for the user application. Parameters ---------- max_leases: int (optional) The maximum amount of leases that may be granted at the same time. This effectively sets an upper limit to the amount of parallel access to a specific resource. Defaults to 1. name: string (optional) Name of the semaphore to acquire. Choosing the same name allows two disconnected processes to coordinate. If not given, a random name will be generated. register: bool If True, register the semaphore with the scheduler. This needs to be done before any leases can be acquired. If not done during initialization, this can also be done by calling the register method of this class. When registering, this needs to be awaited. scheduler_rpc: ConnectionPool The ConnectionPool to connect to the scheduler. If None is provided, it uses the worker or client pool. This parameter is mostly used for testing. loop: IOLoop The event loop this instance is using. If None is provided, reuse the loop of the active worker or client. Examples -------- >>> from distributed import Semaphore ... sem = Semaphore(max_leases=2, name='my_database') ... ... def access_resource(s, sem): ... # This automatically acquires a lease from the semaphore (if available) which will be ... # released when leaving the context manager. ... with sem: ... pass ... ... futures = client.map(access_resource, range(10), sem=sem) ... client.gather(futures) ... # Once done, close the semaphore to clean up the state on scheduler side. ... sem.close() Notes ----- If a client attempts to release the semaphore but doesn't have a lease acquired, this will raise an exception. When a semaphore is closed, if, for that closed semaphore, a client attempts to: - Acquire a lease: an exception will be raised. - Release: a warning will be logged. - Close: nothing will happen. dask executes functions by default assuming they are pure, when using semaphore acquire/releases inside such a function, it must be noted that there *are* in fact side-effects, thus, the function can no longer be considered pure. If this is not taken into account, this may lead to unexpected behavior. rHNTc Csz"t}|p|j|_|p|j|_Wn2tk rTt}|p@|j|_|pL|j|_YnX|pfdtj|_||_ tj|_ t |_ d|_ d|_|r||_ttjdddd}t|j|dd} | |_|j| jdS) Nz semaphore-Trrrr)Z callback_time)rrlooprBr uuiduuid4hexr?r$idr_leasesr, _registeredregisterrr1r2r3r_refresh_leasesrefresh_callbackZ add_callbackr5) r7r$r?rtZ scheduler_rpcrmZworkerclientZrefresh_leases_intervalr8r9r9r:r;Ss:      zSemaphore.__init__cs2t|jj|j|jd|jd|jdIdHdS)Nzsemaphore register id=z name=)r?r$ operation)r rrr?r$rqr7r9r9r: _registers zSemaphore._registercKs ||jSr<)syncrz)r7kwargsr9r9r:rtszSemaphore.registercsfdd}|S)NcsjrjIdHSr<)rsr9ryr9r:create_semaphores z-Semaphore.__await__..create_semaphore) __await__)r7r}r9ryr:r~s zSemaphore.__await__cs\|jrX|jrXtd|j|j|jt|jjt |j|jd|jt |j|jfdIdHdS)Nz'%s refreshing leases for %s with IDs %sz6semaphore refresh leases: id=%s, lease_ids=%s, name=%s)rFr?rx) r,rrrCrDrqr?r rrrQryr9r9r:rus zSemaphore._refresh_leasesc s^tj}td|j|j|t|jj |j||d|j||jfdIdH}|rZ|j ||S)Nz#%s requests lease for %s with ID %sz.semaphore acquire: id=%s, lease_id=%s, name=%s)r?rNrIrx) rnrorprCrDrqr?r rrrrappend)r7rNrIrJr9r9r:_acquires&    zSemaphore._acquirecCst|}|j|j|dS)aG Acquire a semaphore. If the internal counter is greater than zero, decrement it by one and return True immediately. If it is zero, wait until a release() is called and return True. Parameters ---------- timeout : number or string or timedelta, optional Seconds to wait on acquiring the semaphore. This does not include local coroutine time, network transfer time, etc.. Instead of number of seconds, it is also possible to specify a timedelta in string format, e.g. "200ms". rM)rr{r)r7rNr9r9r:r)szSemaphore.acquirecshz0t|jj|j|d|j||jfdIdHWdStk rbtjd|j||jfddYdSXdS)Nz.semaphore release: id=%s, lease_id=%s, name=%s)r?rIrxTzRRelease failed for id=%s, lease_id=%s, name=%s. Cluster network might be unstable?)exc_infoF)r rrr?rq ExceptionrCerrorr7rIr9r9r:_releases$   zSemaphore._releasecCs<|jstd|j}td|j||j|j|j|dS)a# Release the semaphore. Returns ------- bool This value indicates whether a lease was released immediately or not. Note that a user should *not* retry this operation. Under certain circumstances (e.g. scheduler overload) the lease may not be released immediately, but it will always be automatically released after a specific interval configured using "distributed.scheduler.locks.lease-validation-interval" and "distributed.scheduler.locks.lease-timeout". zReleased too oftenz%s releases %s for %s)rI) rrrPpopleftrCrDrqr?r{rrr9r9r:r*s  zSemaphore.releasecCs|j|jj|jdS)zC Return the number of currently registered leases. r?)r{rrr?ryr9r9r:r-szSemaphore.get_valuecCs ||Sr<r)ryr9r9r: __enter__szSemaphore.__enter__cCs |dSr<r*r7exc_type exc_value tracebackr9r9r:__exit__szSemaphore.__exit__cs|IdH|Sr<rryr9r9r: __aenter__szSemaphore.__aenter__cs|IdHdSr<rrr9r9r: __aexit__szSemaphore.__aexit__cCs |j|jfSr<)r?r$ryr9r9r: __getstate__szSemaphore.__getstate__cCs|\}}|j||dddS)NF)r?r$rt)r;)r7stater?r$r9r9r: __setstate__ s zSemaphore.__setstate__cCs|j|j|jj|jdS)Nr)rvstopr{rrr?ryr9r9r:r+s zSemaphore.closecCs|jdSr<)rvrryr9r9r:__del__szSemaphore.__del__)rHNTNN)N)N)rgrhrirjr;rzrtr~rurr)rr*r-rrrrrrr+rr9r9r9r:rks0X 0  rk) __future__rr loggingrnrar collectionsrrr1Z dask.utilsrZdistributed.compatibilityrZdistributed.metricsrZdistributed.utilsr r r Zdistributed.utils_commr Zdistributed.workerr r getLoggerrgrCrrkr9r9r9r:s"       g