kh%dZddlZddlZddlZddlZddlZddlmZ dZ de fdZ de fdZ de fdZde fdZdefdZd Zej&d Zd Zd Zdd ZddZdZddZddZddZy)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~tjsytjsytjS)N)dist is_availableis_initializedget_world_sizeN/var/www/teggl/fontify/venv/lib/python3.12/site-packages/effdet/distributed.pyr r s/             r c~tjsytjsytjSNr)rrrget_rankr r r rrs,         ==?r ctjsytjsytJtjtS)zh Returns: The rank of the current process within the local (per-machine) process group. rgroup)rrr_LOCAL_PROCESS_GROUPrr r r get_local_rankr%s>          ++ + ==3 44r ctjsytjsytjtS)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_sizer2s4           %9 ::r ctdk(Sr)rr r r is_main_processr?s :?r ctjsytjsytj}|dk(rytjy)zj Helper function to synchronize (barrier) among all processes when using distributed training Nr)rrrr barrier) world_sizes r synchronizerCsC        $$&JQLLNr ctjdk(rtjdStjjS)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$Rs4  V#~~f--zzr ctj|}|dvsJtj|dk(rdnd}t j |}t |dkDrOtjt}|jdjtt |dz |tjj|}tj|j!|}|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%& AF \\$ F 6{Y""8, I P P CK95v  ++F3G   g & ) ) ) 8F Mr ctj|}|dk\sJdtj|j gtj |j }t|Dcgc]3}tjdgtj |j 5}}tj||||Dcgc]}t|j}}t|}||k7rMtj||z ftj|j }tj||fd}||fScc}wcc}w)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(rdim)rr r)r:numelint64r(rangezeros all_gatherintitemmaxuint8cat) r:rr local_size_ size_listsizemax_sizepaddings r _pad_to_largest_tensorrQps $$51JaRQRv||~.ekk&--XJKPQ[K\FG QCu{{6==AI OOIz7.78dTYY[!8I89~HX++x*46ekkRXR_R_`FG,!4 f 9s /8E E cHtdk(r|gS| t}tj|dk(r|gSt||}t ||\}}t |}|Dcgc]3}t j|ft j|j5}}tj|||g}t||D]X\}}|jjjd|} |jt!j"| Z|Scc}w)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 rNr=r)r r$rr;rQrHr)emptyrIr(rEzipr&numpytobytesappendr*loads) r6rr:rMrOrL tensor_list data_listrNr7s r rErEs 1v  }&( 5!Q&v !$ .F.vu=Iv9~H_hhYZ5;;{%++fmmThKhOOKu5II{3/ f##%--/6f-./ is"8Dctdk(r|gS| t}tj|dk(r|gStj|}t ||}t ||\}}||k(rt |}|Dcgc]3}tj|ftj|j5}}tj||||g} t||D]X\} }|jjjd| } | j!t#j$| Z| Stj|g||gScc}w)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. rNrr=dstr)r r$rrr;rQrHr)rSrIr(gatherrTr&rUrVrWr*rX) r6r]rrankr:rMrOrLrYrZrNr7s r r^r^s61v  }&( '1,v ==u %D !$ .F.vu=Iv s{y>bkl]^u{{H;ekk&--Xl l FKS>  ;7 3LD&ZZ\'')113ET:F   V\\&1 2 3 FBCu5 ms>8Ec`tjjd}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)nprandomrandintrE)intsall_intss r shared_random_seedrfs, 99  W %D$H A;r ct}|dkr|Stj5g}g}t|j D]'}|j ||j ||)tj |d}tj|dtjdk(r|r||z}t||Dcic]\}}|| }}}ddd|Scc}}w#1swYSxYw)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. rr?)r]N) r r)no_gradsortedkeysrWstackrreducerrT) input_dictaveragernamesvalueskv reduced_dicts r reduce_dictrus !JA~  = )* )A LLO MM*Q- ( )V+ F" ==?a G j F),UF);<A1< < = = = sB#C) C#C)#C))C3cxstjjtjfd}t |t r1t }|j D]\}}||}|||<|St |ttfr2|Dcgc] }|| }}t |tr t|}|St |tjsJ||Scc}w)NctDcgc]}tj|}}tj||tj |Scc}w)Nrr?)rCr) empty_likerrErJ)r:rLrYcat_dimrrs r _do_gatherz(all_gather_container.._do_gathersM9>z9JKAu''/K K  V59yy'22LsA) rrr#r isinstancedictitemslisttupler)Tensor) containerrryrzgatheredrrrsrs `` @r all_gather_containerrs  %TZZ%%E$$U+J3 )T"6OO% DAq1 AHQK  Ie} -+45aJqM55 i 'XH)U\\222)$$6sC+c xstjjtj tj fd}t |t r1t }|jD]\}}||}|||<|St |ttfr2|Dcgc] }|| }}t |tr t|}|St |tjsJ||Scc}w)Nck(r,tDcgc]}tj|}}nd}tj||tj |Scc}w)Nr\r?)rCr)rxrr^rJ)r:rLrYryr]r this_rankrs r rzz$gather_container.._do_gathers\  =B:=NO5++F3OKOK FKS>yy'22 PsA%) rrr#r rr{r|r}r~rr)r) rr]rryrzrrrrsrrs ``` @@r gather_containerrs  %TZZ%%E$$U+J e$I33)T"6OO% DAq1 AHQK  Ie} -+45aJqM55 i 'XH)U\\222)$$6s2D)N)rN)Tr)__doc__ functoolsr-rUrar*r)torch.distributed distributedrrrFr rrrboolrr lru_cacher$r;rQrEr^rfrurrr r r rs   !!# 5 5 ; ;   $6D$N <%4%r