o Zhm@sBddlmZddlmZddlmZddlmZmZddl Z e j j Z dZ dZdZd Zd Zgd Zed efd dZdededefddZ d%dedededed edeef ddZdedededeegefdedeeef ddZdededefddZ    d&deded!edeed"eeegefdeddfd#d$ZdS)')Iterable)contextmanager) timedelta)CallableOptionalNz /num_membersz /last_memberz/TRACEz /TRACING_GATE) store_timeoutget_all synchronizebarriertimeoutccs,|j}|t|ddV||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 Z set_timeoutr)storer Z old_timeoutrT/var/www/auris/lib/python3.10/site-packages/torch/distributed/elastic/utils/store.pyrs  rrankprefix world_sizecsH|fddt|D}t||dd}|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 csg|]}|qSrr).0idxrrr Bszget_all..z /finishedrr key_prefixr)Z multi_getrange_barrier_nonblockingwait)rrrrZdata_arrZ barrier_keyrrrr /s r ,datarreturncCsTt||||||t||||}|WdS1s#wYdS)aT Synchronizes ``world_size`` agents between each other using the underlying c10d store. The ``data`` will be available on each of the agents. Note: The data on the path is not deleted, as a result there can be stale data if you use the same key_prefix twice. Time complexity: O(N) per worker, O(N^2) globally. N)rsetr )rrrrrr Z agent_datarrrr Rs $r rank_decoder trace_timeoutc sh|tdfdd}fdd}|dkr1|}td|S|S)N c st}d}tdD]D}|tkr|Sz'|dkr)|tgtdn|tgtddWq tyN|d7}||Yq w|S)Nrr )Z milliseconds)r!r_MAX_TRACE_MISSING_RANKSr_TRACErDistStoreErroradd)missing_rank_infoZ ranks_missingirr"rr#rrr_find_missing_ranksss$   z9_try_detecting_missing_ranks.._find_missing_rankscs>ztgdddgWStyYdSw)Nz[])r _TRACING_GATEr(r)rr"rrr_checkins  z._try_detecting_missing_ranks.._checkinr)r!r'r.) rrrrr"r#r-r/r*rr,r_try_detecting_missing_ranksisr0cCs4|t}|t}||d}||kr||d|S)zq Does all the non-blocking operations for a barrier and returns the final key that can be waited on. r%r$) _NUM_MEMBERS_LAST_MEMBER_CHECKINr)r!)rrrZnum_members_keylast_member_keyrrrrrs   r barrier_timeoutrank_tracing_decoderc Cs|dur |dus Jdt||Qt|||d}z||gWn8tyY}z,|dur/|t|||||p9dd|} | durStd|||dd| d |d|d}~wwWddS1sewYdS) as 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 meaninful 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 informationrcSst|S)N)str)xrrrszbarrier..ziTimed out waiting on barrier on rank {}, for key prefix: {} (world_size={}, missing_ranks={}, timeout={})[z, ])rrrr(r0formatjoin) rrrr5rr6r#r3eZ missing_ranksrrrr sJ    "r )r)rNNr4)collections.abcr contextlibrdatetimertypingrrZtorchZ_CZ_DistStoreErrorr(r1r2r'r.r&__all__floatrintr7r byteslistr r0rr rrrrs|  )    /