U /eN@sddlmZddlZddlZddlZddlZddlZddlZddlZddl Z ddl Z ddl Z ddl m Z mZddlmZmZddlmZddlmZddlmZmZmZmZmZmZmZddlZddlm Z dd l!m"Z"ddl#Z#dd l$m%Z%dd l&m'Z'm(Z(dd l)m*Z*m+Z+m,Z,m-Z-m.Z.m/Z/m0Z0dd l1m2Z2ddl3m4Z4ddl5m6Z6ddl7m8Z8m9Z9m:Z:m;Z;mm?Z?e?dZ@edZAedZBeeeeBfZCGdddeZDddeDDeD_EGdddeFZGeHeIZJddZKe%e#jLMdddZNe#jLMd ZOd!d"d#d$d%ZPGd&d'd'ZQGd(d)d)eRZSd*d+d*d,d-d.ZTGd/d0d0eQZUGd1d2d2ZVd3d4ZWd5ddd6d7d"d8d9d:ZXdXd;d<ZYGd=d>d>ZZGd?d@d@Z[GdAdBdBZ\dCdDZ]dEdFdGdHdIZ^GdJdKdKeZ_dYdEdMdKdNdOdPZ`dZdQdRdSdTdUdVdWZadS)[) annotationsN) defaultdictdeque) Container Coroutine)Enum)partial) TYPE_CHECKINGAnyCallableClassVar TypedDictTypeVarfinal)merge)IOLoop)parse_timedelta)profileprotocol)CommCommClosedErrorconnectget_address_host_portlistennormalize_addressunparse_host_port)PeriodicCallback)time) SystemMonitor) NoOpAwaitable get_traceback has_keywordiscoroutinefunctionrecursive_to_dicttruncate_exception) ParamSpecPRTc@sDeZdZdZdZdZdZdZdZdZ dZ d Z d Z d Z d Zd ZdZdS)Statusa  This Enum contains the various states a cluster, worker, scheduler and nanny can be in. Some of the status can only be observed in one of cluster, nanny, scheduler or worker but we put them in the same Enum as they are compared with each other. undefinedcreatedinitstartingrunningpausedstoppingstoppedclosingclosing_gracefullyclosedfailed dont_replyN)__name__ __module__ __qualname____doc__r*r+r,r-r.r/r0r1r2r3r4r5r6r;r;4/tmp/pip-unpacked-wheel-g426oqom/distributed/core.pyr)9sr)cCsi|] }|j|qSr;)name).0sr;r;r< Psr@c@s eZdZdS) RPCClosedNr7r8r9r;r;r;r<rASsrAcsfdd}|S)NcsdSNr;argskwargsexcr;r<_raise[szraise_later.._raiser;)rHrIr;rGr< raise_laterZs rJzdistributed.admin.tick.limitmsdefaultzdistributed.admin.pdb-on-errr bool)funcreturncCsTt|}t|j}|r(|ddkr(dS|rP|ddkrPtd|dtdSdS)NrcommTstreamzCalling the first argument of a RPC handler `stream` is deprecated. Defining this argument is optional. Either remove the argument or rename it to `comm` in .F)inspect signaturelist parameterswarningswarn FutureWarning)rOsigparamsr;r;r< _expects_commhs   r]c@s$eZdZdZeZdZddZdS)_LoopBoundMixinz@Backport of the private asyncio.mixins._LoopBoundMixin from 3.11Nc CsPt}|jdkr4|j|jdkr*||_W5QRX||jk rLt|d|S)N# is bound to a different event loop)asyncioget_running_loop_loop _global_lock RuntimeError)selfloopr;r;r< _get_loops   z_LoopBoundMixin._get_loop) r7r8r9r: threadingLockrcrbrgr;r;r;r<r^xsr^c@s eZdZdS)AsyncTaskGroupClosedErrorNrBr;r;r;r<rjsrjzCallable[P, Coro[T]]float)corofuncdelayrPcsddddfdd }|S)zXDecorator to delay the evaluation of a coroutine function by the given delay in seconds.P.argsP.kwargsr()rErFrPcs tIdH||IdHSrC)r`sleeprDrlrmr;r<wrappersz_delayed..wrapperr;)rlrmrrr;rqr<_delayedsrsc@sveZdZUdZded<ddddZdd d dd d d Zddd d ddddZddddZddddZ ddZ dS)AsyncTaskGroupzKCollection tracking all currently running asynchronous tasks within a grouprNr4NonerPcCsd|_t|_dS)NF)r4set_ongoing_tasksrer;r;r<__init__szAsyncTaskGroup.__init__zCallable[P, Coro[None]]rnro)afuncrErFrPcOs@|jrtd||||}||jj|j|dS)aTSchedule a coroutine function to be executed as an `asyncio.Task`. The coroutine function `afunc` is scheduled with `args` arguments and `kwargs` keyword arguments as an `asyncio.Task`. Parameters ---------- afunc Coroutine function to schedule. *args Arguments to be passed to `afunc`. **kwargs Keyword arguments to be passed to `afunc` Returns ------- None Raises ------ AsyncTaskGroupClosedError If the task group is closed. zHCannot schedule a new coroutine function as the group is already closed.N)r4rjrg create_taskadd_done_callbackrxremoveadd)rer{rErFtaskr;r;r< call_soons zAsyncTaskGroup.call_soonrk)rmr{rErFrPcOs|jt||f||dS)aSchedule a coroutine function to be executed after `delay` seconds as an `asyncio.Task`. The coroutine function `afunc` is scheduled with `args` arguments and `kwargs` keyword arguments as an `asyncio.Task` that is executed after `delay` seconds. Parameters ---------- delay Delay in seconds. afunc Coroutine function to schedule. *args Arguments to be passed to `afunc`. **kwargs Keyword arguments to be passed to `afunc` Returns ------- The None Raises ------ AsyncTaskGroupClosedError If the task group is closed. N)rrs)rermr{rErFr;r;r< call_laters!zAsyncTaskGroup.call_latercCs d|_dS)znCloses the task group so that no new tasks can be scheduled. Existing tasks continue to run. TNr4ryr;r;r<closeszAsyncTaskGroup.closec s|t|}d}|j|h}r||D] }|q.zt|IdHWqtjk rx}z|}W5d}~XYqXq|dk r|dS)zClose the group and stop all currently running tasks. Closes the task group and cancels all tasks. All tasks are cancelled an additional time for each time this task is cancelled. N)rr` current_taskrgrxcancelwaitCancelledError)rererrZ tasks_to_stoprer;r;r<stops zAsyncTaskGroup.stopcCs t|jSrC)lenrxryr;r;r<__len__szAsyncTaskGroup.__len__N) r7r8r9r:__annotations__rzrrrrrr;r;r;r<rts ##rtc @seZdZdZdZdZdWddZed d d d Zej d d ddd Zedd ddZ edd ddZ edd ddZ edd ddZ dd ddZddZddZd d!Zed"d#Zd$d%Zd&d'Zd(d)Zd*d+Zed,d-Zd.d/Zd0d1Zed2d d3d4Zed2d d5d6Zed7d8Zed9d:Zed;d<Zd=d d>d?Z d@dAdBdCdDdEdFZ!dXdGdHZ"dYdIdJZ#dKdLZ$dMdNZ%dZdOdPZ&d[dQdRZ'd2dSd dTdUdVZ(dS)\ServeraODask Distributed Server Superclass for endpoints in a distributed cluster, such as Worker and Scheduler objects. **Handlers** Servers define operations with a ``handlers`` dict mapping operation names to functions. The first argument of a handler function will be a ``Comm`` for the communication established with the client. Other arguments will receive inputs from the keys of the incoming message which will always be a dictionary. >>> def pingpong(comm): ... return b'pong' >>> def add(comm, x, y): ... return x + y >>> handlers = {'ping': pingpong, 'add': add} >>> server = Server(handlers) # doctest: +SKIP >>> server.listen('tcp://0.0.0.0:8000') # doctest: +SKIP **Message Format** The server expects messages to be dictionaries with a special key, `'op'` that corresponds to the name of the operation, and other key-value pairs as required by the function. So in the example above the following would be good messages. * ``{'op': 'ping'}`` * ``{'op': 'add', 'x': 10, 'y': 20}`` rNTc  s| dk rtjdtddtj_jjjj d_ j ||dkrht j dtjg}|_i_j |pitjdtt_d_d_d_d_i_|_t_d_ t!_"t#$_%g_&t'(_)_*t+j)dsft j dr\t,-j)d d fd d } t.j/d t j dt j d| dj)_.n t0j)_.z$ddl1m2} t3t4| j)d_5Wnt6k rd_5YnXt3t7_8t3t7_9ddl1m:} t3t4| j)d_ i_;t<jj t=t j dd}|j;d<t>_?d_@d_At>_Bt=t j ddd_CjC_Dt<jEjCdj;d<t<jFt=t j ddj;d<d_Gfdd}j)H|t#I_Jd_KtL|||||| d _Md!_NdS)"Nz=The io_loop kwarg to Server is ignored and will be deprecated stacklevel)identityechoZconnection_streamZ dump_statezdistributed.%s.blocked-handlers-rz"distributed.worker.profile.enabledrNrvcs}|dkp|jSrC) asyncio_loopZ is_closedrf)refr;r<rhszServer.__init__..stop)z profile.pyz selectors.pyz#distributed.worker.profile.intervalz distributed.worker.profile.cycle)Zomitintervalcyclerr)Digestr)Counterz)distributed.admin.system-monitor.intervalimonitorzdistributed.admin.tick.intervalrKrLZtickzdistributed.admin.tick.cycleZtickscst_dSrC)rh get_ident thread_idr;ryr;r<set_thread_identsz)Server.__init__..set_thread_ident)limit deserialize serializers deserializersconnection_argstimeoutserverF)OrXrYDeprecationWarningr)r,_statusrr handle_stream_to_dicthandlersupdatedaskconfiggettyper7lowerblocked_handlersstream_handlersstruuidZuuid4id_address_listen_address_port_host_commsrrrcountersrt_ongoing_background_tasksr`Event_event_finished listenersrcurrentio_looprfhasattrweakrefrrZwatchrZdistributed.counterrrrdigests ImportErrorrk digests_total digests_maxrperiodic_callbacksrrr _last_tick _tick_counter_last_tick_counter_last_tick_cycleZ_tick_interval_tick_interval_observed _measure_tick _cycle_ticksr add_callbackri _startup_lock_Server__startup_excConnectionPoolrpc_Server__stopped)rerrrZconnection_limitrrrrrrrrrpcrr;)rrer<rz1s                      zServer.__init__r)rvcCs(z|jWStk r"tjYSXdSrC)rAttributeErrorr)r*ryr;r;r<statussz Server.statusru)valuerPcCs"t|tstd|||_dS)NzExpected Status; got ) isinstancer) TypeErrorr)rerr;r;r<rs intcCs t|jS)zAThe number of total incoming connections listening to remote RPCs)rrryr;r;r<incoming_comms_openszServer.incoming_comms_opencCstdd|jDS)z9The number of connections currently handling a remote RPCcSsg|]\}}|dk r|qSrCr;)r>copr;r;r< sz0Server.incoming_comms_active..)rritemsryr;r;r<incoming_comms_activeszServer.incoming_comms_activecCs|jjS)zEThe number of connections currently open and waiting for a remote RPC)ropenryr;r;r<outgoing_comms_openszServer.outgoing_comms_opencCs|jjS)zSThe number of outgoing connections that are currently used to execute a RPC)ractiveryr;r;r<outgoing_comms_activeszServer.outgoing_comms_activezdict[str, int]csfdddDS)zA dict with various connection counters See also -------- Server.incoming_comms_open Server.incoming_comms_active Server.outgoing_comms_open Server.outgoing_comms_active csi|]}|t|qSr;)getattr)r>attrryr;r<r@sz2Server.get_connection_counters..)rrrrr;ryr;ryr<get_connection_counterss zServer.get_connection_counterscs|jIdHdS)z"Wait until the server has finishedN)rrryr;r;r<finishedszServer.finishedcCs |SrC)start __await__ryr;r;r<rszServer.__await__cs|jIdH|S)aAttempt to start the server. This is not idempotent and not protected against concurrent startup attempts. This is intended to be overwritten or called by subclasses. For a safe startup, please use ``Server.start`` instead. If ``death_timeout`` is configured, we will require this coroutine to finish before this timeout is reached. If the timeout is reached we will close the instance and raise an ``asyncio.TimeoutError`` N)rrryr;r;r< start_unsafes zServer.start_unsafec sFj4IdH&jtjkr4jdk s,tjn"jtjkrVW5QIdHRStdd}dddfdd }ztj |dIdHWntj k r}z0||IdHt t j d|d |W5d}~XYnHtk r(}z(||IdHtt j d |W5d}~XYnXtj_W5QIdHRXS) NZ death_timeout Exceptionru)rHrPcs IdHtj_|_dSrC)rr)r5rrrGryr;r<_close_on_failure sz'Server.start.._close_on_failure)rz start timed out after zs.z failed to start.)rrr)r5rAssertionErrorr,rr`wait_forr TimeoutErrorrr7rrdr.)rerrrHr;ryr<rs,   (z Server.startcs|IdH|SrCr;ryr;r;r< __aenter__s zServer.__aenter__cs|IdHdSrCrreexc_type exc_value tracebackr;r;r< __aexit__ szServer.__aexit__cCsJ|jjtk rt|dt|_|jD]}| s0| q0dS)zStart Periodic Callbacks consistently This starts all PeriodicCallbacks stored in self.periodic_callbacks if they are not yet running. It does this safely by checking that it is using the correct event loop. r_N) rrr`rardrrrvaluesZ is_runningr)rerr;r;r<start_periodic_callbacks#s zServer.start_periodic_callbackscs^|jr dSd|_t|jD] }|}t|r|qrZfdd}|j|dS)NTcstjIdHdSrC)r`gatherr;_stopsr;r<background_stops?sz%Server.stop..background_stops) rrwrrrT isawaitablerrr)relistenerfuturerr;rr<r2s    z Server.stopcCs|jr|jdSdSdSNr)rryr;r;r<rDs zServer.listenercCsPt}||j}||_|jd7_|tkr@tdt|j||d|dS)NzEvent loop was unresponsive in %s for %.2fs. This is often caused by long-running GIL-holding functions or moving large chunks of data. This can cause timeouts and instability.z tick-duration) rrrtick_maximum_delayloggerinforr7 digest_metric)renowZ tick_durationr;r;r<rKs zServer._measure_tickcCsH|js dSt}|j|}|_|j|j}|j|_|||p>d|_dS)Nr)rrrrr)rer Zlast_tick_cyclecountr;r;r<r]s zServer._cycle_ticksrcCs(|js"|jdkrtd|jj|_|jS)z~ The address this Server can be contacted on. If the server is not up, yet, this raises a ValueError. Nz(cannot get address of non-running Server)rr ValueErrorZcontact_addressryr;r;r<addressfs   zServer.addresscCs$z|jWStk rYdSXdS)z The address this Server can be contacted on. If the server is not up, yet, this returns a ``"not-running"``. z not-runningN)rrryr;r;r< address_saferszServer.address_safecCs(|js"|jdkrtd|jj|_|jS)z The address this Server is listening on. This may be a wildcard address such as `tcp://0.0.0.0:1234`. Nz/cannot get listen address of non-running Server)rrrlisten_addressryr;r;r<r}s   zServer.listen_addresscCs|jst|j\|_|_|jS)z The host this Server is running on. This will raise ValueError if the Server is listening on a non-IP based protocol. )rrrrryr;r;r<hostsz Server.hostcCs|jst|j\|_|_|jS)z The port number this Server is listening on. This will raise ValueError if the Server is listening on a non-IP based protocol. )rrrrryr;r;r<portsz Server.portzdict[str, str]cCst|j|jdS)N)rr)rr7rryr;r;r<rszServer.identityr;excludezContainer[str]dict)rrPcsH|}|j|jj|jd}||fdd|D}t|dS)zDictionary representation for debugging purposes. Not type stable and not intended for roundtrips. See also -------- Server.identity Client.dump_cluster_state distributed.utils.recursive_to_dict )rrrcsi|]\}}|kr||qSr;r;)r>kvrr;r<r@sz#Server._to_dict..r)rrrr=rrrr#)rerr extrar;rr<rs  zServer._to_dictcCs|SrCr;)redatar;r;r<rsz Server.echocs~|dkr|j}t|tr&t|j|}n&t|tr:t|}n|}t|tsLtt||j f|j |d|IdH}|j |dS)N)r allow_offload) default_portrrr default_iptuplerrr handle_commrrappend)reZ port_or_addrrrFaddrrr;r;r<rs$    z Server.listencCs8z|j|j|Wntk r0|YnXtS)zPStart a background task that dispatches new communications to coroutine-handlers)rr _handle_commrjabortr)rerQr;r;r<r s zServer.handle_commc sj|jr|dS|j}d}td|t|j||j|<|IdHz|jsz | IdH}td||Wntk r}z&t std|||WYqW5d}~XYn\t k r}zr;r;r<ras\    zServer.handle_streamcsz|jD] }|q |js~d|_t}|jD]6}|}t|r2t t |dt | |q2|r~tj|IdH|jIdH|jIdHtjddt|jDIdHW5|jXdS)NTzu is using an asynchronous `stop` method. Support for asynchronous `Listener.stop` will be removed in a future versioncSsg|] }|qSr;rr>rQr;r;r<rsz Server.close..)rrwrrrrrrTrrXrYrPendingDeprecationWarningrr`rrrrrVr)rerrrrrr;r;r<rs(     $z Server.closerk)r=rrPcCsF|jdk r|j|||j||7<t|j|||j|<dSrC)rrrmaxr)rer=rr;r;r<r s zServer.digest_metric) NNrTNNNNN)N)NT)N)N))r7r8r9r:rrrzpropertyrsetterrrrrrrrrrrrrrrrrrrrrrrrrrrr r#rrr r;r;r;r<r sx$             5 rcCsdS)Nspongr;rQr;r;r<pingpongsrFT)r&rrr)rQr&c s,|}||d<|dd}d}|dkr(|}|dk r8||d<zvz6|j||ddIdH|rj|j|dIdH}nd}Wn:tjtfk rd }Yntjk rd }YnXW5|r|n|r|IdHXt |t r(|d d kr(|j rt f|\} } } | st | | n t|d |S) zSend and recv with a Comm. Keyword arguments turn into the message response = await send_recv(comm, op='ping', reply=True) r&rFNrraise)rZon_error)rTrr%exception_text)rr$rr2r/r`rr0rrrrclean_exceptionrwith_tracebackr) rQr&rrrFr9Z please_closeZ force_closeresponse_rHtbr;r;r< send_recvs<      rNcCs@|dkr||f}n|dkr"|dks&tt|tr8t|}t|SrC)rrrrrr"iprr;r;r<addr_from_argss   rQc@seZdZUdZeZded<dZdZ dddZ d d Z d d Z d dZ ddZddZddZddZddZddZddZdS)raConveniently interact with a remote server >>> remote = rpc(address) # doctest: +SKIP >>> response = await remote.add(x=10, y=20) # doctest: +SKIP One rpc object can be reused for several interactions. Additionally, this object creates and destroys many comms as necessary and so is safe to use in multiple overlapping communications. When done, close comms explicitly. >>> remote.close_comms() # doctest: +SKIP zClassVar[weakref.WeakSet[rpc]]rr;NTcCs`i|_t||_||_tj|_||_||_|dk r6|n||_ |pBi|_ t |_ tj|dSrC)commscoerce_to_addressrrr)r.rrrrrrWeakSet_createdrrr)reargrQrrrrrr;r;r<rzs    z rpc.__init__cs|jtjkrtdt}d}|jD]"\}}|rB|||r(qLq(|D] }|j|=qP|rj|rt|j |j fd|j i|j IdH}d|_ d|j|<|S)a?Get an open communication Some comms to the ip/port target may be in current use by other coroutines. We track this with the `comms` dict :: {comm: True/False if open and ready for use} This function produces an open communication, either by taking one that we've already made or making a new one if they are all taken. This also removes comms that have been closed. When the caller is done with the stream they should set self.comms[comm] = True As is done in __getattr__ below. z RPC ClosedFrNr)rr)r4rArwrRrrrrrrrr=)reZto_clearrrQr?r;r;r< live_comm+s.      z rpc.live_commcCsdd}g}t|jD](}|r|st||}||qt|jD](}|rJ|sJt||}||qJ|j|S)NcsRz0|s.|dddIdH|IdHWntk rL|YnXdS)NrF)rr&)r4r2rr0r$rEr;r;r< _close_commTs z$rpc.close_comms.._close_comm)rVrRr4r` ensure_futurer!rUclear)rerXZtasksrQrr;r;r< close_commsSs      zrpc.close_commscsfdd}|S)Nc sjdk r"|ddkr"j|d<jdk rD|ddkrDj|d<d}z6IdH}d|_tf|d|IdH}Wn^ttfk r}z<|rt|dd|nt|dd|d|W5d}~XYnXd j |<|S) Nrrzrpc.rQrz-Exception while trying to call remote method z before comm was established.z using comm rST) rrrrWr=rNrArrrR)rFrQr:rkeyrer;r<send_recv_from_rpcks.     z+rpc.__getattr__..send_recv_from_rpcr;rer^r_r;r]r< __getattr__jszrpc.__getattr__cs4|jtjkrtj|tj|_tj|IdHSrC) rr)r4rrdiscardr`rr[ryr;r;r< close_rpcs  z rpc.close_rpccCstjdtdd|SNz1the rpc synchronous context manager is deprecatedrrrXrYrryr;r;r< __enter__s z rpc.__enter__cCst|dSrC)r`rYrcrr;r;r<__exit__sz rpc.__exit__cs|SrCr;ryr;r;r<rszrpc.__aenter__cs|IdHdSrC)rcrr;r;r<rsz rpc.__aexit__cCs\|jtjkrXtj|tj|_dd|jD}|rXtd|t ||D] }| qJdS)NcSsg|]}|s|qSr;rr@r;r;r<rszrpc.__del__..z(rpc object %s deleted with %d open comms) rr)r4rrrbrRr r7rr$)reZ still_openrQr;r;r<__del__s  z rpc.__del__cCsd|jt|jfS)Nz)rrrRryr;r;r<__repr__sz rpc.__repr__)NNTNNNN)r7r8r9r:rrTrrrRrrzrWr[rarcrfrgrrrhrir;r;r;r<rs,  ( rc@s^eZdZdZdddZeddZddZd d Zd d Z d dZ ddZ ddZ ddZ dS) PooledRPCCallzVThe result of ConnectionPool()('host:port') See Also: ConnectionPool NcCs(||_||_||_|dk r|n||_dSrC)r"poolrr)rer"rkrrr;r;r<rzszPooledRPCCall.__init__cCs|jSrCr"ryr;r;r<rszPooledRPCCall.addresscsfdd}|S)Nc sjdk r"|ddkr"j|d<jdk rD|ddkrDj|d<jjIdH}|jd}|_ztf|d|IdHWSjj|||_XdS)NrrzConnectionPool.r\) rrrrkrr"r=reuserN)rFrQZ prev_namer]r;r<r_s  z5PooledRPCCall.__getattr__..send_recv_from_rpcr;r`r;r]r<ras zPooledRPCCall.__getattr__csdSrCr;ryr;r;r<rcszPooledRPCCall.close_rpccCstjdtdd|Srdreryr;r;r<rfs zPooledRPCCall.__enter__cCsdSrCr;rr;r;r<rgszPooledRPCCall.__exit__cs|SrCr;ryr;r;r<rszPooledRPCCall.__aenter__csdSrCr;)rerEr;r;r<rszPooledRPCCall.__aexit__cCsd|jdS)Nzrlryr;r;r<riszPooledRPCCall.__repr__)NN)r7r8r9r:rzrCrrarcrfrgrrrir;r;r;r<rjs  rjc@seZdZUdZeZded<d'ddZd d Z e d d Z e d dZ ddZ d(ddZddZddZe ddddZd)ddZd*ddZdd Zd!d"Zd#d$Zd%d&ZdS)+rapA maximum sized pool of Comm objects. This provides a connect method that mirrors the normal distributed.connect method, but provides connection sharing and tracks connection limits. This object provides an ``rpc`` like interface:: >>> rpc = ConnectionPool(limit=512) >>> scheduler = rpc('127.0.0.1:8786') >>> workers = [rpc(address) for address in ...] >>> info = await scheduler.identity() It creates enough comms to satisfy concurrent connections to any particular address:: >>> a, b = await asyncio.gather(scheduler.who_has(), scheduler.has_what()) It reuses existing comms so that we don't have to continuously reconnect. It also maintains a comm limit to avoid "too many open file handle" issues. Whenever this maximum is reached we clear out all idling comms. If that doesn't do the trick then we wait until one of the occupied comms closes. Parameters ---------- limit: int The number of open comms to maintain at once deserialize: bool Whether or not to deserialize data by default or pass it through z)ClassVar[weakref.WeakSet[ConnectionPool]] _instancesrTNc Cs||_tt|_tt|_||_||_||_|dk r8|n||_|pDi|_ ||_ |r\t |nd|_ t |_|j|t|_d|_d|_tj|_dSr)rrrw availableoccupiedrrrrrrrrrrTrUror _connecting_pending_count_connecting_countr)r,r) rerrrrrrrrr;r;r<rz s      zConnectionPool.__init__cCs"|jj|j|j|jkstdS)zh Validate important invariants of this class Used only for testing / debugging N) semaphore_valuerr _n_connectingrryr;r;r< _validate+szConnectionPool._validatecCsttt|jSrC)summaprrqrryr;r;r<r3szConnectionPool.activecCs|jttt|jSrC)rryrzrrprryr;r;r<r7szConnectionPool.opencCsd|j|jt|jfS)Nz3)rrrrrryr;r;r<ri;s zConnectionPool.__repr__cCs"t|||d}t|||j|jdS)zCached rpc objectsrO)rr)rQrjrr)rer"rPrr;r;r<__call__BszConnectionPool.__call__csfdd}|S)NcsIdHSrC)rr;ryr;r<rLJsz#ConnectionPool.__await__.._)r)rerLr;ryr<rIs zConnectionPool.__await__cst|j|_tj|_dSrC)r` Semaphorerrur)r.rryr;r;r<rPszConnectionPool.startrrvcCs|jSrC)rtryr;r;r<rwUszConnectionPool._n_connectingc s|jd7_zz|jIdHzzz|jd7_t|f|pB|j|jd|jIdH}d|_t ||_ |j |_ |j ||j|||WW,WW\Stk r|jYnXW5|jd8_XWntjk rtdYnXW5|jd8_XdS)Nr)rrrzConnectionPool closing.)rsruacquirertrrrrr=rr_poolrrUrrq BaseExceptionreleaser`rr)rer"rrQr;r;r<_connectYs4     zConnectionPool._connectc s|j|}|j|}|rD|}|r4|jq|||Sq|jrV|t | ||}t |j ||fdd||j jzIdHWnDt jk r|z|IdHWntk rYnXYnX|IdHS)zE Get a Comm to the given address. For internal use. csSrC)rw)rLdoner;r<z(ConnectionPool.connect..N)rprqr4r4rurrlockedcollectr`r|rrrrr}rbrrrr)rer"rrprqrQZconnect_attemptr;rr<rws2       zConnectionPool.connectcCsp||j|kr t|jnL|j|||rD|jn(|j | ||j rl|j rl| dS)zV Reuse an open communication to the given address. For internal use. N)rqrrrrr~r4rurrprrrsr)rer"rQr;r;r<rms zConnectionPool.reusecCsZtd|j|jt|j|jD]0}|D]}t |j |j q,|q$dS)zV Collect open but unused communications, to allow opening other ones. z>Collecting unused comms. open: %d, active: %d, connecting: %dN)r r rrrrrrprrrrrrurrZ)rerRrQr;r;r<rs zConnectionPool.collectcCstd|||jkrF|j|}|D]}t|j|j q&||j kr|j |}|D]}t|j|j q`dS)z6 Remove all Comms to a given address. zRemoving comms to %sN) r r rpr4rrrrrurrq)rer"rRrQr;r;r<r~s      zConnectionPool.removecstj|_|jD] }|q|j|jfD]V}t}|rJ|| dq2t j dd|DddiIdH|D]}|j qnq(|jrt dIdHqdS)z* Close all communications rcss|]}|VqdSrCrr@r;r;r< sz'ConnectionPool.close..Zreturn_exceptionsTNg{Gzt?)r)r4rrrrrprqrwrpopitemr`rrurrp)reZconn_futdrRrLr;r;r<rs     zConnectionPool.close)rTNTNNNN)NNN)N)N)r7r8r9r:rrTrorrzrxrCrrrir{rrrwrrrmrr~rr;r;r;r<rs8 ! "     *rcCst|ttfrt|}t|SrC)rrVrrr)or;r;r<rSsrSrzlist[BaseException])rrPcCs&g}|jdk r"||j|j}q|SrC) __cause__r!)rZcausesr;r;r<collect_causess   rc@s6eZdZUded<ded<ded<ded<ded<d S) ErrorMessagerrzprotocol.Serializer1zprotocol.Serialize | NonerrHtraceback_textN)r7r8r9rr;r;r;r<rs rr.r)rrrPc Cstjd}tjj|ft|t}dt |}t ||}zt j |}t j |Wn(tk rt j tt|}YnXt |}zt j |}t j |Wn tk rt j |}YnXt||krd}n t |}|||t||dS)aProduce message to send back given an exception has occurred This does the following: 1. Gets the traceback 2. Truncates the exception and the traceback 3. Serializes the exception and traceback or 4. If they can't be serialized send string versions 5. Format a message and return See Also -------- clean_exception : deserialize and unpack message into exception/traceback z"distributed.admin.max-error-lengthrN)rr1rrHr)rrrtblibZpickling_supportinstallrr joinr format_tbr$rpickledumpsloadsrreprZ to_serializer) rrZ MAX_ERROR_LENrMZtb_textZe_bytesZ e_serializedZtb_bytesZ tb_serializedr;r;r<r3s2       r3z.BaseException | bytes | bytearray | str | Nonez(types.TracebackType | bytes | str | Noner zStuple[type[BaseException | None], BaseException | None, types.TracebackType | None])r1rrFrPc Kst|ttfr>ztj|}WqPtk r:t|}YqPXnt|trPt|}t|trztj|}Wqtt fk rd}YqXnt|trd}t|t s|dkst t|t j s|dkst t|||fS)zReraise exception and traceback. Deserialize if necessary See Also -------- error_message : create and serialize errors into message N)rbytes bytearrayrrrrrrrrrtypes TracebackTyper)r1rrFr;r;r<rI1s"     rI)NNN)r.)N)b __future__rr`rTloggingr,rhrrrrXr collectionsrrcollections.abcrrenumr functoolsrtypingr r r r r rrrZtlzrZtornado.iolooprrZ dask.utilsrZ distributedrrZdistributed.commrrrrrrrZdistributed.compatibilityrZdistributed.metricsrZdistributed.system_monitorrZdistributed.utilsrr r!r"r#r$Ztyping_extensionsr%r&r'r(ZCoror)lookupIOErrorrA getLoggerr7r rJrrr r=r]r^rdrjrsrtrrFrNrQrrjrrSrrr3rIr;r;r;r<s   $   $         p89 ):1