o }oi @sddlZddlZddlZddlZddlmZddlmZddl m Z z ddl m Z dZ Wn eefy9dZ Ynwddd Zdd d Zejd d ZddZdS)N)logging)is_global_rank_zero)parallel_stateTFncclc Cs|j}ttdd}ttdd}td|d|d||tj}|dur.|}tj |d }td d }td d }||d|7}tj j ||||d|||fS)zInitialize torch.distributed.RANK0 WORLD_SIZE1z0Initializing torch.distributed with local_rank: z, rank: z, world_size: Nztcp:// MASTER_ADDR localhost MASTER_PORT6000:)backend world_sizerank init_method) local_rankintosgetenvrinfotorchcuda device_count set_device distributedinit_process_group) argsrrrrdevicer master_ip master_portr"J/home/ubuntu/.local/lib/python3.10/site-packages/nemo/utils/distributed.pyinitialize_distributed!s     r$cCsxts|St}t}|dkr|Sddt|D}tj|||dur.||kr.dSg}|D]}||q2|S)a Collect objects (e.g., results) from all GPUs. Useful for inference over multiple GPUs with DDP. Use main_rank to specify which rank will be used to gather results. This allows to continue execution on the main_rank only after the gather. Args: partial_results_list: list of partial results from each GPU main_rank: rank of the main process to collect results from all GPUs (useful for collecting results in a target rank) Example: predictions = gather_objects(predictions,main_rank=0) # all but rank 0 will return None if predictions is None: return # from here only rank 0 should contiue pickle.dump(predictions, open(output_fname, "wb")) cSsg|]}dqSNr").0_r"r"r# ]sz"gather_objects..N) ris_initializedget_data_parallel_rankget_data_parallel_world_sizerangerrall_gather_objectextend)partial_results_list main_rankrrgathered_results results_listrr"r"r#gather_objects=s r5ccsRtr tg}ndg}t||djVttr'|ddSdS)aCreate a shared temporary directory across ranks in distributed setup. This function assumes that the distributed setup has been already correctly initialized. It is intended to be used only in single-node setup so that all ranks can access the directory created.Nr)rtempfileTemporaryDirectorydistbroadcast_object_listnamebarriercleanup)tmp_dirr"r"r#temporary_directoryms   r>ccsXtjj}d}|dur|j}|j}|dkr%t||d|EdHdS|EdHdS)z This is for latest webdataset>=0.2.6 This function will make sure that each worker gets a different subset of the dataset. r%N)rutilsdataget_worker_infoid num_workerslist)src worker_inforCworkerr"r"r#webdataset_split_by_workerss rH)rr&) contextlibrr6rtorch.distributedrr8 nemo.utilsrnemo.utils.get_rankr megatron.corerHAVE_MEGATRON_CORE ImportErrorModuleNotFoundErrorr$r5contextmanagerr>rHr"r"r"r#s$      0