∋d]ddlmZddlZddlZddlZddlZddlZddlZddlZddl Z ddl Z ddl Z ddlm Z m Z ddlmZmZddlmZddlmZmZddlmZmZddlmZdd lmZddlZdd lmZdd l m!Z!m"Z"dd l#m$Z$m%Z%m&Z&m'Z'm(Z(dd l)m*Z*ddl+m,Z,m-Z-m.Z.m/Z/m0Z0ddl1m2Z2m3Z3ddl4m5Z5ddl6m7Z7m8Z8m9Z9m:Z:m;Z;ej<e=Z>e j?dkrdej@ejAzdzdz ZBndej@ejCzdz ZBe5dz ZDdZEdZFdZGdZHGdde$ZIGddeIZJdZKGdd ZLejMejNzZOejPd!d"ZQGd#d$ejRZSGd%d&e&eLZTGd'd(eTZUGd)d*eTZVGd+d,e(eLZWGd-d.eWZXGd/d0eWZYGd1d2e*ZZGd3d4eZZ[Gd5d6eZZ\dS)7) annotationsN)SSLCertVerificationErrorSSLError)AnyClassVar)sliding_window)gennetutil)IOStreamStreamClosedError) TCPClient) TCPServer)parse_timedelta)parse_host_portunparse_host_port)CommCommClosedError ConnectorFatalCommClosedErrorListener)Backend)ensure_concrete_host from_framesget_tcp_server_address host_array to_frames)pack_frames_prelude unpack_frames) MEMORY_LIMIT) ensure_ipensure_memoryviewget_ipget_ipv6nbytes) c:|rdStjd}t t |d}|j}d}||dzks Jdtd|d z}td||z |z}|||zz }|d ksJ tj d rFt d ||| t jd|d z|d zfn|t jt jd t j}t j}t j}n)#t*$rtj dkrd}d}d}nd}YnwxYw|t d||||t j|||t j|||t j||tj drFt d|d zd} |t j| |d zdSdS#t.$rtdYdSwxYw)z5 Set kernel-level TCP timeout on the stream. Nzdistributed.comm.timeouts.tcpseconds)defaultr&r)zTimeout too lowr(rwinz+Setting TCP keepalive: idle=%d, interval=%didarwiniiz7Setting TCP keepalive: nprobes=%d, idle=%d, interval=%dlinuxzSetting TCP user timeout: %d msz$Could not set timeout on TCP stream.)closeddaskconfiggetintrsocketmaxsysplatform startswithloggerdebugioctlSIO_KEEPALIVE_VALS setsockopt SOL_SOCKET SO_KEEPALIVE TCP_KEEPIDLE TCP_KEEPINTVL TCP_KEEPCNTAttributeErrorSOL_TCPOSError exception) commtimeoutsocknprobesidleintervalrDrErFTCP_USER_TIMEOUTs 4lib/python3.11/site-packages/distributed/comm/tcp.pyset_tcp_timeoutrS;s {{}}koo=>>G/'9===>>G ;DG gk ! ! !#4 ! ! ! q'Q,  D1w~'122H X' 'D !8888"A < " "5 ) ) I LLFh W W W JJv01dTk8d?2S T T T T OOF-v/BA F F F (%2 & 4 $0 ! ( ( (<8++#'L$)M"'KK#'L  (' M   WEEE dCCC xHHH < " "7 + + N LL:GdN K K K!  OOFN,Nr()r3rr8 getsocknamerI)rKs rRget_stream_addressrVvsb {{}}z $+"9"9";";BQB"?@@ zzs'A A Ac(|jv|j}t|tjr2|jr+d|jvr"t d|d|jjd|td|d|jjd||td|d||)z8 Re-raise StreamClosedError as CommClosedError. N UNKNOWN_CAzin z: ) real_error isinstancesslrreasonr __class____name__r)objexcs rRconvert_stream_closed_errorras ~!n c3< ( ( Yz Ylcj88*+W+W+W 8N+W+WRU+W+WXXXJCJJ3=+AJJSJJKKQTT0C0030011s:c0|}|r d|_dSdS)zCallback to close Dask Comm when Tornado Stream closes Parameters ---------- ref: weak reference to a Dask comm TN_closed)refrKs rR _close_commrgs, 355D  rbc4eZdZUdZejejdZ de d<de d< ddfd Z dZ dZ eddZeddZd dZd!dZejdZd"dZd#dZedZxZS)$TCPzO An established communication based on an underlying Tornado IOStream. zdistributed.comm.shardz ClassVar[int]max_shard_sizezIOStream | NonestreamTr local_addrstr peer_addr deserializeboolcd|_t|||_||_||_t j|||_ d|j _ i|_ t j |}| tjt ||dt%||dS)NF)roT)resuper__init__ _local_addr _peer_addrrkweakreffinalize_get_finalizer _finalizeratexit_extrarfset_close_callback functoolspartialrg set_nodelayrS _read_extra)selfrkrlrnrorfr]s rRrsz TCP.__init__s  [111%# !*41D1D1F1FGG!& k$!!)"3K"E"EFFF4    rbcdSNrs rRrzTCP._read_extras rbc:t|}|j|fd}|S)Nc|G|s5td||dSdSdS)NzClosing dangling stream in )r3r=warningclose)rkrs rRrwz$TCP._get_finalizer..finalizesQ!&--//!@Q@@AAA "!!!rb)reprrk)rrrws rRrxzTCP._get_finalizers/ JJ K1    rbreturnc|jSr)rtrs rR local_addresszTCP.local_addresss rbc|jSr)rurs rR peer_addresszTCP.peer_addresss rbNcnK|j}|td}tj|} ||d{V}tj||\}t |}tdtd|tztD]C\}}|||} | j } | | d{V} | | ks J| | fD t|}t||j||jd{V} n1#t $r$|tdwxYw| S#t$$rB} d|_d|_t)jst-|| Yd} ~ dSYd} ~ dSd} ~ wt.$r|wxYw)NQr(r)ro deserializers allow_offloadz aborted stream on truncated dataT)rkrstructcalcsize read_bytesunpackrrrangeOPENSSL_MAX_CHUNKSIZEr$ read_intorrrorEOFErrorabortr rer: is_finalizingra BaseException)rrrkfmtfmt_size frames_nbytesframesijchunk chunk_nbytesnmsges rRreadzTCP.readsU >!## #?3''( "("3"3H"="=======M%}S-@@ ] ..F&a)>>@UVV < <1qs $|  **511111111L(((1l*;(((( <, J&v..' $ 0"/"&"4   J J J %&HIII JJ7! 5 5 5DKDL$&& 5+D!444444444 5 5 5 5 5 5    JJLLL  s$B1E $2D.E F41F#F4messagec K|j}|tt||j|||j|jd|j|jd{V}d|D}t|}t|}tj dt||z|z}|g|}t|g|}||dz }|dkrd |g}|g} t||D]\} } | rt| } t!dt#d| t$zt$D]U\} } | | | } | j }|jt)|j| |xj|z c_V|dnf#t($r;}d|_d |_t3jst7||Yd}~n&d}~wt8$r|wxYw|S) N)sender recipient)r serializerson_errorcontextframe_split_sizec,g|]}t|Sr)r$).0fs rR zTCP.write..s333q333rbrrirbr(T)rkrrr local_info remote_infohandshake_optionsrjsumrrpackr$joinzipr!rrr _write_bufferr append_total_write_indexwriterer:rrarr)rrrrrkrrframes_nbytes_totalheadereach_frame_nbytes each_framerrrrrs rRrz TCP.write s >!## # ,#/!-( "0          43F333 !-00$V,,S&..3F"FGG&P"6"8-8 }Q//  & &hhv&&'F01M( 14]F1K1K B B-!:$B"3:!>!>J .-0EE1!!BB1!+1Q3',| !/7"3"5"55,33E:::11\A111 LL      5 5 5DKDL$&& 5+D!444    JJLLL  #"s'B>F&& H 01G&&#H c#K|jdc}|_d|_||s |r|dV|jt jn#t$rYnwxYw|j | dS#|j | wxYwdSdS)NTrb) rkrer3writingrr8shutdown SHUT_RDWRrIrydetachrrrks rRrz TCP.closeVs #k4   fmmoo  >>##, ,,s+++++ &&v'78888    &&((( &&(((     s*AB?B? B  B? B  B??/C.Nonec|jdc}|_d|_|C|s1|j|dSdSdS)NT)rkrer3ryrrrs rRrz TCP.abortis_"k4   fmmoo  O " " $ $ $ LLNNNNN    rbc|jSrrdrs rRr3z TCP.closedps |rbc|jSr)r{rs rR extra_infozTCP.extra_infoss {rbT)rkr rlrmrnrmrorp)rrmr)Nr)rr)rrp)r^ __module__ __qualname____doc__r4utils parse_bytesr5r6rj__annotations__rsrrxpropertyrrrrr coroutinerrr3r __classcell__)r]s@rRriris%)J$:$: 011%%N! 0         X X0000dJ#J#J#J#X ]]$Xrbric:eZdZdZeeejZdZdS)TLSz( A TLS-specific version of TCP. cBt||jj}|v|j|||jd\}}}t d|j |||dSdS)N)peercertcipherrz7TLS connection with %r: protocol=%s, cipher=%s, bits=%d) rirrkr8r{update getpeercertrr=r>ru)rrMrprotobitss rRrzTLS._read_extras {!   K  (8(8(:(:4;;==  Q Q Q"&+h"7 FE4 LLI        rbN) r^rrrminrrirjrrrbrRrrxsDS.0BCCN     rbrc|d}t|tjst d||S)N ssl_contextzpTLS expects a `ssl_context` argument of type ssl.SSLContext (perhaps check your TLS configuration?) Instead got )r6rZr[ SSLContext TypeError)connection_argsctxs rR_expect_tls_contextrsT   m , ,C c3> * *  $ $ $   JrbceZdZdZdS)RequireEncryptionMixincv|js/|drtd|j|zdSdS)Nrequire_encryptionzJencryption required by Dask configuration, refusing communication from/to ) encryptedr6 RuntimeErrorprefix)raddressrs rR_check_encryptionz(RequireEncryptionMixin._check_encryptions]~ /"5"56J"K"K ,7;{W7L7LO     rbN)r^rrrrrbrRrrs#rbr)typecK tj||||tS#tj$r }|jtjkrYd}~nd}~wwxYwt j|||tjd{VS)N)familyrflagsrr) r8 getaddrinfo _NUMERIC_ONLYgaierrorerrno EAI_NONAMEasyncioget_running_loop SOCK_STREAM)hostportrrrs rR _getaddrinfors !        ? 7f' ' '  ( ' ' ' ' )++77 d6(:8      s"AA  Ac(eZdZdZejfd d Zd S) _DefaultLoopResolvera Resolver implementation using `asyncio.loop.getaddrinfo`. backport from Tornado 6.2+ https://github.com/tornadoweb/tornado/blob/3de78b7a15ba7134917a18b0755ea24d7f8fde94/tornado/netutil.py#L416-L432 With an additional optimization based on https://github.com/python-trio/trio/blob/4edfd41bd5519a2e626e87f6c6ca9fb32b90a6f4/trio/_socket.py#L125-L192 (Copyright Contributors to the Trio project.) And proposed to cpython in https://github.com/python/cpython/pull/31497/ rrmrr7rsocket.AddressFamilyrlist[tuple[int, Any]]c`Kdt|||tjd{VDS)Nc"g|] \}}}}}||f Srr)rfam_rs rRrz0_DefaultLoopResolver.resolve..s6   %Q1g'N   rbr)rr8r)rrrrs rRresolvez_DefaultLoopResolver.resolves`  /;d60B000******    rbN)rrmrr7rrrr )r^rrrr8 AF_UNSPECrrrbrRrrsA  DJCS        rbrcJeZdZUeeZded<ddZdS)BaseTCPConnector)resolverzClassVar[TCPClient]clientTcK|||t|\}}|jdi|} d|vr<|j||t d{V}|jdi|d{V}n!|jj||fdt i|d{V}|r|jrt|jn`#t$r}t||Yd}~nAd}~wt$r} td| d} ~ wt$r} t| d} ~ wwxYw|jt|z} ||| |j|z|} | S)Nserver_hostname)max_buffer_sizeFrzzTLS certificate does not match. Check your security settings. More info at https://distributed.dask.org/en/latest/tls.htmlr)F)rr_get_connect_argsrconnectMAX_BUFFER_SIZE start_tlsr3errorr rarrrrrV comm_class) rrroriprkwargsrkrerrrrKs rRrzBaseTCPConnector.connects w888"7++D''::/:: 2 F**#{22o 3   0v/@@@@@@@@@@2t{2  .= AG  }} 66< 6' 555  1 1 1 'a 0 0 0 0 0 0 0 0'   &O  2 2 2&((c 1 2 &8&@&@@  M4;#8+   s0BC D'C)) D'6D D'D""D'Nr)r^rrr rrrrrrbrRrrsR"+)5I5I5K5K"L"L"LFLLLL((((((rbrc eZdZdZeZdZdZdS) TCPConnectortcp://Fc iSrrrrs rRrzTCPConnector._get_connect_args rbN)r^rrrrirrrrrbrRr!r! s2 FJIrbr!c eZdZdZeZdZdZdS) TLSConnectortls://Tc hdt|i}|dr |d|d<|S)N ssl_optionsr)rr6)rrtls_argss rRrzTLSConnector._get_connect_argssC!#6#G#GH   0 1 1 M*9:K*LH& 'rbN)r^rrrrrrrrrbrRr'r's2 FJIrbr'cheZdZ d dZdZdZdZdZd Ze d Z e d Z dS) BaseTCPListenerTNrc |||t||\|_|_||_||_||_||_|jdi||_ d|_ d|_ dS)Nr) rrrr default_host comm_handlerror_get_server_args server_args tcp_server bound_address)rrr0rorr/ default_portrs rRrszBaseTCPListener.__init__s w888,WlCC((&*040CC?CC!rbcKtddti|j|_|j|j_t tj d}tdD]y} tj |j |j|}|j|n<#t $r-}|j dks|jt"jkr|}Yd}~rd}~wwxYw||dS)Nrzdistributed.comm.socket-backlog)rbacklogrr)rrr2r3_handle_stream handle_streamr7r4r5r6rr bind_socketsrr add_socketsrIr EADDRINUSE get_host_port)rr8r socketsrr`s rRstartzBaseTCPListener.start3s#XXOXtGWXX(,(;%dkoo&GHHIIq  A  ".Itw++G444   9>>QW0@%@%@  I s/!B,, C#6#CC#cT|jdc}|_||dSdSr)r3stop)rr3s rRrBzBaseTCPListener.stopMs6&*ot# DO  ! OO      " !rbc2|jtddS)Nz,invalid operation on non-started TCPListener)r3 ValueErrorrs rR_check_startedzBaseTCPListener._check_startedRs! ? "KLL L # "rbcK|jt|ddz}|||d{V}|dStd||j|jt |z}|||||j}|j |_ | |d{Vn,#t$rt d|YdSwxYw| |d{VdS)Nr(z!Incoming connection from %r to %rz4Connection from %s closed before handshake completed)rr_prepare_streamr=r>contact_addressrVrror on_connectionrinfor0)rrkrrrKs rRr9zBaseTCPListener._handle_streamVs9+ 172A2; ??++FG<<<<<<<< > F 8'4CWXXX &8&@&@@ v}gt?OPP!/ $$T** * * * * * * * *    KKNPW X X X FF %%%%%%%%%%%sB::%C#"C#c||jt|j|_|jddS)z@ The listening address as a (host, port) tuple. Nr()rEr4rr3rs rRr>zBaseTCPListener.get_host_portisC    %!7!H!HD !"1"%%rbcH|jt|zS)z4 The listening address as a string. )rrr>rs rRlisten_addresszBaseTCPListener.listen_addressts# {.0B0B0D0DEEErbc|\}}t||j}|jt ||zS)z2 The contact address as a string. )r/)r>rr/rr)rrrs rRrHzBaseTCPListener.contact_address{sF '')) d#Dt7HIII{.tT::::rb)TTNr) r^rrrsr@rBrEr9r>rrMrHrrbrRr-r-s """"(4 MMM&&&& & & &FFXF ;;X;;;rbr-c&eZdZdZeZdZdZdZdS) TCPListenerr"Fc iSrrr$s rRr1zTCPListener._get_server_argsr%rbc K|Srr)rrkrs rRrGzTCPListener._prepare_streams  rbN) r^rrrrirrr1rGrrbrRrPrPsA FJIrbrPc&eZdZdZeZdZdZdZdS) TLSListenerr(Tc (t|}d|iS)Nr*)r)rrrs rRr1zTLSListener._get_server_argss!/22s##rbc K |d{V|S#t$r>}td|j|t |ddp|Yd}~dSd}~wwxYw)Nz7Listener on %r: TLS handshake failed with remote %r: %srY)wait_for_handshakerIr=rrMgetattr)rrkrrs rRrGzTLSListener._prepare_streams ++-- - - - - - - -M    NNI#<..3!           s A(3A##A(N) r^rrrrrrr1rGrrbrRrTrTsA FJI$$$     rbrTc2eZdZdZdZdZdZdZdZdS)BaseTCPBackendc*|Sr)_connector_classrs rR get_connectorzBaseTCPBackend.get_connectors$$&&&rbc "|j|||fi|Sr)_listener_class)rloc handle_commrors rR get_listenerzBaseTCPBackend.get_listeners!#t#CkUU_UUUrbc,t|dS)Nrrrr`s rRget_address_hostzBaseTCPBackend.get_address_hostss##A&&rbc t|Srrdres rRget_address_host_portz$BaseTCPBackend.get_address_host_portss###rbc`t|\}}tt||Sr)rrr )rr`rrs rRresolve_addresszBaseTCPBackend.resolve_addresss)$S)) d 4$777rbct|\}}t|}d|vrt|}nt|}t |dS)N:)rr r#r"r)rr`rr local_hosts rRget_local_address_forz$BaseTCPBackend.get_local_address_forsO$S)) d $;;!$JJJ T222rbN) r^rrr]rbrfrhrjrnrrbrRrZrZsq'''VVV '''$$$88833333rbrZceZdZeZeZdS) TCPBackendN)r^rrr!r\rPr_rrbrRrprp#!OOOrbrpceZdZeZeZdS) TLSBackendN)r^rrr'r\rTr_rrbrRrsrsrqrbrs)] __future__rrctypesrr}loggingr8r[rr:rvrrtypingrrtlzrtornador r tornado.iostreamr r tornado.tcpclientr tornado.tcpserverrr4 dask.utilsrdistributed.comm.addressingrrdistributed.comm.corerrrrrdistributed.comm.registryrdistributed.comm.utilsrrrrrdistributed.protocol.utilsrrdistributed.systemrdistributed.utilsr r!r"r#r$ getLoggerr^r= version_infosizeofc_intrc_size_trrSrVrargrirrrAI_NUMERICHOSTAI_NUMERICSERVrrrResolverrrr!r'r-rPrTrZrprsrrbrRrsn""""""  22222222  88888888'''''''''''' &&&&&&JJJJJJJJ.-----JIIIIIII++++++TTTTTTTTTTTTTT  8 $ $ g=6=#>#>>!CaG=6=#A#AAAE"8A8A8Av    ; ; ;   VVVVV$VVVr#,%(== 4:3E0     7+   :+++++y"8+++\#     #   d;d;d;d;d;h 6d;d;d;N     /   /033333W333>"""""""" """"""""""rb