o wO iXÁã@slddlZddlZddlZddlZddlZddlZddlZddlmZm Z ddl m Z ddl Z ddl mZddlmZmZmZmZmZddlmZe d¡Ze ej¡Ze e¡e e¡Zd e_ e !ej"¡e #e¡Gd d „d e$ƒZ%Gd d „d e$ƒZ&dZ'dZ(dZ)dZ*dZ+dd„Z,Gdd„deƒZ-Gdd„de.ƒZ/Gdd„deƒZ0dede j1fdd„Z2dedefdd„Z3dS)éN)Ú b64decodeÚ b64encode)ÚOptional)ÚStore)ÚRendezvousClosedExceptionÚRendezvousHandlerÚRendezvousNonRetryableErrorÚRendezvousParametersÚRendezvousTimeoutExceptioné)Ú_parse_hostname_and_portz%%(levelname)s %(asctime)s %(message)sFc@ó eZdZdS)ÚEtcdRendezvousRetryableFailureN©Ú__name__Ú __module__Ú __qualname__©rrú[/home/ubuntu/.local/lib/python3.10/site-packages/torchelastic/rendezvous/etcd_rendezvous.pyr/órc@r )ÚEtcdRendezvousRetryImmediatelyNrrrrrr5rréé i cCst t dd¡¡dS)Nrçš™™™™™¹?)ÚtimeÚsleepÚrandomÚuniformrrrrÚ cas_delayKsrc@sjeZdZdZdd„Zdd„Zdefdd„Zd d „Zd d „Z d d„Z dd„Z defdd„Z de fdd„ZdS)ÚEtcdRendezvousHandleruå Implements a :py:class:`torchelastic.rendezvous.RendezvousHandler` interface backed by :py:class:`torchelastic.rendezvous.etcd_rendezvous.EtcdRendezvous`. Torchelastic 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 - torchelastic 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``) | +--------------------------------------------+--------------------------+ cCs ||_dS©N©Ú _rdzv_impl)ÚselfÚ rdzv_implrrrÚ__init__s zEtcdRendezvousHandler.__init__cCs|`dSr r!©r#rrrÚ__del__’szEtcdRendezvousHandler.__del__ÚreturncCsdS)NÚetcdrr&rrrÚ get_backend–sz!EtcdRendezvousHandler.get_backendcCs0|j ¡\}}}t d¡|j |¡}|||fS)Nz4Creating EtcdStore as the c10d::Store implementation)r"Úrendezvous_barrierÚlogÚinfoÚsetup_kv_store)r#Ú rdzv_versionÚrankÚ world_sizeÚstorerrrÚnext_rendezvous™s   z%EtcdRendezvousHandler.next_rendezvouscCs4z|j ¡\}}|ddkWStjyYdSw)NÚstatusÚclosedF©r"Úget_rdzv_stater)ÚEtcdKeyNotFound©r#Ú_ÚstaterrrÚ is_closed¡s þzEtcdRendezvousHandler.is_closedcCs|j ¡dSr )r"Ú set_closedr&rrrr=©sz EtcdRendezvousHandler.set_closedcCsBz|j ¡\}}|ddkr|dWSWdStjy YdSw)Nr4ÚfinalÚnum_workers_waitingrr6r9rrrÚnum_nodes_waiting¬s  ÿþþz'EtcdRendezvousHandler.num_nodes_waitingcCs|jjSr )r"Ú_run_idr&rrrÚ get_run_idµsz EtcdRendezvousHandler.get_run_idc CsJz| ¡WdSty$}zt dt|ƒ›¡WYd}~dSd}~ww)NTz!Shutdown failed. Error occurred: F)r=Ú BaseExceptionr,ÚwarningÚstr©r#ÚerrrÚshutdown¸s€þzEtcdRendezvousHandler.shutdownN)rrrÚ__doc__r%r'rEr*r3r<r=r@rBÚboolrHrrrrrOs? rc@sÖeZdZdZdd„Zdd„Zdd„Zdd „Zd d „Zd d „Z dd„Z dd„Z dd„Z dd„Z dd„Zdd„Zdd„Zdd„Zdd„Zd d!„Zd"d#„Zd3d%d&„Zd'd(„Zd3d)d*„Zd+d,„Zd-d.„Zd3d/d0„Zd1d2„Zd$S)4ÚEtcdRendezvousze A rendezvous implementation that uses `etcd `__ as the backend store. cCsì||_t dt|jjƒ¡||_||_||_||_||_ ||_ d|_ d|_ |j  d¡s3|jd7_|jdkr>| |j¡|j| d¡td|j| d¡td|_ | | d¡¡z|jj| d¡ddd WdStjyuYdSw) NzEtcd machines: ú/Ú©Úttlú/rdzvú/rdzv/version_counterÚ0F©ÚkeyÚvalueÚ prevExist)Úclientr,r-rEÚmachinesÚ_prefixrAÚ_num_min_workersÚ_num_max_workersÚ_timeoutÚ_last_call_timeoutÚ_lease_run_id_stopÚ_lease_this_rank_stopÚendswithÚcreate_path_if_not_existsÚget_pathÚCONST_RUNID_SUBROOT_TTLÚsetup_lease_renewalÚwriter)ÚEtcdAlreadyExist)r#rWÚprefixÚrun_idÚnum_min_workersÚnum_max_workersÚtimeoutÚlast_call_timeoutrrrr%Ís4     ÿ  ÿÿzEtcdRendezvous.__init__cCs0|jdur |j ¡|jdur|j ¡dSdSr )r^Úsetr_r&rrrr'ýs   ÿzEtcdRendezvous.__del__c Cst ¡|j|_ t ¡|jkrtƒ‚t d¡z|jdur#|j ¡| ¡WSt y0YnQt y=t  d¡YnDtyIt d¡‚t yZt d|j ›d¡‚tya‚ty€}zt dt|ƒ¡t  d¡WYd}~nd}~wwq ) 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: RendezvousTimeoutException - timeout waiting for rendezvous RendezvousNonRetryableError - other persistent errors that render the rendezvous non-retryable RendezvousClosedException - rendezvous is or was closed while waiting Tz"Attempting to join next rendezvousNr z3Rendezvous timeout occured in EtcdRendezvousHandlerzRendezvous for run_id=z was observed to be closedz/Rendezvous attempt failed, will retry. Reason: )rr\Ú_rendezvous_deadliner r,r-r_rmÚ init_phaserrrrrArÚ ExceptionrErFrrrr+s:          ÿ €úÞz!EtcdRendezvous.rendezvous_barriercCsÂz| ¡}t |j¡}t dt|ƒ¡Wntjy.|  ¡\}}t dt|ƒ¡Ynw|ddkr8t ƒ‚|ddkrE|  |d¡S|ddkrU|  |d¡t ƒ‚|j|jdd tƒ‚) 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: RendezvousClosedException - current rendezvous was/is closed EtcdRendezvousRetryableFailure - observed some intermediate state, which is best handled by retrying later zNew rendezvous state created: z$Observed existing rendezvous state: r4r5ÚjoinableÚversionr>r ©Ú etcd_index)Útry_create_rendezvousÚjsonÚloadsrUr,r-rEr)rfr7rÚ join_phaseÚhandle_existing_rendezvousrÚtry_wait_for_state_changertr©r#Úactive_versionr;rrrro?s"  ü   zEtcdRendezvous.init_phasecCsÆ| |¡\}}t |j¡}t d |d||¡¡||jdkrC|ddkrCt d |¡¡t ¡|j }|  ||¡t d |¡¡t d¡|  |¡}t |j¡}|d|ks]Jd ƒ‚|  ||¡S) z¥ We observed a rendezvous state in 'joinable' state, and attempt to join this particular version, and then wait for all other peers to join. z7Joined rendezvous version {} as rank {}. Full state: {}rrr r4rqz*Rank {} is responsible for join last call.z Rank {} finished join last call.zWaiting for remaining peers.z/Logic error: failed to observe version mismatch) Újoin_rendezvousrvrwrUr,r-ÚformatrZrr]Úhandle_join_last_callÚwait_for_peersÚ confirm_phase)r#Úexpected_versionr|Ú this_rankr;Úlast_call_deadlinerrrrxis&  ÿÿ    ÿ zEtcdRendezvous.join_phasecCsbt d¡| ||¡t d¡| |¡}t |j¡}t d |d|¡¡|d|t|dƒfS)a7 Once the rendezvous state trainsitions 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 {} is complete. Final state: {}rrÚ participants) r,r-Úconfirm_membershipÚwait_for_finalrvrwrUr~Úlen)r#r‚rƒr|r;rrrr’s     ÿÿzEtcdRendezvous.confirm_phasecCs4| |¡}t d |j¡¡| |¡t d¡dS)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: {}zBPreviously existing rendezvous state changed. Will re-try joining.N)Úannounce_self_waitingr,r-r~rUÚwait_for_rendezvous_to_free)r#r‚Ú active_staterrrryªs ÿÿ z)EtcdRendezvous.handle_existing_rendezvousc CsÆ|jj| d¡t ddi¡dtd}z|j | d¡¡}tt|j ƒdƒ|_ |j  |¡Wnt j t j fy<tdƒ‚w|jj| d  |j ¡¡d d dd |jj| d¡t d |j gdœ¡|j dS)zÔ Create new rendezvous state or raise an exception that indicates an unexpected state (e.g. already exists) Raises: RendezvousNonRetryableError - on unexpected state ú/rdzv/active_versionr4ÚsetupF)rTrUrVrOrQr z?Unexpected state of EtcdRendezvousHandler, worker needs to die.ú /rdzv/v_{}NT)rTrUÚdirrVrq)r4rrr…©rTrUÚ prev_value)rWrerbrvÚdumpsÚCONST_ETCD_SETUP_TTLÚgetrEÚintrUÚupdater)r8ÚEtcdCompareFailedrr~Ú test_and_set)r#r|Úversion_counterrrrru¿s>  üÿÿ ü ýÿ÷z$EtcdRendezvous.try_create_rendezvouscCs tƒ| ¡\}}|ddkrtdƒ‚|d|krtdƒ‚t|dƒ|jks+Jdƒ‚t|dƒ}|d |¡t|dƒ|jkrLd |d<g|d <t}nt|dƒ|jkrXt }nd }z|j j |  d ¡t  |¡|j|d }||fWStjyt d¡Ynwq)ú3 Helper method for the join phase. Tr4rqzNRendezvous state became non-joinable before we could join. Must join next one.rrú6Rendezvous version changed. Must try join the new one.r…z>Logic error: joinable rendezvous should always have space leftÚfrozenÚ keep_alivesNrŒ©rTrUr‘rOz*Join rendezvous CAS unsuccessful, retrying)rr7rrrˆr[ÚappendÚCONST_ETCD_FROZEN_TTLrZÚ!CONST_ETCD_JOINABLE_EPHEMERAL_TTLrWr˜rbrvr’rUr)r—r,r-)r#r‚r|r;rƒÚset_ttlrrrr}ösF  ÿ ÿÿ ü ÿÔzEtcdRendezvous.join_rendezvouscCód| ¡\}} |ddkr|d|kr|S|ddkr-|d|kr-|j|jdd\}}ntdƒ‚q) ršTr4rœrrrqr rsú>Rendezvous state transition no longer possible. Must re-enter.©r7rzrtr©r#r‚r|r;rrrr€+ó  ÿÿózEtcdRendezvous.wait_for_peerscCs tƒ| ¡\}}|ddkrtdƒ‚|d|krtdƒ‚| d ||¡¡}|jj|dtd |d  |¡t |d ƒt |d ƒkrNd |d<d |d<d}nd}z |jj | d¡t   |¡|j |rbdntd}|j|td|_|WStjyt d¡Ynwq)ú5 Helper method for the confirm phase Tr4rœzDRendezvous no longer frozen, before we confirmed. Must join next onerrr›z/rdzv/v_{}/rank_{}N)rUrOrr…r>rr?FrŒržrNz-Confirm membership CAS unsuccessful, retrying)rr7rrbr~rWrmÚCONST_WORKER_KEEPALIVE_TTLrŸrˆr˜rvr’rUr rdr_r)r—r,r-)r#r‚rƒr|r;Úthis_lease_keyÚfinalizerrrr†AsH  ÿ ÿ ÿ üÿÿÖz!EtcdRendezvous.confirm_membershipcCr£) r¨Tr4r>rrrœr rsr¤r¥r¦rrrr‡tr§zEtcdRendezvous.wait_for_finalcCsŠ tƒ| ¡\}}|ddks|d|krtƒ‚|dd7<z|jj| d¡t |¡|jd}|WSt j yCt   d ¡Ynwq) z™ Announce this worker is waiting (via num_workers_waiting counter) to join next rendezvous, but only if state and version match. Tr4r>rrr?r rŒrz3Announce self as waiting CAS unsuccessful, retrying) rr7rrWr˜rbrvr’rUr)r—r,r-r¦rrrr‰Šs" ýÿîz$EtcdRendezvous.announce_self_waitingc Cs<| ¡\}} |ddks|d|krdS|j | dj|d¡¡}dd „|jDƒ}|d D]2}||vr`t d  |¡¡t d  |¡¡t d ¡|jj| d¡|j dt d |¡¡dSq.zt |j t   ¡dƒd}|jj | d¡|jdd|dWn tjtjfyŒYnwt   ¡|j kr—tƒ‚| ¡\}}q)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. Tr4r>rrNz/rdzv/v_{version})rrcSsg|]}|j‘qSr©rT)Ú.0ÚchrrrÚ »sz>EtcdRendezvous.wait_for_rendezvous_to_free..rz!Keep-alive key {} is not renewed.z$Rendevous version {} is incomplete. zAttempting to destroy it.rŒ)rTÚ prevValuez-Destroyed rendezvous version {} successfully.ççð?rPr )rTÚindexÚ recursiverk)r7rWr”rbr~Úchildrenr,r-ÚdeleterUÚmaxrnrÚwatchrtr)ÚEtcdEventIndexClearedÚEtcdWatchTimedOutr )r#r‚r|r;Ú alive_membersÚkeep_alive_keysrTÚoverall_timeoutrrrrŠ¥sT ÿ ÿ þÿÿèÿ üÿ Èz*EtcdRendezvous.wait_for_rendezvous_to_freecCsH| ¡\}} |ddkr|d|krdS|ddks!|d|kr%tdƒ‚t ¡|kr`d|d<g|d<z|jj| d ¡t |¡|jt d }WdSt j y_t   d ¡tƒ| ¡\}}Yqwz*|jj| d ¡|j|jtd }ttd |t ¡d ƒ}|j|jd|d\}}Wnt j y¢t   d¡tƒ| ¡\}}Ynwq)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 tranisiton to Exit with exception otherwise. Tr4rœrrNrqr¤rrŒržz6Join last-call transition CAS unsuccessful. Will retryér²r )rtrkz7Join last-call TTL refresh CAS unsuccessful, will retry)r7rrrWr˜rbrvr’rUr r)r—r,r-rr¡Úminrzrt)r#r‚Údeadliner|r;rkrrrrësZ ÿ ü  ü üþ ÿ ýÎz$EtcdRendezvous.handle_join_last_callcCst | ¡\}}|ddkrdSd|d<z|jj| d¡t |¡|jdWdStjy8t   d¡t ƒYnwq)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. Tr4r5NrŒrz%Set closed CAS unsuccessful, retrying) r7rWr˜rbrvr’rUr)r—r,r-rr{rrrr=4s"  ý  þðzEtcdRendezvous.set_closedcCs$|jj| d¡d}|t |j¡fS)NrŒr¬)rWr”rbrvrwrU)r#r|rrrr7NszEtcdRendezvous.get_rdzv_stateNc Cs‚t|jt ¡dƒd}|dur|nt||ƒ}z|jj| d¡||dWn tjtj fy2Ynwt ¡|jkr=t ƒ‚|  ¡S)Nr±r²rŒ)r³rk) r·rnrr¿rWr¸rbr)r¹rºr r7)r#rtrkr½rrrrzRs  ÿÿz(EtcdRendezvous.try_wait_for_state_changecCs&| d¡s d|}dj|j|j|dS)NrLz{prefix}run_{run_id}{path})rgrhÚpath)Ú startswithr~rYrA)r#rÁrrrrbds  ÿzEtcdRendezvous.get_pathcCs4z|jj|ddd|dWdStjyYdSw)NTF)rTrUrrVrO)rWrer)rf)r#Ú full_pathrOrrrrals  ÿÿz(EtcdRendezvous.create_path_if_not_existscCs:dd„}t ¡}tj||j|||fd}d|_| ¡|S)NcSsT z |j||dWntjyYdStyYdSw|j|ddr)dSq)NTrNr¾©rk)Úrefreshr)r8ÚConnectionRefusedErrorÚwait)rWrÁrOÚ stop_eventrrrÚ lease_workerzs ýõz8EtcdRendezvous.setup_lease_renewal..lease_worker)ÚtargetÚargsT)Ú threadingÚEventÚThreadrWÚdaemonÚstart)r#rÃrOrÉÚlease_stop_eventÚ lease_threadrrrrdtsÿz"EtcdRendezvous.setup_lease_renewalcCs¶| d |¡¡}z|jj|t ||i¡dd}WdStjy#Ynw |j |¡}t  |j ¡}|||<z|jj |t |¡|j d}WdStj yYt  d¡t d¡Ynwq%)Nú/rdzv/v_{}/extra_dataFrSTrz+Store extra_data CAS unsuccessful, retryingr)rbr~rWrervr’r)rfr”rwrUr˜r—r,r-rr)r#r/rTrUÚnodeÚ extra_dataÚnew_extra_data_valuerrrÚstore_extra_data’s2ÿÿ  ý þòzEtcdRendezvous.store_extra_datac s¶| d |¡¡‰| d |¡¡} |j |¡}‡fdd„|jDƒ}t|ƒdks)J‚t|ƒdkr?t |dj¡}||vr?||Sz |jj ˆ|j ddWn t j t j fyYYnwq) NrÓrŽTcsg|] }|jˆkr|‘qSrr¬)r­Ún©rÔrrr¯¼sz2EtcdRendezvous.load_extra_data..r r)r³)rbr~rWr”rµrˆrvrwrUr¸rtr)r¹rº)r#r/rTrkÚnode_dirÚrootrÕÚextra_data_dictrrÙrÚload_extra_data°s   ÿîzEtcdRendezvous.load_extra_datacCs*| d|›d¡}| |¡t|j|dS)Nz/rdzv/v_z/kv)Ú etcd_clientÚetcd_store_prefix)rbraÚ EtcdStorerW)r#r/Ú store_pathrrrr.Ìs zEtcdRendezvous.setup_kv_storer )rrrrIr%r'r+rorxrryrur}r€r†r‡r‰rŠrr=r7rzrbrardr×rÝr.rrrrrKÇs40:*)753FI    rKcs²eZdZdZ ddeejf‡fdd„ Zdd„Zde fd d „Z d e de fd d „Z ddeejfdd„Z defdd„Zdejfdd„Zdefdd„Zde fdd„Zddd„Z‡ZS)ràz— Implements a c10 Store interface by piggybacking on the rendezvous etcd instance. This is the store object returned by ``EtcdRendezvous`` NrkcsRtƒ ¡||_||_|dur|ntjdd|_|j d¡s'|jd7_dSdS)Ni,)ÚsecondsrL)Úsuperr%rWrgÚdatetimeÚ timedeltarkr`)r#rÞrßrk©Ú __class__rrr%Ùs ÿ ÿzEtcdStore.__init__cCs&|jj|j| |¡| |¡ddS)z‚ Write a key/value pair into ``EtcdStore``. Both key and value may be either Python ``str`` or ``bytes``. ©rTrUN)rWrmrgÚ_encode)r#rTrUrrrrmës&z EtcdStore.setr(cCsB|j| |¡}| |g¡}|durtd|›dƒ‚| ||¡S)aV Get a value by key, possibly doing a blocking wait. If key is not immediately present, will do a blocking wait for at most ``timeout`` duration or until the key is published. Returns: value ``(bytes)`` Raises: LookupError - If key still not published after timeout NzKey z not found in EtcdStore)rgréÚ _try_wait_getÚ LookupErrorÚ_decode)r#rTÚb64_keyÚkvsrrrr”òs  z EtcdStore.getÚnumcCsÎ| |¡}z|jj|j|| t|ƒ¡dd}t| |j¡ƒWStj y)Ynw |jj |j|d}| tt| |j¡ƒ|ƒ¡}z|jj |j ||jd}t| |j¡ƒWStj yetƒYnwq+)a Atomically increment a value by an integer amount. The integer is represented as a string using base 10. If key is not present, a default value of ``0`` will be assumed. Returns: the new (incremented) value FrSTr¬r)rérWrergrEr•rìrUr)rfr”r˜rTr—r)r#rTrïrírÔÚ new_valuerrrÚadds.  ýÿ ÿ ÿöz EtcdStore.addÚoverride_timeoutcs2‡fdd„|Dƒ}ˆ ||¡}|durtdƒ‚dS)z‹ Waits until all of the keys are published, or until timeout. Raises: LookupError - if timeout occurs cóg|] }ˆjˆ |¡‘qSr©rgré©r­rTr&rrr¯4óz"EtcdStore.wait..Nz+Timeout while waiting for keys in EtcdStore)rêrë)r#ÚkeysròÚb64_keysrîrr&rrÇ-s  ÿzEtcdStore.waitcs0‡fdd„|Dƒ}ˆj|tjddd}|duS)zU Check if all of the keys are immediately present (without waiting). crórrôrõr&rrr¯>röz#EtcdStore.check..r )Ú microseconds)ròN)rêrärå)r#r÷rørîrr&rÚcheck:s  þzEtcdStore.checkcCs ||_dS)zD Change the timeout used for all future operations. NrÄ)r#rkrrrÚ set_timeoutEs zEtcdStore.set_timeoutcCs<t|ƒtkr t|ƒ ¡St|ƒtkrt| ¡ƒ ¡Stdƒ‚©Nz"Value must be of type str or bytes)ÚtypeÚbytesrÚdecoderEÚencodeÚ ValueError©r#rUrrrréPs   zEtcdStore._encodecCs4t|ƒtkr t|ƒSt|ƒtkrt| ¡ƒStdƒ‚rü)rýrþrrErrrrrrrì[s   zEtcdStore._decodecsÐ|dur|jn|}t ¡| ¡} |jj|jd}‡fdd„|jDƒ}t|ƒtˆƒkr.|S|t ¡}|dkr:dSz|jj|jd||j ddWnt j y]t ¡|kr[YdSYqt j yfYqwq)NTr¬cs i|] }|jˆvr|j|j“qSrrè)r­rÔ©rørrÚ qs  þz+EtcdStore._try_wait_get..rr )rTr´rkr³) rkrÚ total_secondsrWr”rgrµrˆr¸rtr)rºr¹)r#røròrkrÀÚ all_nodesÚ req_nodesÚ watch_timeoutrrrrêjs6 þ  ü ÿãzEtcdStore._try_wait_getr )rrrrIrrärår%rmrþr”r•rñrÇrJrúrûrErérìrêÚ __classcell__rrrærràÓs üü%    ràÚparamsr(cCs”t|jdƒ\}}|j d¡}|durd}n |dkr!|dkr!tdƒ‚|j d¡}|dur9|j d¡}|dur9||f}|j d ¡}tj|||||d d S) zT Creates a new ``etcd.Client`` from the specified ``RendezvousParameters``. iK ÚprotocolNÚhttpÚhttpsz(The etcd protocol must be HTTP or HTTPS.ÚcertrTÚcacertT)r rÚca_certÚallow_reconnect)r ÚendpointÚconfigr”rr)ÚClient)r ÚhostnameÚportr Ússl_certÚcert_keyrrrrÚ_create_etcd_clients(    úrc Cs>t|ƒ}| dd¡}t|||j|j|j|j|jd}t|dS)a6 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 RendezvousTimeoutException 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/p2p)rWrgrhrirjrkrl)r$) rr”rKrhÚ min_nodesÚ max_nodesrkrlr)r rWrÚrdzvrrrÚcreate_rdzv_handler´s1 ù r)4rärvÚloggingrÚsysrÌrÚbase64rrÚtypingrr)Útorch.distributedrÚtorchelastic.rendezvousrrrr r Úutilsr Ú FormatterÚ_log_fmtÚ StreamHandlerÚstderrÚ _log_handlerÚ setFormatterÚ getLoggerrr,Ú propagateÚsetLevelÚINFOÚ addHandlerrprrr“r r¡r©rcrrÚobjectrKràrrrrrrrÚsP         x=%