0Fie1ddlmZddlZddlZddlZddlZddlZddlZddlm Z m Z ddl m Z m Z ddlZddlmZddlmZddlmZmZmZddlmZdd lmZdd lmZdd lmZeje Z!Gd d e"Z#Gdde#Z$Gdde Z%Gdde Z&Gdde&Z'Gdde Z( ddZ)ddZ*dS)) annotationsN)ABCabstractmethod)AnyClassVar)parse_timedelta)registry)get_address_host parse_addressresolve_address)time)get_compression_settings)HIGHEST_PROTOCOL)wait_forceZdZdS)CommClosedErrorN__name__ __module__ __qualname__5lib/python3.11/site-packages/distributed/comm/core.pyrrDrrceZdZdS)FatalCommClosedErrorNrrrrrrrrrceZdZUdZejZded<ded<ded<ded<ded <d ed <d d!d Ze d"dZ e d#dZ e dZ e dZ e dZee d$dZee d$dZed%dZedZd&dZed'dZdZdS)(Comma A message-oriented communication object, representing an established communication channel. There should be only one reader and one writer at a time: to manage current communications, even with a single peer, you must create distinct ``Comm`` objects. Messages are arbitrary Python objects. Concrete implementations of this class can implement different serialization mechanisms depending on the underlying transport's characteristics. zClassVar[weakref.WeakSet[Comm]] _instancesz str | Nonenamedict local_info remote_infohandshake_optionsbool deserializeTc|j|d|_d|_i|_i|_i|_||_dS)NT)radd allow_offloadr r"r#r$r&)selfr&s r__init__z Comm.__init__4sL D!!!! !#&rNc KdS)aW Read and return a message (a Python object). This method returns a coroutine. Parameters ---------- deserializers : dict[str, tuple[Callable, Callable, bool]] | None An optional dict appropriate for distributed.protocol.deserialize. See :ref:`serialization` for more. Nr)r* deserializerss rreadz Comm.read? rc KdS)a( Write a message (a Python object). This method returns a coroutine. Parameters ---------- msg on_error : str | None The behavior when serialization fails. See ``distributed.protocol.core.dumps`` for valid values. Nr)r*msg serializerson_errors rwritez Comm.writeMr/rc KdS)z Close the communication cleanly. This will attempt to flush outgoing buffers before actually closing the underlying transport. This method returns a coroutine. Nrr*s rclosez Comm.close\r/rcdS)z Close the communication immediately and abruptly. Useful in destructors or generators' ``finally`` blocks. Nrr6s rabortz Comm.abortercdS)z$Return whether the stream is closed.Nrr6s rclosedz Comm.closedlr:rreturnstrcdS)zThe local addressNrr6s r local_addresszComm.local_addresspr:rcdS)zThe peer's addressNrr6s r peer_addresszComm.peer_addressur:rctt|j}tt|j}||kS)z8Return True if the peer is on localhost; False otherwise)r r r@rB)r* local_ipaddr peer_ipaddrs r same_hostzComm.same_hostzsA(8J(K(KLL &t7H'I'IJJ {**rciS)z Return backend-specific information about the communication, as a dict. Typically, this is information which is initialized when the communication is established and doesn't vary afterwards. rr6s r extra_infozComm.extra_infos  rdict[str, Any]c|jrd}ntd}|ttjddt dS)aShare environment information with the peer that may differ, i.e. compression settings. Notes ----- By the time this method runs, the "auto" compression setting has been updated to an actual compression algorithm. This matters if both peers had compression set to 'auto' but only one has lz4 installed. See distributed.protocol.compression._update_and_check_compression_settings() See also -------- handshake_configuration Nzdistributed.comm.compression) compressionpythonpickle-protocol)rFrtuplesys version_infor)r*rLs rhandshake_infozComm.handshake_infosR > SKK23QRRK'C,--bqb1/   rlocalremotec dt|d|di}n"#t$r}td|d}~wwxYw|d|dkr |d|d<nd|d<|S)a7Find a configuration that is suitable for both local and remote Parameters ---------- local Output of handshake_info() in this process remote Output of handshake_info() on the remote host See also -------- handshake_info rNzYour Dask versions may not be in sync. Please ensure that you have the same version of dask and distributed on your client, scheduler, and worker machinesNrL)minKeyError ValueError)rSrToutes rhandshake_configurationzComm.handshake_configurations" !3+,f5F.G$$CC    Q     6-#8 8 8!&}!5C  !%C  s! A;Acd|jj|rdnd|jpd|j|jS)Nz<{}{} {} local={} remote={}>z (closed))format __class__rr<r r@rBr6s r__repr__z Comm.__repr__sK-44 N #;;== 0KKb IO        rT)r&r%N)NN)r=r>)r=r%)r=rI)rSrIrTrIr=rI)rrr__doc__weakrefWeakSetr__annotations__r+rr.r4r7r9r<propertyr@rBrFrHrR staticmethodr[r`rrrrr!s  3B'/2C2CJCCCC'''''    ^      ^    ^   ^ 33^3   ^X !!!^X!+++X+X    4"""\"H     rrceZdZedZedZeedZeedZdZ dZ dZ dddZ dS)Listenerc KdS)z; Start listening for incoming connections. Nrr6s rstartzListener.startr/rcdS)z Stop listening. This does not shutdown already established communications, but prevents accepting new ones. Nrr6s rstopz Listener.stopr:rcdS)z8 The listening address as a URI string. Nrr6s rlisten_addresszListener.listen_addressr:rcdS)z An address this listener can be contacted on. This can be different from `listen_address` if the latter is some wildcard address such as 'tcp://0.0.0.0:123'. Nrr6s rcontact_addresszListener.contact_addressr:rc>K|d{V|Srbrlr6s r __aenter__zListener.__aenter__s+jjll rcnK|}tj|r |d{VdSdSrb)rninspect isawaitable)r*exc_type exc_value tracebackfutures r __aexit__zListener.__aexit__sE  v & & LLLLLLLLL  rcFfd}|S)Nc@Kd{VSrbrtr6sr_zListener.__await__.._s,**,,       Kr) __await__)r*rs` rrzListener.__await__s2     qss}}rNcommrhandshake_overridesdict[str, Any] | Noner=NonecJKi||pi}||d{V|d{V}||_|j|jd<||_|j|jd<||j|j|_dS)Naddress) rRr4r.r#rBr"r@r[r$)r*rrr" handshakes r on_connectionzListener.on_connectionsN++--M2E2KM jj$$$$$$$$$))++%%%%%% $&*&7#$%)%7 "!%!=!= OT-" " rrbrrrrr=r) rrrrrlrnrgrprrrur}rrrrrrjrjs  ^   ^   ^X   ^X  HL       rrjc4eZdZd dZ d d fd Zd d ZxZS) BaseListenerr=rc,t|_dSrb)set_BaseListener__commsr6s rr+zBaseListener.__init__s"%%% rNrrrrcK|j| t||d{V |j|S#|j|wxYwrb)rr(superrdiscard)r*rrr_s rrzBaseListener.on_connections  '..t5HIIIIIIII I L  & & & &DL  & & & &s 'A""A>cn|jtc}|_|D]}|dSrb)rrr9)r*commsrs rabort_handshaking_commsz$BaseListener.abort_handshaking_comms$s>"lCEEt|  D JJLLLL  r)r=rrbr)rrrr+rr __classcell__)r_s@rrrsq((((HL'''''''rrc&eZdZeddZdS) ConnectorTc KdS)a  Connect to the given address and return a Comm object. This function returns a coroutine. It may raise EnvironmentError if the other endpoint is unreachable or unavailable. It may raise ValueError if the address is malformed. Nr)r*rr&s rconnectzConnector.connect+r/rNra)rrrrrrrrrr*s2   ^   rrTcKtjdtdt |\}}t j|}|}d} tfd} d} d} t d|d z } d}| dkr t|j |fd |i|t| |  d{V} n#t$rtjt"f$r|}|}} t| | d | zz}t%jd|}| d z } t d||tj|d{VYd}~nd}~wwxYw| dkt#d|dd|i| |pi}| |d{V| d{V}|| _| j| jd<|| _| j| jd<| | j| j| _t d|| S)z Connect to the given address (a URI such as ``tcp://127.0.0.1:1234``) and yield a ``Comm`` object. If the connection attempt fails, it is retried until the *timeout* is expired. Nz!distributed.comm.timeouts.connectseconds)defaultcLz}td|tz S)Nr)maxr )deadlinerltimeouts r time_leftzconnect..time_leftHs%7?1h'(((rg{Gz?rzEstablishing connection to %sr&)rz7Could not connect to %s, waiting for %s before retryingzTimed out trying to connect to z after z srzConnection to %s established)daskconfiggetrr r get_backend get_connectorr loggerdebugrrrVrasyncio TimeoutErrorOSErrorrandomuniformsleeprRr4r.r# _peer_addrr" _local_addrr[r$)addrrr&rconnection_argsschemelocbackend connectorrr backoff_baseattemptintermediate_capactive_exceptionexc upper_capbackoffr"rrls ` @rrr5s-+//"EFFgy999G%%KFC"6**G%%''I D FFE))))))LG LL0#666{ )++// )!! !#RR;R/RR,iikk::D #    $g. ) ) )"  ' IIKKG)DEEInQ 22G qLG LLI3PW   -(( ( ( ( ( ( ( ( ( ( ( ( (! ) )++//8 Fd F F7 F F F          $"J **Z        iikk!!!!!!I D"&/DY DO!%!1DOI!99 )D LL/555 Ks<;C88 FA2FFc  t|d\}}nD#t$r7|drd|z}nd|z}t|d\}}YnwxYwtj|}|j|||fi|S)aJ Create a listener object with the given parameters. When its ``start()`` method is called, the listener will listen on the given address (a URI such as ``tcp://0.0.0.0``) and call *handle_comm* with a ``Comm`` object for each incoming connection. *handle_comm* can be a regular function or a coroutine. T)strict ssl_contextztls://ztcp://)r rXrr r get_listener)r handle_commr&kwargsrrrs rlistenrs7#D666  777 ::m $ $ #d?DDd?D#D666  7"6**G 7 [+ H H H HHs>AA)NTNra)+ __future__rrrwloggingrrPrdabcrrtypingrrr dask.utilsrdistributed.commr distributed.comm.addressingr r r distributed.metricsr distributed.protocol.compressionrdistributed.protocol.picklerdistributed.utilsr getLoggerrrIOErrorrrrrjrrrrrrrrsk"""""" ######## &&&&&&%%%%%%XXXXXXXXXX$$$$$$EEEEEE888888&&&&&&  8 $ $     g        ?   s s s s s 3s s s l= = = = = s= = = @8&        ?COOOOdIIIIIIr