U /e".@s*ddlmZddlZddlZddlZddlZddlZddlZddlm Z m Z ddl m Z ddl mZmZddlZddlmZddlmZddlmZdd lmZdd lmZdd lmZeeZGd d d e Z!Gddde!Z"Gddde Z#Gddde Z$Gddde Z%dddZ&dddZ'dS)) annotationsN)ABCabstractmethod)suppress)AnyClassVar)parse_timedelta)registry) parse_address)time)get_default_compression)HIGHEST_PROTOCOLc@s eZdZdS)CommClosedErrorN__name__ __module__ __qualname__rr9/tmp/pip-unpacked-wheel-g426oqom/distributed/comm/core.pyrsrc@s eZdZdS)FatalCommClosedErrorNrrrrrrsrc@seZdZUdZeZded<ded<ded<ded<ded <d ed <d)d d ddZe d*ddZ e d+ddZ e ddZ e ddZ e ddZee ddddZee dddd Zed!d"Zed#d$Zed%d&Zd'd(ZdS),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 deserializeT)rcCs4|j|d|_d|_i|_i|_i|_||_dS)NT)raddZ allow_offloadrrrrr)selfrrrr__init__4s z Comm.__init__NcsdS)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 Z deserializersrrrread?sz Comm.readcsdS)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 msgZ serializersZon_errorrrrwriteMsz Comm.writecsdS)z Close the communication cleanly. This will attempt to flush outgoing buffers before actually closing the underlying transport. This method returns a coroutine. Nrr rrrclose\sz Comm.closecCsdS)z Close the communication immediately and abruptly. Useful in destructors or generators' ``finally`` blocks. Nrr%rrrabortesz Comm.abortcCsdS)z$Return whether the stream is closed.Nrr%rrrclosedlsz Comm.closedstr)returncCsdS)z;The local address. For logging and debugging purposes only.Nrr%rrr local_addresspszComm.local_addresscCsdS)zz (closed))format __class__rr(rr+r,r%rrr__repr__sz Comm.__repr__)T)N)NN)rrr__doc__weakrefWeakSetr__annotations__r!rr"r$r&r'r(propertyr+r,r- staticmethodr5r=rArrrrr!s>           rc@speZdZeddZeddZeeddZeeddZd d Z d d Z d dZ dddddddZ dS)ListenercsdS)z; Start listening for incoming connections. Nrr%rrrstartszListener.startcCsdS)z Stop listening. This does not shutdown already established communications, but prevents accepting new ones. Nrr%rrrstopsz Listener.stopcCsdS)z8 The listening address as a URI string. Nrr%rrrlisten_addressszListener.listen_addresscCsdS)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'. Nrr%rrrcontact_addressszListener.contact_addresscs|IdH|SNrIr%rrr __aenter__szListener.__aenter__cs |}t|r|IdHdSrM)rJinspect isawaitable)r exc_type exc_value tracebackfuturerrr __aexit__s zListener.__aexit__csfdd}|S)NcsIdHSrMrNrr%rr_szListener.__await__.._) __await__)r rWrr%rrXs zListener.__await__Nrzdict[str, Any] | NoneNone)commhandshake_overridesr*c s||p i}tjd}t|dd}z6tj|||dIdHtj||dIdH}WnVt k r}z8t t | IdHW5QRXt d|d|W5d}~XYnX||_ |j|j d<||_|j|jd<||j|j |_dS)N!distributed.comm.timeouts.connectsecondsdefaulttimeoutzComm z closed.address)r5daskconfiggetrasynciowait_forr$r" Exceptionrr&rrr,rr+r=r)r rZr[rra handshaker<rrr on_connections$   $  zListener.on_connection)N) rrrrrIrJrFrKrLrOrVrXrjrrrrrHs    rHc@seZdZedddZdS) ConnectorTcsdS)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 rbrrrrconnectszConnector.connectN)T)rrrrrlrrrrrksrkTc sDdkrtjdtddt|\}}t|}|}d} tfdd} d} d} t d |d } d}| dkrJz8t j |j |fd |i|t| | d IdH} WqbWqvtk rYqvt jtfk rF}zR|}} t| | d | }td|}| d7} t d||t |IdHW5d}~XYqvXqvtd|dd|| |ppi}z6t | | IdH}t | || IdHWn^tk r}z>tt| IdHW5QRXtd|dd|W5d}~XYnX|| _| j| jd<|| _| j| jd<| | j| j| _ | 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. Nr\r]r^cs}td|tS)Nr)maxr )deadlinerIrarr time_leftszconnect..time_leftg{Gz?rzEstablishing connection to %srr`z7Could not connect to %s, waiting for %s before retryingzTimed out trying to connect to z after z sz/Timed out during handshake while connecting to rb)!rcrdrerr r get_backendZ get_connectorr loggerdebugrfrgrlr6r TimeoutErrorOSErrorrandomuniformsleepr5r"r$rhrr&rZ _peer_addrrZ _local_addrr=r)addrrarr[Zconnection_argsschemelocbackendZ connectorrZrpZ backoff_baseattemptZintermediate_capZactive_exceptionexcZ upper_capbackoffrrirrorrls~         $   rlcKsrzt|dd\}}Wn@tk rT|dr8d|}nd|}t|dd\}}YnXt|}|j|||f|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 r8rer rtZ get_listener)r|Z handle_commrkwargsr}r~rrrrlisten\s    r)NTN)T)( __future__rrfrPloggingryr3rCabcrr contextlibrtypingrrrcZ dask.utilsrZdistributed.commr Zdistributed.comm.addressingr Zdistributed.metricsr Z distributed.protocol.compressionr Zdistributed.protocol.pickler getLoggerrruIOErrorrrrrHrkrlrrrrrs8          L Z