B
|
©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 )Nr )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
|
|
|
r0 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 )r1 r )r ÚzerosÚint64r )Ú.0Ú_)r/ r r ú
|
<listcomp>{ s z*_pad_to_largest_tensor.<locals>.<listcomp>c S s g | ]}t | ¡ qS r )ÚintÚitem)r4 Úsizer r r r6 ~ s r )Údim)r r r r r/ Únumelr3 r ÚrangeÚ
|
all_gatherÚmaxr2 Úuint8Úcat)r/ r r Z
|
local_sizeÚ size_listÚmax_sizeÚpaddingr )r/ r Ú_pad_to_largest_tensoro s
|
rD c s¾ t dkr| gS |dkr t }t |¡dkr2| gS t| |t|\}t| fdd|D }tj||d g }x>t||D ]0\} ¡ ¡
|
¡ d|
}| t |¡¡ qW |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 ))r1 r )r Úemptyr? r )r4 r5 )rB r/ r r r6 ¢ s z all_gather.<locals>.<listcomp>)r )r r r r0 rD r> r= Úzipr ÚnumpyÚtobytesÚappendr Úloads)r+ r rA Ú tensor_listÚ data_listr9 r, r )rB r/ r r= s
|
|
|
r= c sð t dkr| gS |dkr t }tj |ddkr4| gS tj|d}t| |t|\}||krÖt| fdd|D }tj|||d g }x>t||D ]0\} ¡
|
¡ ¡ d|
}| t |¡¡ qW |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 N)r c s" g | ]}t j ft jjd qS ))r1 r )r rE r? r )r4 r5 )rB r/ r r r6 É s zgather.<locals>.<listcomp>)Údstr )r r r r
|
r0 rD r> ÚgatherrF r rG rH rI r rJ ) r+ rM r ÚrankrA rK rL r9 r, r )rB r/ r rN ® s(
|
|
rN 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Úrandintr= )ÚintsZall_intsr r r Úshared_random_seed× s rT Tc C s¨ t }|dk r| S t ¡ g }g }x,t| ¡ D ] }| |¡ | | | ¡ q2W tj|dd}tj|dd t ¡ dkr|r|| }dd t
|
||D }W dQ 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: )rM c S s i | ]\}}||qS r r )r4 ÚkÚvr r r ú
|
<dictcomp>þ s z reduce_dict.<locals>.<dictcomp>N) r r Úno_gradÚsortedÚkeysrI Ústackr Úreducer
|
rF )Z
|
input_dictÚaverager ÚnamesÚvaluesrV Z reduced_dictr r r Ú reduce_dictä s
|
|
|
ra )N)r N)T)Ú__doc__Ú functoolsr" rG rP r r Útorch.distributedÚ distributedr r r7 r r
|
r r Úboolr r Ú lru_cacher r0 rD r= rN rT ra r r r r Ú<module> s(
|
$
|
)
|
