sgLUddlZddlZddlZddlZddlZddlmZddlZddlm Z m Z m Z m Z m Z mZmZddlmZmZddlmZgdZej,dZej0ej2Zej7eej8eZd e_ejAejBejEeGd d e#Z$Gd d e#Z%dZ&e'e(d<dZ)e'e(d<dZ*dZ+dZ,dZ-dZ.Gdde Z/GddZ0de dejbfdZ2de de fdZ3y)N)Optional)RendezvousClosedErrorRendezvousErrorRendezvousHandlerRendezvousInfoRendezvousParametersRendezvousStoreInfoRendezvousTimeoutError) cas_delay EtcdStore)parse_rendezvous_endpoint)EtcdRendezvousRetryableFailureEtcdRendezvousRetryImmediatelyEtcdRendezvousHandlerEtcdRendezvouscreate_rdzv_handlerz%%(levelname)s %(asctime)s %(message)sFc eZdZy)rN__name__ __module__ __qualname__g/var/www/html/venv/lib/python3.12/site-packages/torch/distributed/elastic/rendezvous/etcd_rendezvous.pyrr7rrc eZdZy)rNrrrrrr=rrriX_DEFAULT_TIMEOUT_DEFAULT_LAST_CALL_TIMEOUT i cheZdZdZdddeefdZdZdefdZd Z d Z d Z d Z defd Z defdZy)ra Implements a :py:class:`torch.distributed.elastic.rendezvous.RendezvousHandler` interface backed by :py:class:`torch.distributed.elastic.rendezvous.etcd_rendezvous.EtcdRendezvous`. ``EtcdRendezvousHandler`` uses a URL to configure the type of rendezvous to use and to pass implementation specific configurations to the rendezvous module. The basic etcd rendezvous configuration URL looks like the following :: etcd://:/?min_workers=&max_workers= # noqa: W605 -- example -- etcd://localhost:2379/1234?min_workers=1&max_workers=3 The URL above is interpreted as follows: 1. Use the rendezvous handler that is registered with the ``etcd`` scheme 2. The ``etcd`` endpoint to use is ``localhost:2379`` 3. ``job_id == 1234`` is used as the prefix in etcd (this allows one to share a common etcd server for multiple jobs so long as the ``job_ids`` are guaranteed to be unique). Note that the job id can be any string (e.g. does not need to be a number) as long as it is unique. 4. ``min_workers=1`` and ``max_workers=3`` specifies a range for membership size - Torch Distributed Elastic starts running the job as long as the cluster size is greater than or equal to ``min_workers`` and admits up to ``max_workers`` into the cluster. Below are a full list of the parameters that can be passed to etcd rendezvous: +--------------------------------------------+--------------------------+ | Parameter | Description | +============================================+==========================+ | min_workers | minimum number of | | | workers for the | | | rendezvous to be valid | +--------------------------------------------+--------------------------+ | max_workers | maximum number of | | | workers to admit | +--------------------------------------------+--------------------------+ | timeout | total timeout within | | | which next_rendezvous is | | | expected to succeed | | | (default 600s) | +--------------------------------------------+--------------------------+ | last_call_timeout | additional wait amount | | | ("last call") after min | | | number of workers has | | | been reached (defaults | | | to 30s) | +--------------------------------------------+--------------------------+ | etcd_prefix | path prefix (from etcd | | | root), inside which all | | | etcd nodes will be | | | created (defaults to | | | ``/torchelastic/p2p``) | +--------------------------------------------+--------------------------+ rdzv_implr local_addrc ||_||_y)z Args: rdzv_impl: the implementation of the rendezvous local_addr: the local address of the current node N) _rdzv_impl _local_addr)selfr$r%s r__init__zEtcdRendezvousHandler.__init__s$%rc|`yN)r'r)s r__del__zEtcdRendezvousHandler.__del__s Orreturncy)Netcdrr-s r get_backendz!EtcdRendezvousHandler.get_backendsrc|jj\}}}tjd|jj |}t j |||j}t||||S)Nz4Creating EtcdStore as the c10d::Store implementation)r%) r'rendezvous_barrierloggerinfosetup_kv_storer buildr(r)r) rdzv_versionrank world_sizestorebootstrap_store_infos rnext_rendezvousz%EtcdRendezvousHandler.next_rendezvoussn)-)K)K)M& dJ JK..|<288 %D$4$4 eT:7KLLrc |jj\}}|ddk(S#tj$rYywxYw)NstatusclosedFr'get_rdzv_stater1EtcdKeyNotFoundr)_states r is_closedzEtcdRendezvousHandler.is_closedsD 557HAu?h. .##  s $'==c8|jjyr,)r' set_closedr-s rrJz EtcdRendezvousHandler.set_closeds ""$rc |jj\}}|ddk(r|dS y#tj$rYywxYw)Nr@finalnum_workers_waitingrrBrEs rnum_nodes_waitingz'EtcdRendezvousHandler.num_nodes_waitings\ 557HAuX')233*##   s).AAc.|jjSr,)r'_run_idr-s r get_run_idz EtcdRendezvousHandler.get_run_ids&&&rc |jy#t$r)}tjdt |Yd}~yd}~wwxYw)NTz#Shutdown failed. Error occurred: %sF)rJ BaseExceptionr5warningstrr)es rshutdownzEtcdRendezvousHandler.shutdowns:  OO   NN@#a& I s AAAN)rrr__doc__rrUr*r.r2r>rHrJrNrQboolrXrrrrrXs^=~&"2& &S M%'C'$rrceZdZdZdZdZdZdZdZdZ dZ d Z d Z d Z d Zd ZdZdZdZdZdZddZdZddZdZdZddZdZy)rzWA rendezvous implementation that uses `etcd `__ as the backend store.c||_tjd|jj||_||_||_||_||_||_ d|_ d|_ |jjds|xjdz c_|jdk7r|j|j|j|jdt|j!|jdt|_ |j|jd |jj#|jddd y#t$j&$rYywxYw) NzEtcd machines: %s/ttl/rdzv/rdzv/version_counter0Fkeyvalue prevExist)clientr5r6machines_prefixrP_num_min_workers_num_max_workers_timeout_last_call_timeout_lease_run_id_stop_lease_this_rank_stopendswithcreate_path_if_not_existsget_pathCONST_RUNID_SUBROOT_TTLsetup_lease_renewalwriter1EtcdAlreadyExist)r)rhprefixrun_idnum_min_workersnum_max_workerstimeoutlast_call_timeouts rr*zEtcdRendezvous.__init__sJ  ')=)=>  / / "3#'%)"||$$S) LLC L <<3   * *4<< 8 &&t}}R'8>U&V"&":": MM" #:#;#  &&t}}W'=>  KK  MM"9:#QV  $$   s6-E$$E:9E:c|j|jj|j|jjyyr,)rosetrpr-s rr.zEtcdRendezvous.__del__ sC  " " .  # # ' ' )  % % 1  & & * * , 2rctj|jz|_ tj|jkDrttj d |j |j j|jS#t$rYnt$rtjdYnt$rtj dt$r"tj d|jt$rt$r5}tj d|tjdYd}~nd}~wwxYw/)a Main entry point for next rendezvous. This method is blocking until rendezvous succeeds or a timeout occurs. Returns: ``(rdzv_version, rank, world_size)`` Raises: RendezvousTimeoutError - timeout waiting for rendezvous RendezvousClosedError - rendezvous is or was closed while waiting RendezvousError - other persistent errors that render the rendezvous non-retryable z"Attempting to join next rendezvousNr z4Rendezvous timeout occurred in EtcdRendezvousHandlerz2Rendezvous for run_id=%s was observed to be closedz1Rendezvous attempt failed, will retry. Reason: %s)timerm_rendezvous_deadliner r5r6rpr init_phaserrsleeprrPr ExceptionrVs rr4z!EtcdRendezvous.rendezvous_barriers%)IIK$--$?!yy{T666,, KK< =# --9..224((1 1  1 )  RS(  H$,,"    OQRS 1  Es%%5B E&EAE+E  Ec |j}tj|j}tj d||ddk(rt|ddk(r|j|dS|ddk(r|j|dt|j|jdz t#t j$r,|j\}}tj d|YwxYw) a Initially, the rendezvous state is expected to be one of: 1. empty (non-existent) - in this case we try to create a new one. 2. joinable - we try to join it. 3. final - we announce ourselves as waiting, and go into monitoring mode Any other state is considered transitional, and will be retried after a short delay. Returns: ``(rdzv_version, rank, world_size)`` Raises: RendezvousClosedError - current rendezvous was/is closed EtcdRendezvousRetryableFailure - observed some intermediate state, which is best handled by retrying later z New rendezvous state created: %sz&Observed existing rendezvous state: %sr@rAjoinableversionrLr  etcd_index)try_create_rendezvousjsonloadsrfr5r6r1rwrCr join_phasehandle_existing_rendezvousrtry_wait_for_state_changerrr)active_versionrGs rrzEtcdRendezvous.init_phaseOs& I!779NJJ~334E KK:E B ?h &' ' ?j (??5#34 4 ?g %  + +E),< =0 0 &&.2K2Ka2O&P,,#$$ I$($7$7$9 !NE KK@% H  IsAB88//0 E )     --1 1eHo6S KKDi P!%t/F/F!F   & &'79K L KK:I F  23,,-=> >//0 )  0 0 = < = 0!!"2I>>rc6tjd|j||tjd|j|}t j |j }tjd|d||d|t|dfS)a6 Once the rendezvous state transitions from 'joinable' to 'frozen', we have every participant confirm their membership and setup per-member keep-alive TTL keys, and then wait for all other participants to confirm, which would then successfully conclude this rendezvous. z)All peers arrived. Confirming membership.z)Waiting for confirmations from all peers.z2Rendezvous version %s is complete. Final state: %sr participants)r5r6confirm_membershipwait_for_finalrrrflen)r)rrrrGs rrzEtcdRendezvous.confirm_phases  ?@  0)< ?@,,-=> >//0 @ )   YCn0E,FFFrc|j|}tjd|j|j |tjdy)z Handle the case when there's an existing (state 'final) rendezvous already in place, and we have to announce ourselves waiting, and wait until the next rendezvous opportunity. z5Added self to waiting list. Rendezvous full state: %szBPreviously existing rendezvous state changed. Will re-try joining.N)announce_self_waitingr5r6rfwait_for_rendezvous_to_free)r)r active_states rrz)EtcdRendezvous.handle_existing_rendezvoussL112BC  C\EWEW  (()9: P rc |jj|jdtjddidt } |jj |jd}tt|jdz|_ |jj||jj|jd |jd d d |jj|jdtjd |jgd|jS#tjtjf$r}td|d }~wwxYw)z Create new rendezvous state or raise an exception that indicates an unexpected state (e.g. already exists). Raises: RendezvousError - on unexpected state /rdzv/active_versionr@setupF)rerfrgr`rbr z?Unexpected state of EtcdRendezvousHandler, worker needs to die.N/rdzv/v_T)rerfdirrgr)r@rrrerf prev_value)rhrvrsrdumpsCONST_ETCD_SETUP_TTLgetrUintrfupdater1rDEtcdCompareFailedr test_and_set)r)rversion_counterrWs rrz$EtcdRendezvous.try_create_rendezvoussY** 45**h01$ +  "kkoodmm)>(?@A  {{'' 45**(.44$&&++(  )$$d&<&<= !Q  s A+E#E;* E66E;c t|j\}}|ddk7r td|d|k7r tdt |d|j ksJdt |d}|dj |d}t |d|j k(rd |d<g|d <t}n!t |d|jk\rt} |jj|jd tj||j| }||fS#t j"$rt$j'd YnwxYwV)!Helper method for the join phase.r@rzNRendezvous state became non-joinable before we could join. Must join next one.r6Rendezvous version changed. Must try join the new one.rz>Logic error: joinable rendezvous should always have space leftNfrozen keep_alivesrrerfrr`z*Join rendezvous CAS unsuccessful, retrying)r rCrrrrlappendCONST_ETCD_FROZEN_TTLrk!CONST_ETCD_JOINABLE_EPHEMERAL_TTLrhrrsrrrfr1rr5r6)r)rrrGrset_ttls rrzEtcdRendezvous.join_rendezvouss| K$($7$7$9 !NEX*,4* Y#334L E.)*T-B-BB PO PBE.12I . ! ( ( 3&*G5()T-B-BB"*h')m$/U>*+t/D/DD; J!%!9!9 &<=**U+-33 ":"&y00)) J HI JWsAD(((EEc|j\}} |ddk(r |d|k(r|S|ddk(r+|d|k(r#|j|jdz\}}n tdQ)rr@rrrr r>Rendezvous state transition no longer possible. Must re-enter.rCrrrr)rrrGs rrzEtcdRendezvous.wait_for_peers3s $ 3 3 5X(*uY/?CS/S%%xJ.53CGW3W(,(F(F-881<)G)% 5Trc t|j\}}|ddk7r td|d|k7r td|jd|d|}|jj |d t |d j|t|d t|d k(r d |d<d|d<d}nd} |jj|jdtj||j|rd nt}|j|t |_|S#t j"$rt$j'dYnwxYw^)$Helper method for the confirm phase.Tr@rzDRendezvous no longer frozen, before we confirmed. Must join next onerrrz/rank_N)rfr`rrrLrrMFrrr_z-Confirm membership CAS unsuccessful, retrying)r rCrrsrhrCONST_WORKER_KEEPALIVE_TTLrrrrrrfrrurpr1rr5r6)r)rrrrGthis_lease_keyfinalizes rrz!EtcdRendezvous.confirm_membershipGs K$($7$7$9 !NEX(*4)Y#334L"]]+,F9+>N KKOON$BR.R%%xH,y1AEU1U(,(F(F-881<)G)% 5Trc| t|j\}}|ddk7s|d|k7rt|dxxdz cc< |jj |j dt j||j}|S#tj$rtjdYnwxYw) z Announce this worker is waiting (via num_workers_waiting counter) to join next rendezvous, but only if state and version match. r@rLrrMr rrz3Announce self as waiting CAS unsuccessful, retrying) r rCrrhrrsrrrfr1rr5r6rs rrz$EtcdRendezvous.announce_self_waitings  K$($7$7$9 !NEX')U9-=AQ-Q44 ' (A - ( S!%!9!9 &<=**U+-33":" &%)) S QR S%sA B(B:9B:c|j\}} |ddk7s|d|k7ry|jj|jd|}|jDcgc]}|j }}|dD]}||vst jd|t jd |t jd |jj|jd |j t jd |y t|jtjz ddz}|jj|jd|jdzd|tj|jkDrt$|j\}}cc}w#tj tj"f$rYfwxYw)a When there's an existing valid rendezvous in state 'final', we have to wait until the next opportunity to join. Such opportunity may come from: 1. rendezvous state changed by someone else, in which case we unblock and retry. 2. rendezvous becomes invalid because at least one member failed to renew their leased keep_alive node. We detect this, and destroy the rendezvous. Tr@rLrNrrz!Keep-alive key %s is not renewed.z%Rendezvous version %s is incomplete. zAttempting to destroy it.r)re prevValuez-Destroyed rendezvous version %s successfully.?rar )reindex recursiver|)rCrhrrschildrenrer5r6deleterfmaxrrwatchrr1EtcdEventIndexClearedEtcdWatchTimedOutr ) r)rrrG alive_memberschkeep_alive_keysreoverall_timeouts rrz*EtcdRendezvous.wait_for_rendezvous_to_frees!% 3 3 5X')U9-=AQ-Q !KKOO )9(:;<M1>0F0FG"rvvGOG]+ o-KK CSIKK?AQKK ;< KK&& MM*@A"0"6"6' KKG( 1 8 11DIIK?EK  !! g.(33a7"+ "yy{T666,,$($7$7$9 !NEoHP..0F0FG  s!F9A)F>>#G$#G$c|j\}} |ddk(r |d|k(ry|ddk7s|d|k7r tdtj|k\r[d|d<g|d< |jj |j dt j||jt }y |jj |j d|j|jt }t!td z |tjz d z}|j#|j$d z|\}}5#tj$r6tjd t|j\}}Y~wxYw#tj$r5tjdt|j\}}YwxYw)a After we reach min number of workers, one particular worker takes on the responsibility of waiting an additional timeout before closing the join window. If the worker responsible for this fails, the rendezvous will be destroyed due to expiring TTL, and the other participants will re-rendezvous. Here we expect to see state Exit gracefully if either: 1. state becomes 2. timeout happens (reaching deadline), in which case we try the transition to Exit with exception otherwise. r@rrNrrrrrz6Join last-call transition CAS unsuccessful. Will retryrr )rr|z7Join last-call TTL refresh CAS unsuccessful, will retry)rCrrrhrrsrrrfrr1rr5r6r rminrr)r)rdeadlinerrGr|s rrz$EtcdRendezvous.handle_join_last_calls !% 3 3 5X(*uY/?CS/SX*,i0@DT0T4T yy{h&"*h')m$%)[[%=%= MM*@A"jj/#1#7#71 &>&N >!%!9!9 &<=(..-339 ":"59tyy{*S0)-(F(F-881 UV (,(;(;(=% >s(#AE 4BF AFFAGGcP |j\}}|ddk(ryd|d< |jj|jdt j ||j y#tj$r"tjdtYnwxYw)z Mark rendezvous 'closed' for current run_id, which is used to signal other participants to not attempt to perform (re-)rendezvous. This is useful when one of the workers decides the job is complete. r@rANrrz%Set closed CAS unsuccessful, retrying) rCrhrrsrrrfr1rr5r6r rs rrJzEtcdRendezvous.set_closed3s $($7$7$9 !NEX(*&E(O  (( &<=**U+-33) ))  CD  !sA A//2B$#B$c|jj|jd}|tj|j fS)Nr)re)rhrrsrrrf)r)rs rrCzEtcdRendezvous.get_rdzv_stateMs:T]];Q-RStzz.*>*>???rNct|jtjz ddz}||n t||} |jj |j d||tj|jkDrt|jS#tjtjf$rY\wxYw)Nrrr)rr|) rrrrrhrrsr1rrr rC)r)rr|rs rrz(EtcdRendezvous.try_wait_for_state_changeQsd77$))+EsKcQ%,_/#g:W  KK   45ZQX   99;22 2( (""$$**D,B,BC   s-B$$#C  C ch|jdsd|z}|jd|j|S)Nr]run_) startswithrjrP)r)paths rrszEtcdRendezvous.get_pathcs4s#:D,,tDLL>$88rcx |jj|ddd|y#tj$rYywxYw)NTF)rerfrrgr`)rhrvr1rw)r) full_pathr`s rrrz(EtcdRendezvous.create_path_if_not_existsisC  KK  Ttu#  $$   s #99cd}tj}tj||j|||f}d|_|j |S)Nc |j|||j |dz ry+#tj$rYyt$rYywxYw)Nr_r)r|)refreshr1rDConnectionRefusedErrorwait)rhrr` stop_events r lease_workerz8EtcdRendezvous.setup_lease_renewal..lease_workerws]NN4SN1??37?3++-s-A A  A )targetargsT) threadingEventThreadrhdaemonstart)r)rr`rlease_stop_event lease_threads rruz"EtcdRendezvous.setup_lease_renewalqsU %??, ''t{{IsDT&U # rcJ|jd|d} |jj|tj||id}y#t j $rYnwxYw |jj|}tj|j}|||< |jj|tj||j}y#t j$r-tjdtjdYnwxYw)Nr /extra_dataFrdrz+Store extra_data CAS unsuccessful, retryingg?)rsrhrvrrr1rwrrrfrrr5r6rr)r)r9rerfnode extra_datanew_extra_data_values rstore_extra_datazEtcdRendezvous.store_extra_datas }}x ~[AB ** C< 8E+J $$   .J#'::j.>.>#? (-  % ![[55**%9:)//6 ))  IJ 3 s#3A A! A!%;C!!=D! D!c$|jd|d}|jd|} |jj|}|jDcgc]}|j|k(s|}}t |dksJt |dk(r+t j|dj} || vr| |S |jj||jdzcc}w#tjtjf$rY+wxYw)Nrrr r)r)rsrhrrrerrrrfrrr1rr) r)r9rer|rnode_dirrootnrextra_data_dicts rload_extra_datazEtcdRendezvous.load_extra_datas}}x ~[AB==8L>!:;;;??8,D&*]]Daeetm!DJDz?a' '':!#"&**Z]-@-@"A/)*3//  !!$doo.A!B# E..0F0FG  sC$)C$9*C))#DDc||jd|d}|j|t|j|S)Nrz/kv) etcd_clientetcd_store_prefix)rsrrr rh)r)r9 store_paths rr7zEtcdRendezvous.setup_kv_stores8]]Xl^3#?@  &&z2T[[JOOrr,)rrrrYr*r.r4rrrrrrrrrrrrrJrCrrsrrrurrr7rrrrrsa.`-8t(-T'?RG. (3 j/Jb(.M`(S4B:HH>T4@%$9  < <8Prrparamsr/ct|jd\}}|jjd}|d}n|dk7r|dk7r t d|jjd}|!|jjd}|||f}|jjd}t j |||||d S) zICreate a new ``etcd.Client`` from the specified ``RendezvousParameters``.iK protocolhttphttpsz(The etcd protocol must be HTTP or HTTPS.certrecacertT)rr ca_certallow_reconnect)rendpointconfigr ValueErrorr1Client)rhostnameportrssl_certcert_keyr s r_create_etcd_clientrs.vENHd}}  ,H v (g"5GH H}}  (H==$$U+  !(+Hmm)G ;;    rc t|}|jdd}t|||j|j|j |j dt|j dt}t||jS)a2 Usage: :: rdzv_params = RendezvousParameters( backend="etcd", endpoint="192.168.0.42:2379", run_id="123", min_nodes=4, max_nodes=8, timeout=300, last_call_timeout=30, etcd_prefix="custom_prefix", protocol="https", cacert="/etc/kubernetes/certs/ca.crt", cert="/etc/kubernetes/certs/client.crt", key="/etc/kubernetes/certs/client.key") # -- or -- rdzv_params = RendezvousParameters( backend="etcd", endpoint="192.168.0.42:2379", run_id="123", min_nodes=4, max_nodes=8) etcd_rdzv_handler = create_etcd_rendezvous_handler(rdzv_params) Where: run_id - unique id for this training job instance, min_nodes - min number of workers expected to join the rendezvous, max_nodes - max number of workers allowed to join the rendezvous, defaults to min_workers is not specified. timeout - total timeout within which next_rendezvous is expected to succeed; a RendezvousTimeoutError is raised otherwise; Defaults is 600 (10 minutes). last_call_timeout - additional wait amount ("last call") after min number of workers has been reached. Defaults to 30 seconds. etcd_prefix - path prefix (from etcd root), inside which all etcd nodes will be created. Default is "/torchelastic/p2p". protocol - http (default) or https to access etcd. cacert - CA cert to access etcd, only makes sense with https. cert - client cert to access etcd, only makes sense with https. key - client key to access etcd, only makes sense with https. etcd_prefixz/torchelastic/p2pr|r})rhrxryrzr{r|r})r$r%) rrrry min_nodes max_nodes get_as_intrr rr%)rrhrrdzvs rrrsb! (F**],?@K }}((((!!)-=> ++ !;  D !$$ r)4rloggingsysrrtypingrr1$torch.distributed.elastic.rendezvousrrrrrr r etcd_storer r utilsr__all__ Formatter_log_fmt StreamHandlerstderr _log_handler setFormatter getLoggerrr5 propagatesetLevelINFO addHandlerrrrrr__annotations__r rrrrrtrrrrrrrrr0sQ    -,  7  D E$w$$SZZ0 (#   8 $ , Y  Y ##%C$$&! y-yDr Pr Pj 4FC 4C9JCr