Ph>yddlZddlZddlZddlmZddlmcmZddl m Z m Z m Z m Z mZmZddlmZddlmZddlmZddlmZ ddlmZej8rd Zn dd lmZ e e e"e e e"ejFd e d e"ffZ$ dZ%dNdejLde"de$de'fdZ(dNdejLde'de$de'fdZ) dNdejLde"de$de'fdZ* dNdejLde'de"de$de'f dZ+dNde ejLde'de$de'de ejLf dZ,dNde ejLde$de'de ejLfdZ- dNde ejLde'de e"de$de'de ejLf dZ.dZ/ dNdejLd e e e"d!e e e"de$de'dejLf d"Z0Gd#d$ejLZ1 dNde$de'de e'e e"e"ffd%Z2de3fd&Z4dejLfd'Z5d(Z6d)Z7d*Z8d+Z9d,Z:d-Z;d.Zd1Z?d2Z@d3ZAd4ZBd5ZCd6ZDd7ZEej8sejjd8d9ZHejjd8d:ZIeEejjd;d:ZJeJjde=d=eJjd?ed=eJjdAe9d=eJjdBeAd=eJjdCeBd=eJjdDeCd=eJjdEeDd=nejBdF dOdGejLdHejLdIe3de'de"f dJZL dPdGejLdHejLdKe'dIe3de"de'f dLZMddMlmNZOm+ZPeOeLePeMiZQy#e$r ddlmZYbwxYw#e $rejBd d ZYgwxYw)QN)TupleUnionListOptionalcast TYPE_CHECKING)_functional_collectives_impl)_register_tensor_wrapper)get_innermost_proxy_mode) impl_abstract) tree_map_onlycy)z9Can't import torchdynamo in torchdeploy builds currently.FrtC:\Users\daisl\Desktop\realtime-object-detection\venv\Lib\site-packages\torch/distributed/_functional_collectives.pyis_torchdynamo_compilingrsr) is_compilingzdUnable to import torchdynamo util `is_torchdynamo_compiling`, so won't support torchdynamo correctlycy)NFrrrrrr srzdist._tensor.DeviceMeshcTtjjj|S)z Wait on a tensor returned by the collectives ops. Waiting follows device semantics, which means blocking on CPU and synchronizing streams on CUDA. )torchopsc10d_functional wait_tensor)tensors rrrzs 99 $ $ 0 0 88rselfsrcgrouptagct||\}}}tjjj |||||}t |S)a Broadcasts the tensor to all processes in the given process group. Args: src (int): Source rank group (ProcessGroup or List[int]): The process group to work on. tag (str, optional): A unique identifier for the collective. Default: empty string ) _expand_grouprrr broadcast_maybe_wrap_tensor)rrrrrankset group_sizers rr"r"sD -UC8C* YY & & 0 0sC* UF f %%rreduceOpct||\}}}tjjj |||||}t |S)a Reduces the tensor data across all machines in such a way that all get the final result. The input tensor is left unmodified. Group can be one of: List[int]: ranks participating in the collective. List[List[int]]: 2D mesh of ranks taking part of this collective in MPMD. ProcessGroup: Will perform a collective using the ranks and tag of the PG. DeviceMesh: Do a SPMD collective over all ranks of the mesh (DeviceMesh, int): Do a MPMD collective over one dimension of the DeviceMesh :: N.B. If you pass a PG or a 1D list to perform a MPMD collective, the compiler won't be able to recover that information and perform collective algebraic optimization. Use other forms of input for that. )r!rrr all_reducer#)rr&rrr$r%rs rr(r(sE" -UC8C* YY & & 1 1$#wPZ [F f %%r gather_dimc|jsJt||\}}}tjjj ||||}t |}|dk7r-tjtj||d|}|S)a% Gather tensor data across from all machines and concatenate over ``gather_dim``. Note that it currently only supports gather_dim = 0. The input tensor is left unmodified. Group can be one of: List[int]: ranks participating in the collective. List[List[int]]: 2D mesh of ranks taking part of this collective in MPMD. ProcessGroup: Will perform a collective using the ranks and tag of the PG. DeviceMesh: Do a SPMD collective over all ranks of the mesh (DeviceMesh, int): Do a MPMD collective over one dimension of the DeviceMesh :: N.B. If you pass a PG or a 1D list to perform a MPMD collective, the compiler won't be able to recover that information and perform collective algebraic optimization. Use other forms of input for that. rdim) is_contiguousr!rrrall_gather_into_tensorr#catchunk)rr)rrr$r%rress rall_gather_tensorr2s},     ,UC8C* YY & & = =dCR\ ]F V $CQii C;L Jr scatter_dimc`t||\}}}|j||zdk(sJd|jdd||dk7r-tj|||}tj|}tj j j|||||}t|} | S)a( Reduces the tensor data across all machines in such a way that all get the final result, then scatter the results to corresponding ranks. The input tensor is left unmodified. Group can be one of: List[int]: ranks participating in the collective. List[List[int]]: 2D mesh of ranks taking part of this collective in MPMD. ProcessGroup: Will perform a collective using the ranks and tag of the PG. DeviceMesh: Do a SPMD collective over all ranks of the mesh (DeviceMesh, int): Do a MPMD collective over one dimension of the DeviceMesh :: N.B. If you pass a PG or a 1D list to perform a MPMD collective, the compiler won't be able to recover that information and perform collective algebraic optimization. Use other forms of input for that. rzinput dimension 0 (" must be a multiple of group_size r+) r!sizerr0r/rrreduce_scatter_tensorr#) rr&r3rrr$r% tensor_listrr1s rr7r7s, -UC8C* ++q0Z TYYq\N*LZLYZ0akk$ D yy% YY & & < z$all_to_all_single..\s#XEWT:dS%,,$78EW*,c3\K|]$}t|ttjf&ywrXrYr\s rr^z$all_to_all_single..^s#WEVT:dS%,,$78EVr_)allr!rrrall_to_all_singler#)rrTrUrrr$r%rs rrbrbEs,%XEWXXlZllX$WEVWWjYjjW,UC8C* YY & & 8 8?QSdfikrt~ F f %%rceZdZUdZej ed<dgZejjZ e dej fdZ dZ dZe dZdZdZd ej fd Zd Zedd ZdZy )AsyncCollectiveTensora A Tensor wrapper subclass that is used to trigger a call to wait prior to first use of the underlying tensor. Use it inside functional collective pytorch wrappers like the following: def functional_collective(self, group, tag): tag, rankset, group_size = _expand_group(group, tag) tensor = torch.ops.c10d_functional.{collective}(self, tag, rankset, group_size) return _maybe_wrap_tensor(tensor) elemc tjj||j|j |j |j |j|jd}||_ |S)NF)stridesstorage_offsetdtypelayoutdevice requires_grad) rTensor_make_wrapper_subclassr6striderhrirjrkre)clsrers r__new__zAsyncCollectiveTensor.__new__ts` LL / / KKM$2E2E2G**T[[;;e 0  rc dgdfSNrerrs r__tensor_flatten__z(AsyncCollectiveTensor.__tensor_flatten__sx~rc`t|j|jjSrX)rretolistrus rrxzAsyncCollectiveTensor.tolists!DIIyy!!rc*|J|d}t|Srt)rd) inner_tensorsmetares r__tensor_unflatten__z*AsyncCollectiveTensor.__tensor_unflatten__s!||V$$T**rcLt|jd|jdS)NzAsyncCollectiveTensor()rrerus r__repr__zAsyncCollectiveTensor.__repr__s!DII' {!44rc0t|j|SrXrrus r trigger_waitz"AsyncCollectiveTensor.trigger_waitsDII rr9cDt|j|jSrXrrus rwaitzAsyncCollectiveTensor.waitsDIIyyrc|jS)zOThis method enables _functional_collectives_impl to test if a tensor is an ACS)rerus r_get_acs_underlying_tensorz0AsyncCollectiveTensor._get_acs_underlying_tensors yyrNc t| dtf fd }dtjfd}t t||}t t||}||i|} rt tj|| } | S)NecJst|j|jSrXr)r is_view_ops runwrapz8AsyncCollectiveTensor.__torch_dispatch__..unwrapsAFF#66MrcVt|trJt|}t||SrX)rIrdr )rr1s rwrapz6AsyncCollectiveTensor.__torch_dispatch__..wraps*!!%:; ;;'*C $S )Jr)rSrdrrmr) rpfunctypesargskwargsrrunwrapped_argsunwrapped_kwargsoutrs @r__torch_dispatch__z(AsyncCollectiveTensor.__torch_dispatch__sy &  +   ELL ''ON7&67  dC8C rc>|jjSrX)rnumpyrus rrzAsyncCollectiveTensor.numpysyy{  ""r)rN)__name__ __module__ __qualname____doc__rrm__annotations__ __slots___C_disabled_torch_function_impl__torch_function__ staticmethodrrrvrxr|rrrr classmethodrrrrrrdrdds ,,I?? 5<<  "++ 5ell8#rrdc ddlmcm}trd}d}nd}d}t |t rt |dt r^||}g}d}|D]K}|j ||dk7r(|t|k7rtd|d t|t|}Mn;||}t|}n&t |tjr:tj|}t|}|xstj|}nt ||jr4|jd k(sJd |j d\}}t|}nt |t"rmt|d k(rTt |d|jr;t |d t$r(|d} |d } | j | \}}t|}ntd td|||fS)a5 _expand_group desugars the different RANK_TYPES types into a canonical format that is traceable. By having this be part of the explicit eager codepath, we avoid having to specialize behavior inside torchdynamo and can still interoperate with processgroup objects or other untraceable forms. rNc>tttt|SrXrrrZxs rcast_listlistintz'_expand_group..cast_listlistintsT#Y+ +rc0ttt|SrXrrs r cast_listintz#_expand_group..cast_listintsS 1% %rc|SrXrrs rrz'_expand_group..cast_listlistintHrc|SrXrrs rrz#_expand_group..cast_listintrrz$group sizes must be identical found z and r zJOnly 1D mesh is supported, pass in (DeviceMesh, int) together if mesh > 1Dz1Invalid tuple for group must be (DeviceMesh, int)z[Invalid type for group, must be one of List, Processgroup, DeviceMesh or (DeviceMesh, int).)torch.distributed._tensor distributed_tensorrrIr<extendrB ValueErrordist ProcessGroupget_process_group_ranksc10d_get_group_tag DeviceMeshndim_dim_group_infostuplerZ) rrdtrr nested_listr$r%rsdmeshr,s rr!r!s+*  , &  % eAh %*51KGJ!r"# c"g(=$>zl%PSTVPWyY!W "#5)GWJ E4,, -..u5\ /T((/ E2== )zzQl ll--a0 W\ E5 ! u:?z%(BMMBzRWXYRZ\_G`!HE(C 11#6LCWJPQ Qvww * %%rctrytjjtjjj yt }|y|jduS)NTF)rrr_get_dispatch_mode_TorchDispatchModeKey FUNCTIONALr tracer)modes r_are_we_tracingrsT! xx""588#A#A#L#LMY # %D | ;;d ""rctr t|St|}t|t t j |SrX)rrrdr rrrm)rr1s rr#r#s64   %CS!  c ""rcDfd}|Dcgc] }|| c}Scc}w)Ncvt|j}|dxxzcc<|j|}|SrHr<r6 new_empty)shardout_size out_tensorr%s r mk_out_tensorz=_all_gather_into_tensor_coalesced_meta..mk_out_tensors5 % z! __X. rr)rrr$r%rts ` r&_all_gather_into_tensor_coalesced_metars' '+ +dM! d ++ +c,tj|SrXr empty_likerrs r_broadcast_metar"   D !!rc,tj|SrXrrs r_all_reduce_metar%rrc,tj|SrXrrs r_wait_tensor_metar(rrcpt|j}|dxx|zcc<|j|SrHr)rrr$r%rs r_all_gather_into_tensor_metar+s/EJJL!H QK:K ??8 $$rcpt|j}|dxx|zcc<|j|SrHr)input reduce_oprr$r%rs r_reduce_scatter_tensor_metar0s/EJJL!H QKJK ??8 $$rcR|Dcgc]}tj|c}Scc}wrXr)rrrs r_all_reduce_coalesced_metar5s%)- .AE  Q  .. .s$c|SrXr)inprs r_all_reduce__metar8s Jrc|SrXr)r@rs r_all_reduce_coalesced__metar;s MrcDfd}|Dcgc] }|| c}Scc}w)Ncvt|j}|dxxzcc<|j|}|SrHr)rrrr%s rrz<_reduce_scatter_tensor_coalesced_meta..mk_out_tensor?s5 % " __X. rr)r@r&rr$r%rrs ` r%_reduce_scatter_tensor_coalesced_metar>s' '- -fM! f -- -rc||j|jS|D]}tj|t |j}t ||d<|j|SrH)rr6r_check_is_sizer<sum)rrTrUrr$r%srs r_all_to_all_single_metarLsc!uzz|,,#A   #$ %,- x((rcpt|j}|dxx|zcc<|j|SrHr)rr% group_nameshapes r#_all_gather_into_tensor_native_metarVs/  E !H H ??5 !!rcB|Dcgc]}t|||c}Scc}wrX)r)r@r%rrs r-_all_gather_into_tensor_coalesced_native_metar[s2 E ,E:zJ  scpt|j}|dxx|zcc<|j|SrHr)rrr%rrs r"_reduce_scatter_tensor_native_metaras/  E !HH == rc D|Dcgc]}t||||c}Scc}wrX)r)r@rr%rrs r,_reduce_scatter_tensor_coalesced_native_metarfs4 C +3 :zR  scjgd}tjt}|D]}|d|jd}t t d|}t |d|d}t j|tjjtj||dtd||y) N) zObroadcast(Tensor self, int src, str tag, int[] ranks, int group_size) -> TensorzUall_reduce(Tensor self, str reduceOp, str tag, int[] ranks, int group_size) -> Tensorzcall_reduce_coalesced(Tensor[] self, str reduceOp, str tag, int[] ranks, int group_size) -> Tensor[]z"wait_tensor(Tensor self) -> TensorzTall_gather_into_tensor(Tensor shard, str tag, int[] ranks, int group_size) -> Tensorzball_gather_into_tensor_coalesced(Tensor[] input, str tag, int[] ranks, int group_size) -> Tensor[]zareduce_scatter_tensor(Tensor input, str reduceOp, str tag, int[] ranks, int group_size) -> Tensorzpreduce_scatter_tensor_coalesced(Tensor[] inputs, str reduceOp, str tag, int[] ranks, int group_size) -> Tensor[]zall_to_all_single(Tensor input, SymInt[]? output_split_sizes, SymInt[]? input_split_sizes, str tag, int[] ranks, int group_size) -> Tensorr(__meta)tagsCompositeExplicitAutogradzc10d_functional::)sysmodulesrindexgetattr fun_col_implc10_libdefinerTagpt2_compliant_tag c10_lib_implimplr )ops_defs my_moduleop_defop_name backend_impl meta_impls r _register_opsrls H H%I6<<,-|q ]; I7)5'9: vEII$?$?@'<1LM4 )'34Y? rrDEFIMPL_c10d_functionalr(Meta all_reduce_r;all_reduce_coalesced_rr.r?r7rEzJPyTorch Distributed functional collectives do not work with torch::deploy.outputrasync_opcN|rJd|jt||||SNz@Can't remap async version of inplace op to functional collective)copy_r2)rrrrrr)s rall_gather_tensor_inplacer!s-[[[< <<)%UCH IIropc P|rJd|jt|||||Sr)r r7)rrr"rrr3rs rreduce_scatter_tensor_inplacer$s/[[[< <<-eReSQ RRr)r.r7))Fr%r)rNFrr%)Rwarningsrrtorch.distributedrr"torch.distributed.distributed_c10ddistributed_c10drtypingrrrrrrr%r rr "torch.fx.experimental.proxy_tensorr torch._custom_opsr torch.utils._cxx_pytreer ImportErrortorch.utils._pytree_running_with_deployrtorch._dynamo.external_utilsr ExceptionwarnrZr RANK_TYPESrrmstrr"r(r2r7r;r?rErSrbrdr!boolrr#rrrrrrrrrrrrrrrrlibraryLibraryr r  _c10_lib_implrr!r$r.legacy_allgatherlegacy_reducescattertraceable_collective_remapsrrrr=s| 11DD:B,25 5Y #J 49d49ot/@/@B[]bc|BdB^CCD 89 &ELL &s &: &C &&U\\&S&&#&4  ,,  H  ,,      F6tELL16S66Z]6gklqlxlxgy6,64 +=6j6WZ6dhiniuiudv66 "6  "6"6c"6  "6 "6  %,, "6NV & ,,& c+& S *&  & &  \\ &>W#ELLW#rB&B&#B&uS$s)S=P7QB&H # ## #,"""% % /.)"    @."u!!#mm##$5u=G==(():FCLOMM))*}&7@-/I6R.0KVT}&7@/1TV\]9;hjpq.0RTZ[8:fhnoHMM^_ J LL J << J J J  J  S LL S << S  S  S  S  S/7a2112  r   s$N!N1 N.-N.1OO