tjX@ddlZddlZddlZddlZddlZddlZddlmZmZddl m Z ddl m Z ddl mZddlmZmZmZddlZddlmZddlmZddlmZd d lmZmZmZmZd d l m!Z!m"Z"d d gZ#edZ$edZ%de&e$ezde'e(effdZ) d0de'e*efdej+dzde,e*fdZ- d0de'e*efdej+dzddfdZ.GddZ/dede!defdZ0dej1de!dej1fdZ2de"de!defdZ3d ee(d!ee(de&e(fd"Z4d ee(d!ee(de&e(fd#Z5Gd$d%ej6Z7d&ej6d'e(d(e(dej6fd)Z8d*e*d+e(de*fd,Z9d-Z:e d.Z;d/Zz%_get_failure_dict..'s,OOOFAs4I#4N4NOCOOO)r dictintr enumerate)rs r _get_failure_dictr&"s=  S# #$OOi00OOO  r" local_dictgroupct|}dgtj|z}tj|||t t j|S)z*Gathers all keys, and returns them sorted.Nr() listkeysdistget_world_sizeall_gather_objectset itertoolschain from_iterable)r'r(r, gathered_keyss r _all_gather_keysr5+sh  !! " "D&*Vd.A%.H.H%HM=$e<<<< y,,];; < <6s =))Q.. M::H*//##$$G g D 4yy1}} Q4 Q Q   }r"c eZdZdZdejdzdedefdZdefdZ defd Z d e dzde fd Z d e de e dzfd Zd e de e fd Zde e dzde fdZdedege fdee e ge efdefdZdedege fdee e gefdefdZdedege fde e fdZdedege fde fdZddZdS) _DistWrapperaG This is a wrapper around PG that provides a series of features around object collectives. It works without distributed initialized, where most collectives turn into nops. All variants that take functions are exception robust, meaning that if one or more ranks raise errors, all ranks will observe those. r(Nuse_distcoordinator_rankc||_||_||_|jrI|tj||n||_tj||_|j|k|_dSd|_d|_d|_dS)NrT) r(rArBr-get_global_rankglobal_coordinator_rankget_rankrankis_coordinator)selfr(rArBs r __init__z_DistWrapper.__init__Us    0 = '$$U,<===%  (  e,,DI"&)/?"?D   +,D (DI"&D   r"rc|jSN)rGrIs r rFz_DistWrapper.get_rankks yr"cF|jrtj|jSdS)Nr )rAr-r.r(rMs r r.z_DistWrapper.get_world_sizens$ = 3&tz22 2qr"objectc|g}|jr!tj||j|jt t |dS)z_Implement functionality similar to c10d::broadcast_object_list but without distributed enabled.) object_listr(srcr)rAr-broadcast_object_listr(rEr r)rIrOrQs r broadcast_objectz_DistWrapper.broadcast_objectssQh =   &'j0     A{1~&&&r"c|jrr|jr;tttdgt j|jznd}t j||jr|nd|j |j|}n|g}|S)zWImplement functionality similar to c10d::gather_object but without distributed enabled.N)objobject_gather_listdstr() rArHr r+rr-r.r( gather_objectrE)rIrO gather_objsresults r rYz_DistWrapper.gather_object~s = &T!Wtft':4:'F'FFGGG   262E#O;;40j      !FFXF r"c|jrXtttdgt j|jz}t j|||jn|g}|S)z[Implement functionality similar to c10d::all_gather_object but without distributed enabled.N)rQrVr()rAr r+rr-r.r(r/)rIrOrZs r r/z_DistWrapper.all_gather_objectsj = #tAw1DTZ1P1P(PQQK  "'V4:     "(Kr"rQc|jrUtttdg}t j||jr|nd|j|j|d}n|td|d}|S)zXImplement functionality similar to c10d::scatter_object but without distributed enabled.N)scatter_object_output_listscatter_object_input_listrRr(rzobject_list is None) rAr r+rr-scatter_object_listrHrEr(r:)rIrQ gather_result local_replys r scatter_objectz_DistWrapper.scatter_objects = ) a4&11M  $+89=9L*V++RV0j     (*KK"$%:;;;%a.Kr"stepmap_fun reduce_func  |}n&#t$r}t|}Yd}~nd}~wwxYw||}d}|jr|t dt |}t |dkr ttttz|ttt|}n.#t$r!}t|||j <Yd}~nd}~wwxYwt |dkr&t||g| z}||} t| tr| | S)a^ Compute a value on each rank, then do centralized reduce on a single rank, followed by a scatter. This method operates in the following way: Run ``map_fun`` on all ranks Gather results on rank 0 Call ``reduce_fun`` on all those values Scatter to each rank part of the result. Nall_data is Noner) BaseExceptionrrYrHr:r&r9r r+rrrrGr.rc isinstance) rIrdrerf local_dataeall_data all_results node_failuresr[s r reduce_scatterz_DistWrapper.reduce_scatters  , JJ , , ,(++JJJJJJ ,%%j11<@   *$%7888-h77M=!!Q&&B"&Q!445" 4Q#:#:;;##KK%BBB/>q/A/AM$),,,,,,B=!!A%%'m<<''))* $$[11 f1 2 2 L s) 0+0AC DC;;Dc |}n&#t$r}t|}Yd}~nd}~wwxYw||}d}|jr|t dt |}t |dkrY |ttt|}n.#t$r!}t|||j <Yd}~nd}~wwxYwt |dkrt||}| |} t| tr| tt| S)aa Compute a value on each rank, then do centralized reduce on a single rank, followed by a broadcast. This method operates in the following way: Run ``map_fun`` on all ranks Gather results on rank 0 Call ``reduce_fun`` on all those values Broadcast the reduced value to all ranks. Nrhr)rirrYrHr:r&r9r r+rrGrrTrjr) rIrdrerfrkrlrmr[ro final_results r all_reducez_DistWrapper.all_reducesr  , JJ , , ,(++JJJJJJ ,%%j1115   B$%7888-h77M=!!Q&&B'ZT!Wh(?(?@@FF$BBB/>q/A/AM$),,,,,,B=!!A%%,T=AA,,V44 l$7 8 8  A|$$$s( 0+0)B// C9CCc4 |}n&#t$r}t|}Yd}~nd}~wwxYw||}t|}t |dkrt ||t tt|S)z Compute a value on each rank, then all_gather them. This method operates in the following way: Run ``map_cp`` on all ranks all_gather the values to all ranks Nr) rirr/r&r9rr r+r)rIrdrer[rlrnros r all_gatherz_DistWrapper.all_gather s (WYYFF ( ( ($Q''FFFFFF (,,V44 )+66 }   ! !%dM:: :DG[)))s 0+0c$d}|jrG |}n;#t$r.}t||jt |i}Yd}~nd}~wwxYw||}t |tr|tt|S)z Compute a value on rank 0 and broadcast it. This method operates in the following way: Run ``map_cp`` on rank 0 broadcast the value N) rHrirrGrrTrjr r)rIrdrer[rlrrs r broadcastz_DistWrapper.broadcast$s26   T T   T T T,TDIq?Q?Q3RSS T,,V44 l$7 8 8  A|$$$s  A$A  AcL|jsdStj|jdS)a Add a synchronization point across all processes when using distributed. If torch.distributed is initialized, this function will invoke a barrier across the global process group. If torch.distributed is not initialized, this function is a no-op. Nr*)rAr-barrierr(rMs r ryz_DistWrapper.barrier=s. }  F 4:&&&&&&r")rN)__name__ __module__ __qualname____doc__r- ProcessGroupboolr$rJrFr.rrTr+rYr/rcstrrrrprsrurwryrr"r r@r@Ks]' 4''' '''',# 'q4x 'A ' ' ' 'A$q'D.(  d1g    $q'D.Q$.."a%.d1gYQ/0 . ....`)%)%"a%)%d1gY\* )% )%)%)%)%V**"a%* a ****2%%"a%% %%%%2''''''r"r@tensorindexc|jtd|jd|}|jWt ||jkr?t j||jjj |jkr ||jS|D]-}t j|jj |jkr|cS.td|jd|jd)NzCannot lookup z5 since its a ShardedTensor and no offset was providedzCould not find shard at 'z ' for FQN: '') offset ValueErrorfqn local_shardsrr9torchSizemetadata shard_offsets)rrshardsshards r _find_shardrHs | ]UY ] ] ]    " "F { KK%+ % % 6%+.7EFF%,VV%+& & :en2 3 3u| C CLLL D WWW59WWW X XXr"cvt|dr||St|trt ||jS|jb|jtjdgt| zkr|Std|j d|jd|S)N__get_tensor_shard__rFQN: '1' is not a ShardedTensor, can't find by offset: 'r) hasattrrrjr rrrrrr9sizerr)rrs r rr]sv-..2**5111&-((165))00 | <5:qcC ,>,>&>?? ? ?M `UY ` `QVQ] ` ` `    Mr"c|j|vrtd|jd||j}t|tjrt ||S|j td|jd|jd|S)NzCould not find FQN: 'rrr)rrrjrTensorrr)r6rrVs r rrms y ""====>>> UY C#u|$$  e,,,  ! `UY ` `QVQ] ` ` `    Jr"abc6dt||DS)Ncg|] \}}||z Srrri_ai_bs r z%_element_wise_add..| 0 0 0(#sC#I 0 0 0r"ziprrs r _element_wise_addr{ 0 0c!Qii 0 0 00r"c6dt||DS)Ncg|] \}}||z  Srrrs r rz%_element_wise_sub..rr"rrs r _element_wise_subrrr"ceZdZdejdedeffd ZejfdededefdZ defdZ de fd Z de fd Z d Zdd ZxZS) _ReaderView base_streamrr9ct||_||_||_|ddSNr)superrJrr9rseek)rIrrr9 __class__s r rJz_ReaderView.__init__sB  & ! r"whencerc|tjkr |j|z}n.|tjkrtj}|j|jz|z }|j||SrL)osSEEK_SETrSEEK_ENDr9rr)rIrrs r rz_ReaderView.seeks] R[ [6)FF r{ " "[FkDH,6F$$VV444r"cD|j|jz SrL)rtellrrMs r rz_ReaderView.tells$$&&44r"c4|jSrL)rreadablerMs r rz_ReaderView.readable((***r"c4|jSrL)rseekablerMs r rz_ReaderView.seekablerr"c|j|z }|dkrdSt||krt|d|}|j|Sr)r9r memoryviewrreadinto)rIrmax_sizes r rz_ReaderView.readintos`8diikk) q==1 q66H  1 ixi(A((+++r"c|j|z }|dks||kr|}|j|S)Nr)r9rrread)rIrrs r rz_ReaderView.readsC8diikk) 2::D$$T***r")r)rzr{r|ioIOBaser$rJrrrrrrrrr __classcell__)rs@r rrsBIs/1k5535555555c5555+$+++++$++++,,,++++++++r"rfilerlengthc$t|||SrL)r)rrrs r _create_file_viewrs tVV , ,,r" device_type device_idc |dkrdS|d|S)zDevice info normalization.cpu:r)rrs r _normalize_device_infors&eu ' 'I ' ''r"Fc#Ktrtjrtjdkrt j}| dV|t|}| d ddS#|t|}| d dwxYwdVdS)Nrtime ) ENABLE_PROFILEr- is_availablerFcProfileProfileenabledisabler sort_stats print_stats)profilerstatss r _profilers t022 dmoo6J6J#%% 5 EEE      (OOE   V $ $ 0 0 4 4 4 4 4      (OOE   V $ $ 0 0 4 4 4 4 s B--A C:cLtdtffd }|S)Nrct|dkrtjdjddt j}d|jD}d|vr$d|vrtd||f|d|d<n:d |vr$d |vrtd ||f|d|d <ntd ||d fi|S|i|S) NzThe argument order of zG has been changed. Please check the document to avoid future breakages.) stacklevelc<g|]}|j|jk|jSr)kind KEYWORD_ONLYname)rps r rz5_api_bc_check..inner_func..s/16Q^;S;S;S;S;Sr"storage_writerzstorage_writer in kwargs: r storage_readerzstorage_reader in kwargs: zUnexpected kwonlyargs = r) r9warningswarnrzinspect signature parametersvaluesr: RuntimeError)argskwargssig kwonlyargsfuncs r inner_funcz!_api_bc_check..inner_funcsX t99>> MGGGG     #D))C # 5 5 7 7J :--#v--()VtVn)V)VWWW+/7'((!Z//#v--()VtVn)V)VWWW+/7'(("#Jj#J#JKKK4Q**6** *4((( (r")rr)rrs` r _api_bc_checkrs@ 4[[)s)))))[)2 r"rL)=rrrr1rrcollections.abcrr contextlibr functoolsrpstatsrtypingrr r rtorch.distributed distributedr-'torch.distributed._shard.sharded_tensorr -torch.distributed._shard.sharded_tensor.shardr apirrrrrrr__all__rrr+r#r$r&rr~r0r5r>r@rrrrrrrrrrrrrrr"r rs  ........%%%%%%%%%%%%%%%% AAAAAA?????? 54444444  8 9 GCLL GCLL !'' ( # !CG==S#X='+'84'?=X====KO  S#X /3/@4/G      *z'z'z'z'z'z'z'z'zY YmYYYYY* el = U\      } QT    1#18C=1T#Y11111#18C=1T#Y1111%+%+%+%+%+")%+%+%+P-BI-s-C-BI---- (((((((    r"