Ng%dZddlZddlZddlZddlZddlZddlmZ dZ de fdZ de fdZ de fdZde fdZdefdZd Zejd Zd Zd Zdd ZddZdZddZddZddZdS)z PyTorch distributed helpers Some of this lifted from Detectron2 with other fns added by myself. Some of the Detectron2 fns were intended for use with GLOO PG. I am using NCCL here with default PG so not everything will work as is -RW Nreturnc|tjsdStjsdStjS)N)dist is_availableis_initializedget_world_sizeN/var/www/html/ai-engine/env/lib/python3.11/site-packages/effdet/distributed.pyr r s=    q   q    r c|tjsdStjsdStjSNr)rrrget_rankr r r rrs:    q   q =??r ctjsdStjsdStJtjtS)zh Returns: The rank of the current process within the local (per-machine) process group. rNgroup)rrr_LOCAL_PROCESS_GROUPrr r r get_local_rankr%sP    q   q  + + + =3 4 4 44r ctjsdStjsdStjtS)zw Returns: The size of the per-machine process group, i.e. the number of processes per machine. rr)rrrr rr r r get_local_sizer2sE    q   q  %9 : : ::r c&tdkSr)rr r r is_main_processr?s ::?r ctjsdStjsdStj}|dkrdStjdS)zj Helper function to synchronize (barrier) among all processes when using distributed training Nr)rrrr barrier) world_sizes r synchronizerCs[       $&&JQLNNNNNr c|tjdkrtjdStjjS)zj Return a process group based on gloo backend, containing all the ranks The result is cached. ncclgloo)backend)r get_backend new_grouprWORLDr r r _get_global_gloo_groupr$Rs6  V##~f----zr c tj|}|dvsJtj|dkrdnd}t j|}t |dkr_tjt}| d tt |dz |tj |}tj||}|S)N)rrrcpucudai@z;Rank {} trying to all-gather {:.2f} GB of data on device {})device)rr!torchr(pickledumpslenlogging getLogger__name__warningformatr ByteStorage from_buffer ByteTensorto)datarr r(bufferloggerstoragetensors r _serialize_to_tensorr;^su%%G & & & & & \7f#4#4%%& A AF \$  F 6{{Y"8,, I P P CKK95v     ++F33G  g & & ) ) ) 8 8F Mr ctj|}|dks Jdtjgtjj}fdt|D}tj|||d|D}t|}||krBtj ||z ftj j}tj |fd|fS) zz Returns: list[int]: size of the tensor, on each rank Tensor: padded tensor that has the max size rrzHcomm.gather/all_gather must be called from ranks within the given group!dtyper(c^g|])}tjdgtjj*S)rr=)r)zerosint64r(.0_r:s r z*_pad_to_largest_tensor..{s=FG QCu{6=AAAr cPg|]#}t|$Sr )intitem)rCsizes r rEz*_pad_to_largest_tensor..s(888dTYY[[!!888r rdim) rr r)r:numelrAr(range all_gathermaxr@uint8cat)r:rr local_size size_listmax_sizepaddings` r _pad_to_largest_tensorrVps $5111JaQ v||~~.ek&-XXXJKPQ[K\K\I OIz777788i888I9~~HX+x*46ekRXR_```FG,!444 f r c8tdkr|gS|t}tj|dkr|gSt||t |\}t |fd|D}tj||g}t||D]l\} d|}| tj |m|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 rNc^g|])}tjftjj*Sr=r)emptyrPr(rCrDrTr:s r rEzall_gather..s1hhhYZ5;{%+fmTTThhhr r)r r$rr;rVrOrNzipr&numpytobytesappendr*loads) r6rrS tensor_list data_listrIr7rTr:s @@r rNrNs'1v  }&(( 5!!Q&&v !$ . .F.vu==Iv9~~Hihhhh^ghhhKOKu5555II{33// f##%%--//6f--.... r c tdkr|gS|t}tj|dkr|gStj|}t || t |\} ||krt | fd|D}tj |||g}t||D]l\} d|}| tj|m|Stj g||gS)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. rNrc^g|])}tjftjj*SrYrZr\s r rEzgather..s1lll]^u{H;ek&-XXXlllr dstr)r r$rrr;rVrOgatherr]r&r^r_r`r*ra) r6rgrrankrSrbrcrIr7rTr:s @@r rhrhsd1v  }&(( '''1,,v =u % % %D !$ . .F.vu==Iv s{{y>>lllllbklll  FKS>>>>  ;77 3 3LD&ZZ\\''))1133ETE:F   V\&11 2 2 2 2 FBCu5555 r cntjd}t|}|dS)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. lr)nprandomrandintrN)intsall_intss r shared_random_seedrps0 9  W % %D$H A;r Tct}|dkr|Stj5g}g}t|D]2}|||||3tj|d}tj|dtj dkr|r||z}dt||D}dddn #1swxYwY|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. rrJ)rgci|]\}}|| Sr r )rCkvs r zreduce_dict..s<<<A1<<._do_gather..s$KKKAu'//KKKr rrJ)rMrrNr)rQ)r:rbcat_dimrrs` r _do_gatherz(all_gather_container.._do_gathersQKKKKz9J9JKKK   V59999y'2222r c&g|] }|Sr r rCrurs r rEz(all_gather_container..!555aJJqMM555r ) rrr#r isinstancedictitemslisttupler)Tensor) containerrrgatheredrtrurrs `` @@r all_gather_containerrs  %TZ%E$U++J3333333 )T""%66OO%%  DAq 1 AHQKK Ie} - -%55559555 i ' ' 'XH)U\22222z)$$$r c" ptjjtj tj fdt |t r:t }|D]\}}|}|||<|St |ttfr4fd|D}t |trt|}|St |tj sJ|S)NckrfdtD}nd}tj|tj|S)Nc8g|]}tjSr rrBs r rEz8gather_container.._do_gather.. s$OOO5+F33OOOr rfrJ)rMrrhr)rQ)r:rbrrgr this_rankrs` r rz$gather_container.._do_gatherse   OOOOU:=N=NOOOKKK FKS>>>>y'2222r c&g|] }|Sr r rs r rEz$gather_container..-rr ) rrr#r rrrrrrr)r) rrgrrrrtrurrrs ``` @@@r gather_containerrs5  %TZ%E$U++J e$$I333333333)T""%66OO%%  DAq 1 AHQKK Ie} - -%55559555 i ' ' 'XH)U\22222z)$$$r )N)rN)Tr)__doc__ functoolsr-r^rkr*r)torch.distributed distributedrrrGr rrrboolrr lru_cacher$r;rVrNrhrprrrr r r rs   !!!!!# 5 5 5 5 5 ; ; ; ; ;      $6D$$$$N   <%%%%4%%%%%%r