o *j#"@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)torchZ ones_like.0_tensorr~/var/www/html/Deteccion_Ine/venv/lib/python3.10/site-packages/modelscope/models/multi_modal/videocomposer/utils/distributed.py s zall_gather..FZasync_oprdim)distget_world_sizerange all_gatherappendrcat)tensorsZ gather_listZ output_tensor world_sizeZtensor_placeholderZgathered_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_reducerZmul_)rZaveragerrrrr 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)rcudaZ set_devicerinit_process_group)Z local_rankZlocal_world_sizeZshard_idZ num_shardsrZ dist_backendZ proc_rankrrrr r8s   rcCstjr t|dkSdS)zB Determines if the current process is the master process. rT)r distributedis_initializedrget_rank)Znum_gpusrrr is_master_proc^s r!cC tsdSts dStS)z$ Get the size of the world. )r is_availablerrrrrr rh rcCr")z. Get the rank of the current process. r)rr$rr rrrr r sr%r cCs8tsdSts dSt}|dkrdStdS)zj Helper function to synchronize (barrier) among all processes when using distributed training Nr#)rr$rrZbarrier)rrrr synchronize~s r&cCs 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_groupgroupZWORLDrrrr _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. )r'rr'cpuri@z;Rank {} trying to all-gather {:.2f} GB of data on device {})device)rr(rr-pickledumpslenlogging getLogger__name__warningformatr Z ByteStorage from_bufferZ ByteTensorto)datar*rr-bufferloggerZstoragerrrr _serialize_to_tensors      r;cstj|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*r#zHcomm.gather/all_gather must be called from ranks within the given group!Zdtyper-cs"g|] }tjdgtjjdqS)r#r=)rzerosint64r-rrrr r z*_pad_to_largest_tensor..cSsg|]}t|qSr)intitem)rsizerrr r srr ) rrrrZnumelr?r-rrmaxr>uint8r)rr*rZ local_size size_listmax_sizepaddingrrr _pad_to_largest_tensors*    rIcstdkr|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 r#Ncs"g|] }tjftjjdqS)r=)remptyrEr-rrGrrr r r@z(all_gather_unaligned..r<)rr+rr;rIrDrzipr,numpytobytesrr.loads)r8r*rFZ tensor_listZ data_listrCr9rrKr all_gather_unaligneds$   rPcCsd|jdkrdS|j}t|}t|D]}tt|||d|}t|}||jkr/|aqdS)z? Initialize variables needed for distributed training. r#N)ZNUM_GPUSrrrlistr)ZSHARD_ID_LOCAL_PROCESS_GROUP)cfgZnum_gpus_per_machineZ num_machinesiZ ranks_on_iZpgrrr init_distributed_trainings     rUreturncCs$tsdSts dStjtdS)zw Returns: The size of the per-machine process group, i.e. the number of processes per machine. r#r<)rr$rrrRrrrr get_local_sizes  rWcCs0tsdSts dStdusJtjtdS)zh Returns: The rank of the current process within the local (per-machine) process group. rNr<)rr$rrRr rrrr get_local_ranks   rX)T)r)r)N)__doc__ functoolsr1r.rZtorch.distributedrrrRrrrr!rr r& lru_cacher+r;rIrPrUrArWrXrrrr s.   &      $'