U
|
©r_û ã @ sÔ d Z ddlZddlZddlZddlZddlZddlmZ dZ
|
e dddZ e dddZ e ddd Ze dd
|
d Zedd d Zdd Ze ¡ dd Zdd Zdd Zd ddZd ddZdd Zd!d d ZdS )"zl
|
This file contains primitives for multi-gpu communication.
|
This is useful when doing distributed training.
|
é N)Úreturnc C s t ¡ s dS t ¡ sdS t ¡ S )Né )ÚdistÚ is_availableÚis_initializedÚget_world_size© r r ú./utils/comm.pyr s
|
r c C s t ¡ s dS t ¡ sdS t ¡ S ©Nr )r r r Úget_rankr r r r r s
|
r c C s0 t ¡ s dS t ¡ sdS tdk s$tt jtdS )zh
|
Returns:
|
The rank of the current process within the local (per-machine) process group.
|
r N©Úgroup)r r r Ú_LOCAL_PROCESS_GROUPÚAssertionErrorr r r r r Úget_local_rank$ s r c C s$ t ¡ s dS t ¡ sdS t jtdS )zw
|
Returns:
|
The size of the per-machine process group,
|
i.e. the number of processes per machine.
|
r r )r r r r r r r r r Úget_local_size1 s
|
r c C s
|
t dkS r
|
)r r r r r Úis_main_process> s r c C s8 t ¡ s dS t ¡ sdS t ¡ } | dkr,dS t ¡ dS )zj
|
Helper function to synchronize (barrier) among all processes when
|
using distributed training
|
Nr )r r r r Úbarrier)Ú
|
world_sizer r r Ú synchronizeB s r c C s$ t ¡ dkrt jddS t jjS dS )zj
|
Return a process group based on gloo backend, containing all the ranks
|
The result is cached.
|
ÚncclÚgloo)ÚbackendN)r Ú get_backendÚ new_groupr ÚWORLDr r r r Ú_get_global_gloo_groupQ s r c C s t |¡}|dkstt |dkr&dnd¡}t | ¡}t|dkrjt t
|
¡}| d t t|d |¡¡ tj |¡}t |¡j|d}|S )N)r r r ÚcpuÚcudai @z;Rank {} trying to all-gather {:.2f} GB of data on device {})Údevice)r r r Útorchr ÚpickleÚdumpsÚlenÚloggingÚ getLoggerÚ__name__ÚwarningÚformatr Ú ByteStorageÚ from_bufferÚ
|
ByteTensorÚto)Údatar r r ÚbufferÚloggerÚstorageÚtensorr r r Ú_serialize_to_tensor] s
|
|
|
|
ÿÿ r2 c s® t j|d}|dks tdtj ¡ gtj jd} fddt|D }t j |||d dd |D }t
|
|}||kr¦tj || ftj jd}tj |fdd | fS )
|
zz
|
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!©Údtyper c s" g | ]}t jd gt j jdqS )r r3 )r ÚzerosÚint64r ©Ú.0Ú_©r1 r r Ú
|
<listcomp>z s z*_pad_to_largest_tensor.<locals>.<listcomp>c S s g | ]}t | ¡ qS r )ÚintÚitem)r8 Úsizer r r r; ~ s r ©Údim)r r r r r1 Únumelr6 r ÚrangeÚ
|
all_gatherÚmaxr5 Úuint8Úcat)r1 r r Z
|
local_sizeÚ size_listÚmax_sizeÚpaddingr r: r Ú_pad_to_largest_tensoro s ÿþ
|
ÿrJ c sº t dkr| gS |dkr t }t |¡dkr2| gS t| |t|\}t| fdd|D }tj||d g }t||D ]0\} ¡ ¡
|
¡ d|
}| t |¡¡ q|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 Nc s" g | ]}t j ft jjd qS ©r3 ©r ÚemptyrE r r7 ©rH r1 r r r; ¡ s z all_gather.<locals>.<listcomp>r )r r r r2 rJ rD rC Úzipr ÚnumpyÚtobytesÚappendr! Úloads)r- r rG Ú tensor_listÚ data_listr> r. r rN r rC s$
|
|
|
ÿrC c sì t dkr| gS |dkr t }tj |ddkr4| gS tj|d}t| |t|\}||krÒt| fdd|D }tj|||d g }t||D ]0\} ¡
|
¡ ¡ d|
}| t |¡¡ q|S tjg ||d g S dS )a
|
Run gather on arbitrary picklable data (not necessarily tensors).
|
Args:
|
data: any picklable object
|
dst (int): destination rank
|
group: a torch process group. By default, will use a group which
|
contains all ranks on gloo backend.
|
Returns:
|
list[data]: on dst, a list of data gathered from each rank. Otherwise,
|
an empty list.
|
r Nr c s" g | ]}t j ft jjd qS rK rL r7 rN r r r; È s zgather.<locals>.<listcomp>)Údstr )r r r r r2 rJ rD ÚgatherrO r rP rQ rR r! rS ) r- rV r ÚrankrG rT rU r> r. r rN r rW ® s,
|
|
ÿrW c C s t j d¡} t| }|d S )zü
|
Returns:
|
int: a random number that is the same across all workers.
|
If workers need a shared RNG, they can use this shared seed to
|
create one.
|
All workers must call this function, otherwise it will deadlock.
|
l r )ÚnpÚrandomÚrandintrC )ÚintsZall_intsr r r Úshared_random_seed× s r] Tc C s¤ t }|dk r| S t ¡ g }g }t| ¡ D ] }| |¡ | | | ¡ q0tj|dd}tj|dd t ¡ dkr|r|| }dd t
|
||D }W 5 Q R X |S )ac
|
Reduce the values in the dictionary from all processes so that process with rank
|
0 has the reduced results.
|
Args:
|
input_dict (dict): inputs to be reduced. All the values must be scalar CUDA Tensor.
|
average (bool): whether to do average or sum
|
Returns:
|
a dict with the same keys as input_dict, after reduction.
|
é r r? )rV c S s i | ]\}}||qS r r )r8 ÚkÚvr r r Ú
|
<dictcomp>þ s z reduce_dict.<locals>.<dictcomp>) r r Úno_gradÚsortedÚkeysrR Ústackr Úreducer rO )Z
|
input_dictÚaverager ÚnamesÚvaluesr_ Z reduced_dictr r r Ú reduce_dictä s
|
|
|
rj )N)r N)T)Ú__doc__Ú functoolsr$ rP rY r! r Ztorch.distributedÚ distributedr r r< r r r r Úboolr r Ú lru_cacher r2 rJ rC rW r] rj r r r r Ú<module> s*
|
|
$
|
)
|
