o ߥi#"@sdZddlZddlZddlZddlZddlmZdaddZ d#ddZ d$d d Z d%d d Z ddZ ddZddZeddZddZddZd&ddZddZdefdd Zdefd!d"ZdS)'zDistributed helpers.Ncslg}g}t}|D]fddt|D}tj|dd||q |D] }|tj|ddq'|S)z All gathers the provided tensors from all processes across machines. Args: tensors (list): tensors to perform all gather across all processes in all machines. csg|]}tqS)torch ones_like.0_tensorrq/home/ubuntu/.local/lib/python3.10/site-packages/modelscope/models/multi_modal/videocomposer/utils/distributed.py s zall_gather..Fasync_oprdim)distget_world_sizerange all_gatherappendrcat)tensors gather_list output_tensor world_sizetensor_placeholdergathered_tensorrrr rs  rTcCs@|D] }tj|ddq|rt}|D] }|d|q|S)a- All reduce the provided tensors from all processes across machines. Args: tensors (list): tensors to perform all reduce across all processes in all machines. average (bool): scales the reduced tensor by the number of overall processes across all machines. Fr g?)r all_reducermul_)raverager rrrr r%s rncclcCs6tj||||}||}tj||||ddS)a~ Initializes the default process group. Args: local_rank (int): the rank on the current local machine. local_world_size (int): the world size (number of processes running) on the current local machine. shard_id (int): the shard index (machine rank) of the current machine. num_shards (int): number of shards for distributed training. init_method (string): supporting three different methods for initializing process groups: "file": use shared file system to initialize the groups across different processes. "tcp": use tcp address to initialize the groups across different dist_backend (string): backend to use for distributed training. Options includes gloo, mpi and nccl, the details can be found here: https://pytorch.org/docs/stable/distributed.html )backend init_methodrrankN)rcuda set_devicerinit_process_group) local_ranklocal_world_sizeshard_id num_shardsr! dist_backend proc_rankrrrr r%8s   r%cCstjr t|dkSdS)zB Determines if the current process is the master process. rT)r distributedis_initializedrget_rank)num_gpusrrr is_master_proc^s r1cC tsdSts dStS)z$ Get the size of the world. )r is_availabler.rrrrr rh rcCr2)z. Get the rank of the current process. r)rr4r.r/rrrr r/sr5r/cCs8tsdSts dSt}|dkrdStdS)zj Helper function to synchronize (barrier) among all processes when using distributed training Nr3)rr4r.rbarrier)rrrr synchronize~s r7cCs tdkr tjddStjjS)z Return a process group based on gloo backend, containing all the ranks The result is cached. Returns: (group): pytorch dist group. rgloo)r )r get_backend new_groupgroupWORLDrrrr _get_global_gloo_groups  r=cCst|}|dvs Jt|dkrdnd}t|}t|dkr5tt }| d t t|d|tj |}t|j|d}|S)a Seriialize the tensor to ByteTensor. Note that only `gloo` and `nccl` backend is supported. Args: data (data): data to be serialized. group (group): pytorch dist group. Returns: tensor (ByteTensor): tensor that serialized. )r8rr8cpur#i@z;Rank {} trying to all-gather {:.2f} GB of data on device {})device)rr9rr?pickledumpslenlogging getLogger__name__warningformatr/ ByteStorage from_buffer ByteTensorto)datar;r r?bufferloggerstorager rrr _serialize_to_tensors      rPcstj|d}|dksJdtjgtjjd}fddt|D}tj|||ddd|D}t |}||krStj ||ftj jd}tj |fdd |fS) a Padding all the tensors from different GPUs to the largest ones. Args: tensor (tensor): tensor to pad. group (group): pytorch dist group. Returns: list[int]: size of the tensor, on each rank Tensor: padded tensor that has the max size r;r3zHcomm.gather/all_gather must be called from ranks within the given group!dtyper?cs"g|] }tjdgtjjdqS)r3rR)rzerosint64r?rrrr r z*_pad_to_largest_tensor..cSsg|]}t|qSr)intitem)rsizerrr r srr) rrrr numelrUr?rrmaxrTuint8r)r r;r local_size size_listmax_sizepaddingrrr _pad_to_largest_tensors*    racstdkr|gS|durt}t|dkr|gSt||t|\}t|fdd|D}tj||dg}t||D]\} d|}| t |qB|S)a= Run all_gather on arbitrary picklable data (not necessarily tensors). Args: data: any picklable object group: a torch process group. By default, will use a group which contains all ranks on gloo backend. Returns: list[data]: list of data gathered from each rank r3Ncs"g|] }tjftjjdqS)rR)remptyr\r?rr_r rr r rVz(all_gather_unaligned..rQ)rr=rrPrar[rzipr>numpytobytesrr@loads)rLr;r^ tensor_list data_listrYrMrrcr all_gather_unaligneds$   rjcCsd|jdkrdS|j}t|}t|D]}tt|||d|}t|}||jkr/|aqdS)z? Initialize variables needed for distributed training. r3N)NUM_GPUSrrrlistr:SHARD_ID_LOCAL_PROCESS_GROUP)cfgnum_gpus_per_machine num_machinesi ranks_on_ipgrrr init_distributed_trainings     rureturncCs$tsdSts dStjtdS)zw Returns: The size of the per-machine process group, i.e. the number of processes per machine. r3rQ)rr4r.rrnrrrr get_local_sizes  rwcCs0tsdSts dStdusJtjtdS)zh Returns: The rank of the current process within the local (per-machine) process group. rNrQ)rr4r.rnr/rrrr get_local_ranks   rx)T)r)r,)N)__doc__ functoolsrCr@rtorch.distributedr-rrnrrr%r1rr/r7 lru_cacher=rPrarjrurWrwrxrrrr s.   &      $'