`inddlmZddlmZddlmZddlmZmZddl Z e j j Z dZ dZdZd Zd Zgd Zed efd ZdededefdZ ddedededed edeef dZdedededeegefdedeeef dZdededefdZ d dedededeedeeegefdeddfdZdS)!)Iterable)contextmanager) timedelta)CallableOptionalNz /num_membersz /last_memberz/TRACEz /TRACING_GATE) store_timeoutget_all synchronizebarriertimeoutc#K|j}|t|dV||dS)z This sets the timeout and then restores the old timeout when the context manager exits. Args: store: the store to set the timeout on timeout: the timeout to set secondsN)r set_timeoutr)storer old_timeouts y/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/torch/distributed/elastic/utils/store.pyr r sQ-K i000111 EEE k"""""rankprefix world_sizec|fdt|D}t||d}|dkr||g|S)ad Given a store and a prefix, the method goes through the array of keys of the following format: ``{prefix}{idx}``, where idx is in a range from 0 to size, and tries to retrieve the data. The Rank0 process waits at the end to make sure all other processes finished the procedure before exiting. Usage :: values = get_all(store, "torchelastic/data", 3) value1 = values[0] # retrieves the data for key torchelastic/data0 value2 = values[1] # retrieves the data for key torchelastic/data1 value3 = values[2] # retrieves the data for key torchelastic/data2 cg|]}| Sr).0idxrs r zget_all..Bs$NNNS6 03 0 0NNNrz /finishedrr key_prefixr) multi_getrange_barrier_nonblockingwait)rrrrdata_arr barrier_keys ` rr r /s&NNNNE*ct}d}tdD]}|tkrn |dkr1|tgt n0|tgt dx#t $r&|dz }||YwxYw|S)Nrr) milliseconds)r+r"_MAX_TRACE_MISSING_RANKSr$_TRACErDistStoreErroradd)missing_rank_info ranks_missingir r-rr.rs r_find_missing_ranksz9_try_detecting_missing_ranks.._find_missing_ranksssEE q*%% 7 7A 888 7 A%%JJ&336334i 6V6V6V JJ: :q :& : :;YTU=V=V=VWWW! 7 7 7" !%%ll1oo66666 7! sA'B-C  C c tgdddgS#t$rYdSwxYw)Nz[])r$ _TRACING_GATEr6)r r-rsr_checkinz._try_detecting_missing_ranks.._checkinsf  JJ:6}667 8 8 8Q||AQQQR R   44 s.2 AAr)r+r5r=) rrr rr-r.r;r>r8s ``` `` r_try_detecting_missing_ranksr?is II+T+6++_===!!!!!!!!!* qyy//11 Z000/BBB  xzzrc|tz}|tz}||d}||kr||d|S)zq Does all the non-blocking operations for a barrier and returns the final key that can be waited on. r2r0) _NUM_MEMBERS_LAST_MEMBER_CHECKINr7r+)rrr num_members_keylast_member_keyrs rr#r#sQ !</O #77O ))OQ ' 'C j /?333 r barrier_timeoutrank_tracing_decoderc | | Jdt||5t|||} ||gno#t$rb}||t |||||pd|} | >td|||dd| d|d|d}~wwxYw ddddS#1swxYwYdS) at A global lock between agents. This will pause all workers until at least ``world_size`` workers respond. This uses a fast incrementing index to assign waiting ranks and a success flag set by the last worker. Time complexity: O(1) per worker, O(N) globally. Optionally, passing rank will enable tracing of missing ranks on timeouts. `rank_tracing_decoder` lambda arg can be used to convert rank data into a more meaningful information at an app level (e.g. hostname). Note: Since the data is not removed from the store, the barrier can be used once per unique ``key_prefix``. Nz!Tracing requires rank informationrc t|S)N)str)xs rzbarrier..s s1vvrziTimed out waiting on barrier on rank {}, for key prefix: {} (world_size={}, missing_ranks={}, timeout={})[z, ])r r#r$r6r?formatjoin) rrr rFrrGr.rDe missing_rankss rr r s4 |#++-P+++ uo . ..J:     JJ( ) ) ) )   | <(>-=-=! !! !,(ddjdj &&; - 8 8;;;+ ee     G1  * s5CA  C B6AB11B66CC C )r')r'NNrE)collections.abcr contextlibrdatetimertypingrrtorch_C_DistStoreErrorr6rArBr5r=r4__all__floatr intrJr byteslistr r?r#r rrrr_sU%$$$$$%%%%%%%%%%%%%% ) %   A @ @ #% # # # #  c s    R         %[.,,,  , C5#:& ,  ,hsm,,,,^ C S S    &!;?;;;; ; 3- ; #8SE3J#78 ;; ;;;;;;r