U yhO@sddlmZddlmZddlmZdZdZdddd gZee d d dZ e e e d d dZ dee e e e eedddZe e e dddZde e e dddd ZdS)) timedelta)List)contextmanagerz /num_membersz /last_member store_timeoutget_all synchronizebarrier)timeoutccs*|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_timeoutr W/var/www/html/venv/lib/python3.8/site-packages/torch/distributed/elastic/utils/store.pyrs )rankprefix world_sizecsF|fddt|D}t||dd}|dkrB|||S)aa 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|]}|qSr r ).0idxrr r 8szget_all..z /finishedr r key_prefixr)Z multi_getrange_barrier_nonblockingget)r rrrZdata_arrZ barrier_keyr rr r$s ,)datarrrr returnc CsHt||4||||t||||}|W5QRSQRXdS)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)r rrrrr Z agent_datar r r rIs )rrrcCs4|t}|t}||d}||kr0||d|S)zq Does all the non-blocking operations for a barrier and returns the final key that can be waited on. z ) _NUM_MEMBERS_LAST_MEMBER_CHECKINaddr)r rrZnum_members_keylast_member_keyrr r r r`s   rN)rrbarrier_timeoutrc Cs2t||t|||d}||W5QRXdS)a 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. Note: Since the data is not removed from the store, the barrier can be used once per unique ``key_prefix``. rN)rrr)r rrr#r"r r r rps )r)r)datetimertypingr contextlibrrr __all__floatrintstrrbytesrrrr r r r  s2    +