tjddlZddlZddlZddlZddlmZmZmZddlZddl m Z ddl m cm ZddlmZddlmZddlmZddlmZ ddlmZn#e$r ddlmZYnwxYw dd lmZn #e$rej d d d ZYnwxYw e!e"e!e!e"ze j#zeze$de"fzej%zZ&ddlm'Z' dZ(ddej)de"de&de*fdZ+ddej)de*de&de*fdZ, ddej)de"de&de*dej)f dZ- ddej)de"de&de*fdZ. ddej)de*de"de&de*f dZ/ ddej)de*de"de&de*f dZ0 ddej)de"de&de*dej)f d Z1 ddej)de"de&de*fd!Z2 ddej)de*de"de&de*f d"Z3 ddej)de*de"de&de*f d#Z4 dde!ej)de*de&de*de!ej)f d$Z5 dde!ej)de&de*de!ej)fd%Z6 dd&e!ej)de*de!e"de&de*de!ej)f d'Z7 dde!ej)de&de*de!ej)fd(Z8 dd&e!ej)de*de!e"de&de*de!ej)f d)Z9d*Z: ddej)d+e!e"dzd,e!e"dzde&de*dej)f d-Z; ddej)d+e!e"dzd,e!e"dzde&de*dej)f d.Zej?@d2e=e>3d/ej)fd4ZAd5ZBej?@d6eAeB3d/ej)fd7ZCd8ZDej?@d9eCeD3d/ej)fd:ZEd;ZFej?@dZHej?@d?eGeH3ej?@d@eCeD3ej?@dAeEeF3ej?@dBeGeH3dCe!ej)fdDZIdEZJej?@dFeIeJ3dCe!ej)fdGZKdHZLej?@dIeKeL3dCe!ej)fdJZMdKZNej?@dLeMeN3 ddej)dMe!e"de&de*dej)f dNZOGdOdPej)ZP dde&de*de$e*e!e"e"ffdQZQ dde&de*de j#ej%zfdRZRdde&de*dej%fdSZSej?TdTdUdVWdXej)dej)fdYZUeUjVdXej)dej)fdZZWd/ej)fd[ZXd\ZYeU@eXeY3deZfd]Z[dej)fd^Z\ej]dd`eZfdaZ^dbZ_dcZ`ddZadeZbdfZcdgZddhZediZfdjZgdkZhdlZidmZjdnZkdoZldpZmdqZndrZodsZpdtZqduZrdvZsdwZtej?udxdyZvevwdzebd{evwd|ejd{evwd}eid{evwd~eld{evwdecd{evwdedd{evwdeed{evwdefd{evwdeod{evwdepd{evwdeqd{evwderd{evwdesd{evwdetd{evwdend{evwdead{evwdekd{ej?uddyZxexwdepd{exwderd{exwdend{ejyjz{ej|j}j(j~ejyjz{ej|j}j(ejyjz{ej|j}jj~ejyjz{ej|j}jejyjz{ej|j}jj~ejyjz{ej|j}jejyjz{ej|j}jj~ejyjz{ej|j}jej?uddZej?uddyZgdZejeZeD]fZededZeedeZeeejjeweedg ddej)dej)deZde*de"f dZ ddej)dXej)de*deZde"de*f dZe jjde jjde jjde jjde jjde jjde jjde jjdiZ ddej)de*deZde*fdZ ddej)dXej)de*fdZ dde!ej)dej)de*fdZ ddej)de"de"de j#dzde"f dZ ddej)de"de"de j#dzde"f dZde!e*de!e"de!e"de!ej)de&f dZde j#ej%zde j#ej%zfdZddl mZmZmZmZm-Zm,Zm;ZmZmZmZm/Zm3ZdZdZdZdZdZdZdZdZeeeeeeeeeeeeeeeeeeeeeeeei ZdS)N)Anycast TYPE_CHECKING)_maybe_view_chunk_cat) DeviceMesh)get_proxy_mode)_functional_collectives_impl) tree_map_only)is_dynamo_compilingzdUnable to import torchdynamo util `is_torchdynamo_compiling`, so won't support torchdynamo correctly stacklevelcdS)NFrg/var/www/html/CA-Chatbot/venv/lib/python3.11/site-packages/torch/distributed/_functional_collectives.pyis_torchdynamo_compilingr surzdist.tensor.DeviceMesh)_chunk_or_narrow_catcJtjj|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. )torchops_c10d_functional wait_tensor)tensors rrrs 9 % 1 1& 9 99rselfsrcgrouptagct||}tjj||t |}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 )_resolve_grouprrr broadcast_group_or_group_name_maybe_wrap_tensor)rrrr rs rr#r#sM 5# & &E Y ' 1 1 c'..F f % %%rreduceOpct||}tjj||t |}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_reducelowerr$r%)rr&rr rs rr(r(sW" 5# & &E Y ' 2 2 hnn 4U ; ;F f % %%r gather_dimreturnct||}tj|}tjj||t|}t|}|dkrt|tr`t|j }|dkrtj|d|nd}|d|ko|dk} | s|}t!|||}|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. rr )r"c10d_get_group_size_by_namerrrall_gather_into_tensorr$r% isinstanceAsyncCollectiveTensorlistshapemathprodwaitr) rr*rr group_sizerresr3 numel_between can_use_views rall_gather_singler;s, 5# & &E-e44J Y ' > > j.u55F V $ $CQ c0 1 1 !OOE>H1nnDIeAjL&9:::RSM 8z1Hmq6HL !hhjj#CZ@@ Jrc&t||||S)a< Gather tensor data across from all machines and concatenate over ``gather_dim``. Note that it currently only supports gather_dim = 0. This function is the same as all_gather_single but will propagate the backwards gradient across workers. See all_gather_single for more details on usage. )r;rr*rr s rall_gather_single_autogradr>s" T:uc : ::r scatter_dimct||}tj|}|||zdkr)t d|dd|d|dkrt |||d}t jj || |t|}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 )) narrow_dimcat_dim) r"r-r.sizeAssertionErrorrrrrreduce_scatter_tensorr)r$r%)rr&r?rr r7rr8s rreduce_scatter_singlerHs, 5# & &E-e44J yy *a// _$))A,, _ _R\ _ _ _   a#D*VWXXX Y ' = = U## F V $ $C Jrc(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. This function is the same as reduce_scatter_single but will propagate the backwards gradient across workers. Currently only the "sum" reduceOp is supported. See reduce_scatter_single for more details on usage. )rHrr&r?rr s rreduce_scatter_single_autogradrK!s& !xeS I IIrctjstjdt dt ||||S)Nz`torch.distributed._functional_collectives.all_gather_tensor` is deprecated. Please use `torch.distributed._functional_collectives.all_gather_single` instead.r r)rcompiler is_compilingwarningswarn FutureWarningr;r=s rall_gather_tensorrR=sU > & & ( (   `      T:uc : ::rctjstjdt dt ||||S)Nz`torch.distributed._functional_collectives.all_gather_tensor_autograd` is deprecated. Please use `torch.distributed._functional_collectives.all_gather_single_autograd` instead.r r)rrMrNrOrPrQr>r=s rall_gather_tensor_autogradrTMsU > & & ( (   i      &dJs C CCrctjstjdt dt |||||S)Nz`torch.distributed._functional_collectives.reduce_scatter_tensor` is deprecated. Please use `torch.distributed._functional_collectives.reduce_scatter_single` instead.r r)rrMrNrOrPrQrHrJs rrGrG]sW > & & ( (   d      !xeS I IIrctjstjdt dt |||||S)Nz`torch.distributed._functional_collectives.reduce_scatter_tensor_autograd` is deprecated. Please use `torch.distributed._functional_collectives.reduce_scatter_single_autograd` instead.r r)rrMrNrOrPrQrKrJs rreduce_scatter_tensor_autogradrWnsW > & & ( (   m      *$+uc R RRrct||}tjj||t |}ttt|S)a  Reduces a list of tensors across all machines in such a way that all get the final result. The all tensors in the input list are 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"rrrall_reduce_coalescedr)r$r2mapr%)rr&rr tensor_lists rrYrYsc& 5# & &E),AA U##K & 44 5 55rct||}tj|}tjj||t|}ttt|S)a Gather a list of tensors across from all machines. 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. ) r"r-r.rrr all_gather_into_tensor_coalescedr$r2rZr%)rrr r7r[s rall_gather_single_coalescedr^si& 5# & &E-e44J),MM U##K & 44 5 55rinputsc t||}tj|}t|t|kr0t dt|dt|dt t ||D]\}\}}|||zdkr.t d|d||d|d||dkr.tj ||| } tj | ||<tj j |||t|} t!t#t$| S) a, Reduces a list of tensors across all machines in such a way that all get the final result, then scatter the results to corresponding ranks. The input tensors are 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. zLength of scatter_dim (z) must equal length of inputs (rBrzinput dimension z (rAz for tensor at index )dim)r"r-r.lenrF enumerateziprErchunkcatrrreduce_scatter_tensor_coalescedr)r$r2rZr%) r_r&r?rr r7idxrarr[s rreduce_scatter_single_coalescedris, 5# & &E-e44J ;3v;;&& ec+&6&6 e eWZ[aWbWb e e e   (K(@(@AA11]c6 ;;s  j (A - - E3EE&++c*:*:EE^hEE@CEE  !88+fjcBBBK)K00F3K),LLU## K & 44 5 55rctjstjdt dt |||S)Nz`torch.distributed._functional_collectives.all_gather_into_tensor_coalesced` is deprecated. Please use `torch.distributed._functional_collectives.all_gather_single_coalesced` instead.r r)rrMrNrOrPrQr^)rrr s rr]r]sS > & & ( (   j      'tUC 8 88rctjstjdt dt |||||S)Nz`torch.distributed._functional_collectives.reduce_scatter_tensor_coalesced` is deprecated. Please use `torch.distributed._functional_collectives.reduce_scatter_single_coalesced` instead.r r)rrMrNrOrPrQri)r_r&r?rr s rrgrgsX > & & ( (   n      +68[%QT U UUrct|tjjst dt |tj|tj j rdS|j }t|j dkr#|j d}|jduo |jj SdS)Nz$Expected torch._ops.OpOverload, got Fr)r0r_ops OpOverloadrFtype_C%_dispatch_has_kernel_for_dispatch_keyname DispatchKeyCompositeImplicitAutograd_schemarb arguments alias_infois_write)tgtschema first_args r _is_view_opr| s c5:0 1 1QODIIOOPPP x55  E%?u [F 6 q  $Q' #4/U 8L8U4UU! routput_split_sizesinput_split_sizesc|+td|Dstd||+td|Dstd|t||}tj|}||)||td|jd|zg|z}|}t jj |||t|}t|S)aC Each process splits input tensor and then scatters the split list to all processes in a group. Then concatenate the received tensors from all the processes in the group and return single output tensor. 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. Nc3XK|]%}t|ttjfV&dSNr0intrSymInt.0rEs r z$all_to_all_single..2sD  6:Jtc5<0 1 1      rz2All output_split_sizes must be int or SymInt, got c3XK|]%}t|ttjfV&dSrrrs rrz$all_to_all_single..9s3WWT:dS%,$788WWWWWWrz1All input_split_sizes must be int or SymInt, got z^output_split_sizes and input_split_sizes must either be specified together or both set to Noner) allrFr"r-r.r3rrrall_to_all_singler$r%)rr}r~rr r7rs rrrsM,%  >P      !YEWYY $WWEVWWWWW  WDUWW  5# & &E-e44J!%6%>"*/@/H 9 #jmz9:ZG. Y ' 9 9 U## F f % %%rc(t|||||S)z: Same as all_to_all_single but supports autograd. )r)rr}r~rr s rall_to_all_single_autogradrPs T#57H%QT U UUr grad_outputc|S)a Backward for wait_tensor: identity (no-op). Wait is just a synchronization primitive, so gradient flows through unchanged. Args: ctx: Context object grad_output: Gradient from downstream operations Returns: Gradient unchanged (identity) rctxrs rwait_tensor_backwardrcs  rcdS)z Setup context for wait_tensor backward. Args: ctx: Context object to save state for backward inputs: Tuple of (tensor,) output: Output from forward pass Nrrr_outputs rwait_tensor_setup_contextrrs  Frz_c10d_functional::wait_tensor) setup_contextc|j}|j}|dkrtd|dtjj|||}t|ddfS)aw Backward for all_reduce: all_reduce with same reduce_op. Forward aggregates tensors, backward aggregates gradients. Args: ctx: Context object grad_output: Gradient from downstream operations Returns: Tuple of (grad_input, grad_group_name, grad_reduce_op) grad_group_name and grad_reduce_op are None (not differentiable) sumz8all_reduce backward only supports 'sum' reduction, got ''N) group_name reduce_op RuntimeErrorrrrr( contiguousr)rrrrrs rall_reduce_backwardrsJ IE Sy S S S   Y ' 2 2  )ZF v  d **rcR|\}}}||_||_dS)z Setup context for all_reduce backward. Args: ctx: Context object to save state for backward inputs: Tuple of (input, reduce_op, group_name) output: Output from forward pass Nrr)r)rr_rinputrrs rall_reduce_setup_contextrs,$* E9jCNOO%%CMMMrz_c10d_functional::all_reducec|j}|j}tjj|d||}t|ddfS)a Backward for all_gather_into_tensor: reduce_scatter with sum. Forward gathers tensors from all ranks, backward scatters gradients back with sum reduction. Args: ctx: Context object with group_name and group_size grad_output: Gradient from downstream operations Returns: Tuple of (grad_input, grad_group_size, grad_group_name) grad_group_size and grad_group_name are None (not differentiable) rN)rr7rrrrGrr)rrrr7rs rall_gather_into_tensor_backwardrs]JJY ' = =    F v  d **rc.|\}}}||_||_dS)z Setup context for all_gather_into_tensor backward. Args: ctx: Context object to save state for backward inputs: Tuple of (input, group_size, group_name) output: Output from forward pass Nrr7)rr_rrr7rs r$all_gather_into_tensor_setup_contextrs"%+!E:zCNCNNNrz(_c10d_functional::all_gather_into_tensorc|j}|j}|j}|dkrtd|dtjj|||}t|dddfS)a Backward for reduce_scatter_tensor: all_gather. Forward reduces and scatters tensors to ranks, backward gathers gradients from all ranks. Args: ctx: Context object with group_name, group_size, and reduce_op grad_output: Gradient from downstream operations Returns: Tuple of (grad_input, grad_reduce_op, grad_group_size, grad_group_name) grad_reduce_op, grad_group_size, grad_group_name are None (not differentiable) rzCreduce_scatter_tensor backward only supports 'sum' reduction, got 'rN) rr7rrrrrr/rr)rrrr7rrs rreduce_scatter_tensor_backwardrsJJ IE ^R[ ^ ^ ^   Y ' > >  F v  dD 00rcb|\}}}}||_||_||_dS)z Setup context for reduce_scatter_tensor backward. Args: ctx: Context object to save state for backward inputs: Tuple of (input, reduce_op, group_size, group_name) output: Output from forward pass Nrr7r)r)rr_rrrr7rs r#reduce_scatter_tensor_setup_contextrs506,E9j*CNCNOO%%CMMMrz'_c10d_functional::reduce_scatter_tensorc|j}|j}|j}tjj||||}t|dddfS)a Backward for all_to_all_single: all_to_all with reversed split sizes. Forward does all-to-all with specified split sizes, backward reverses them. Args: ctx: Context object with group_name, output_split_sizes, and input_split_sizes grad_output: Gradient from downstream operations Returns: Tuple of (grad_input, grad_output_split_sizes, grad_input_split_sizes, grad_group_name) All except grad_input are None (not differentiable) N) rr}r~rrrrrr)rrrr}r~rs rall_to_all_single_backwardrsjJ/-Y ' 9 9   F v  dD 00rc>|\}}}}||_||_||_dS)z Setup context for all_to_all_single backward. Args: ctx: Context object to save state for backward inputs: Tuple of (input, output_split_sizes, input_split_sizes, group_name) output: Output from forward pass N)rr}r~)rr_rrr}r~rs rall_to_all_single_setup_contextr9s3@Fz1all_reduce_coalesced_backward..{$BBBk   ! !BBBrN) rrrrrrrYr2rZr)rrrr grad_inputss rall_reduce_coalesced_backwardrcsJ IE ]QZ ] ] ]   ),AABB\BBBK [+.. / /t <.rrrN) rr7rrrrgr2rZr)rrrr7rs r)all_gather_into_tensor_coalesced_backwardrsiJJ),LLBB\BBB  K [+.. / /t <.rrN) rr7rrrrrr]r2rZr)rrrr7rrs r(reduce_scatter_tensor_coalesced_backwardrsJJ IE h\e h h h   ),MMBB\BBBK [+.. / /tT BBrcb|\}}}}||_||_||_dS)z Setup context for reduce_scatter_tensor_coalesced backward. Args: ctx: Context object to save state for backward inputs: Tuple of (tensor_list, reduce_op, group_size, group_name) output: Output from forward pass Nr)rr_rr[rr7rs r-reduce_scatter_tensor_coalesced_setup_contextrs56<2KJ CNCNOO%%CMMMrz1_c10d_functional::reduce_scatter_tensor_coalescedsrc_dstczt||\}}}tj|||}dg|z}dg|z} t|D]c\} } | t j|kr|| | <| t j|kr||| <dt||| ||S)a" Permutes the elements of the tensor according to the given source/destination pairs. `src_dst` should be defined such that src_dst[m] == n means m sends to n. 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 r) _expand_groupr-#_find_or_create_pg_by_ranks_and_tagrcdistget_ranknumelr) rrrr tranksetr7local_pgr}r~rdsts rpermute_tensorrs"+5#66Aw 77JOOHz)j(g&&33S $-)) ) )%)ZZ\\ c " $-)) ) )&*jjll s # T#57H%QT U UUrceZdZUdZejed<eed<ddgZe dejfdZ dZ dZ e dZ dd ed edzfd Zd efd ZdZd ejfdZdZeddZdZdS)r1a 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) elem completedc tj|||||j|j|j|j }||_ d|_ |S)N)stridesstorage_offsetdtypelayoutdevice requires_gradF) rTensor_make_wrapper_subclassrEstriderrrrrrr)clsrrs r__new__zAsyncCollectiveTensor.__new__/sp L / /  IIKKKKMM..00*;;, 0  rc dgdfS)Nrrrs r__tensor_flatten__z(AsyncCollectiveTensor.__tensor_flatten__?sx~rcN|Sr) trigger_waittolistrs rrzAsyncCollectiveTensor.tolistBs   ""))+++rcR|td|d}t|S)Nz5meta must be None for AsyncCollectiveTensor unflattenr)rFr1) inner_tensorsmeta outer_size outer_striders r__tensor_unflatten__z*AsyncCollectiveTensor.__tensor_unflatten__Es7   G V$$T***rNexpected_metadata expected_typecJ|tjurdS|Sr)rrr)rrrs r#__coerce_same_metadata_as_tangent__z9AsyncCollectiveTensor.__coerce_same_metadata_as_tangent__Ns(  , ,4  """rr+c2d|dS)NzAsyncCollectiveTensor(rB)rrs r__repr__zAsyncCollectiveTensor.__repr__Vs>(9(9(;(;>>>>rcX|jst|j}d|_|S|jSNT)rrr)routs rrz"AsyncCollectiveTensor.trigger_waitYs/~ di((C!DNJ9 rc*t|jSr)rrrs rr6zAsyncCollectiveTensor.waitas49%%%rc|jS)zOThis method enables _functional_collectives_impl to test if a tensor is an ACS)rrs r_get_acs_underlying_tensorz0AsyncCollectiveTensor._get_acs_underlying_tensords yrrc |tjjjjur.||dj|d}t |}|St| dt f fd }dtjfd}tt ||} tt ||} || i| } rttj|| } | S)Nrr ec>s|S|jSr)rr)r is_view_ops runwrapz8AsyncCollectiveTensor.__torch_dispatch__..unwrapts" (~~'''6Mrclt|trtdt|}|S)NzICannot wrap an AsyncCollectiveTensor inside another AsyncCollectiveTensor)r0r1rF)rr8s rwrapz6AsyncCollectiveTensor.__torch_dispatch__..wrapzs>!233 $_(**CJr) rratenviewdefaultrr1r|rr ) rfunctypesargskwargsr8 wrapper_resrrunwrapped_argsunwrapped_kwargsrrs @r__torch_dispatch__z(AsyncCollectiveTensor.__torch_dispatch__hs 59>&. . .$tAw|T!W--C/44K  &&  +        EL    ''OOdN7&677  9 dC88C rcN|Sr)r6numpyrs rr zAsyncCollectiveTensor.numpysyy{{  """rr)rN)__name__ __module__ __qualname____doc__rr__annotations__bool __slots__ staticmethodrrrrrrorstrrrr6r classmethodr r rrrr1r1s] ,OOO%I 5<   \ ,,,++\+DH##!$#59D[####?#????&el&&&&$$$[$L#####rr1c trd}d}nd}d}t|trt|dtru||}g}d}|D]a}|||dkr5|t |kr"t d|dt |t |}bn||}t |}nt|t jr;t j|}t |}|ptj |}nXt|trh|j d krtd |}t j|}t |}|ptj |}nt|trt |d krt|dtrzt|d t r_|d} |d } | | }t j|}t |}|ptj |}nt d t d |||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. cXtttt|Srrr2rxs rcast_listlistintz'_expand_group..cast_listlistintsT#Y++ +rcBttt|Srrrs r cast_listintz#_expand_group..cast_listintsS 1%% %rc|Srrrs rrz'_expand_group..cast_listlistintHrc|Srrrs rrz#_expand_group..cast_listintr!rrz$group sizes must be identical found z and r JOnly 1D mesh is supported, pass in (DeviceMesh, int) together if mesh > 1Dr z1Invalid tuple for group must be (DeviceMesh, int)z[Invalid type for group, must be one of List, Processgroup, DeviceMesh or (DeviceMesh, int).)rr0r2extendrb ValueErrorr ProcessGroupget_process_group_ranksr-_get_group_tagrndimrF get_grouptupler) rr rr nested_listrr7rspgdmeshras rrrs , , , & & & &      %- eAh % % &**511KGJ! % %r"""## c"gg(=(=$YzYYPSTVPWPWYY!WW  %#l5))GWJJ E4, - - .u55\\ /T(// E: & & :?? \ __  .r22\\ ,T(,, E5 ! ! JJ!OO58Z00 58S)) !HE(C%%B2266GWJ0,R00CCPQQ Q i    * %%rc~t|tjr|St|trt t j|}|St|tr]|jdkrtdtj j r%tj j|dS|jdSt|t"rt%|dkrt|dtrnt|dt&rS|d}|d}tj j r%tj j||S|j|St)dt+|dt+|dft|t,r]t/st1jdt4dt jt t,t&||St)d t+|d |) zI Given group in RANK_TYPES, return a ProcessGroup or group name. r r$rr z?Invalid tuple for group must be (DeviceMesh, int). Instead got zThe combination of ranks + tag as process group identifier has been deprecated. Please switch to using ProcessGroup, DeviceMesh, or group name instead.rzUnsupported group type: z, )r0rr'rrr- GroupNamerr*rFconfigcompile_on_one_rankrr_dtensormesh_get_process_group_dim_group_namesr,rbrr&ror2rrOrPrQ$_resolve_group_name_by_ranks_and_tag)rr rr0ras rr"r"s.%*++0L E3  .L $.%00  E: & &&L :?? \  ; * G9%< 5# & &E% ~e$$$rz'_c10d_functional::_wrap_tensor_autogradrz(Tensor input) -> Tensor) mutates_argsrzrc t|S)ao Custom op that allows autograd to propagate from a normal Tensor to an AsyncCollectiveTensor. This is the low-level implementation. Users should call _maybe_wrap_tensor directly. Args: input: Input tensor to wrap in AsyncCollectiveTensor Returns: AsyncCollectiveTensor wrapping the input (or wait_tensor result if tracing) )r1rs r_wrap_tensor_autogradr?-s$ ! ' ''rc*tj|S)z0 Meta kernel for _wrap_tensor_autograd. r empty_liker>s r_rCBs  E " ""rc|S)a0 Backward for _wrap_tensor_autograd: identity (no-op). The wrapping is just for async optimization, gradients flow through unchanged. Args: ctx: Context object (unused) grad_output: Gradient from downstream operations Returns: Gradient unchanged (identity) rrs r_wrap_tensor_autograd_backwardrEJs  rcdS)z Setup context for _wrap_tensor_autograd backward. Args: ctx: Context object to save state for backward (nothing to save) inputs: Tuple of (input,) output: Output from forward pass Nrrs r#_wrap_tensor_autograd_setup_contextrGZs  FrctrdStjtjjjdStjtjjjrdStduSr) rrrp_get_dispatch_mode_TorchDispatchModeKeyFAKE&_dispatch_tls_is_dispatch_key_includedrsPythonDispatcherrrrr_are_we_tracingrNlsu!!t x""58#A#FGGSt x66 -t   4 ''rcZtrt|St|Sr)rNrr?rs rr%r%zs,!4    & &&rTvaluec#<Ktjj} tjj|dVtjj|dS#tjj|wxYw)aC Context manager to temporarily set whether inflight collectives are allowed as torch.compile graph inputs. Common use case is when the collective is issued in eager (with `async_op=True`) but waited in compiled region: ``` def all_reduce_eager(x): y = x * x req = dist.all_reduce(y, op=dist.ReduceOp.SUM, async_op=True) return y @torch.compile(fullgraph=True) def all_reduce_wait_compiled(y): torch.ops.c10d_functional.wait_tensor(y) return y * y x = torch.ones(1280, 1280, device="cuda") + self.rank # the context manager ensures that `wait_tensor(y)` will wait on the correct work object with allow_inflight_collective_as_graph_input_ctx(): y = all_reduce_eager(x) z = all_reduce_wait_compiled(y) ``` With this context manager, when a collective is called, under the hood the work object of the collective will be registered in the work registry, and the wait_tensor() in compiled region called on the output tensor of the collective will wait on the correct work object. N)rrp_distributed_c10d)_allow_inflight_collective_as_graph_input-_set_allow_inflight_collective_as_graph_input)rPpreviouss r,allow_inflight_collective_as_graph_input_ctxrVs8x)SSUUH  "PPQVWWW  "PP      "PP     s (A55&Bct|}t|dkr||n|dxx|zcc<||}|SNr)r2rErbappend new_empty)rr7out_size out_tensors r_make_all_gather_out_tensorr]sjEJJLL!!H 8}} #### z! **J rc fd|DS)Nc0g|]}t|Srr])rrr7s rrz:_all_gather_into_tensor_coalesced_meta..s$ E E E1 ': 6 6 E E Err)rr rr7s `r&_all_gather_into_tensor_coalesced_metaras E E E E E E EErc*tj|SrrArrs r_broadcast_metard  D ! !!rcBtj|tjS)N) memory_format)rrBcontiguous_formatrcs r_all_reduce_metaris  D0G H H HHrc*tj|SrrArcs r_wait_tensor_metarkrercDtjd|j|jS)Nrrrremptyrrrcs r _isend_metarps ;q 4; ? ? ??rc*tj|SrrArcs r _irecv_metarrrerc6dt||DS)Ncfg|].\}}|dkr|n tjd|j|j/S)irecvrrmrnroprs rrz'_batch_p2p_ops_meta..sM    B7]] AQWQX N N N   r)rd)op_list peer_listtag_listtensorsrs r_batch_p2p_ops_metar|s-  '**   rc"t||Srr`)shardr rr7s r_all_gather_into_tensor_metar &uj 9 99rct|}|dxx|zcc<||SrXr2rErZ)rrr rr7r[s r_reduce_scatter_tensor_metars?EJJLL!!H QKKKJKKK ??8 $ $$rcd|DS)Nc6g|]}tj|SrrA)rrs rrz._all_reduce_coalesced_meta..s# . . .AE Q   . . .rrrcs r_all_reduce_coalesced_metars . . . . ..rc|Srrinprs r_all_reduce__metar Jrc|Srrrs r_broadcast__metarrrc|Srr)r_rs r_all_reduce_coalesced__metars Mrc,fdfd|DS)Nct|}|dxxzcc<||}|SrXr)rr[r\r7s r mk_out_tensorz<_reduce_scatter_tensor_coalesced_meta..mk_out_tensorsE %% " __X.. rc&g|] }|Srr)rrrs rrz9_reduce_scatter_tensor_coalesced_meta..s# - - -MM!   - - -rr)r_r&r rr7rs `@r%_reduce_scatter_tensor_coalesced_metars9 . - - -f - - --rc|'||S|D]}tj|dkt |}t ||d<||SrX)rZrEr_checkr2r)rr}r~rrsr[s r_all_to_all_single_metars!uzz||,,,# ! !A La  %%,-- x(((rc"t||Srr`)rr7rrs r'_all_gather_into_tensor_out_native_metarrrc"t||Srr`)rr7rs r#_all_gather_into_tensor_native_metar rrc$fd|DS)Nc2g|]}t|Sr)r)rrrr7s rrzA_all_gather_into_tensor_coalesced_native_meta..s5     ,E:zJJ   rr)r_r7rs ``r-_all_gather_into_tensor_coalesced_native_metars4        rct|}|dxx|zcc<||SrXr)rrr7rr3s r"_reduce_scatter_tensor_native_metars?   E !HHHHHH ==  rct|}|dxx|zcc<||SrXr)rrr7rrr3s r&_reduce_scatter_tensor_out_native_metarsA   E !HHHHHH ==  rc(fd|DS)Nc4g|]}t|Sr)r)rrrr7rs rrz@_reduce_scatter_tensor_coalesced_native_meta..'s7     +3 :zRR   rr)r_rr7rs ```r,_reduce_scatter_tensor_coalesced_native_metar$s:         rrIMPLr(Meta all_reduce_rYall_reduce_coalesced_risendru batch_p2p_opsall_gather_into_tensor_outr/r]rGreduce_scatter_tensor_outrgrr# broadcast__c10d_functional_autogradc10d_functionalDEF) 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) -> Tensorz>isend(Tensor self, int dst, int tag, str group_name) -> Tensorz>irecv(Tensor self, int src, int tag, str group_name) -> Tensorzkbatch_p2p_ops(str[] op_list, int[] peer_list, int[] tag_list, Tensor[] tensors, str group_name) -> Tensor[]zTall_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) -> Tensor(rC)tagsrtF output_tensor input_tensorasync_opc|rtd|ptjj}|td|t ||||SN@Can't remap async version of inplace op to functional collectivegroup cannot be None)rFrrWORLDcopy_r;)rrrrr r*s rall_gather_tensor_inplacersj  N     %TZ%E }3444   0z5RUVV W WWrrrrwc |rtd|ptjj}|td|t |||||Sr)rFrrrrrH)rrrwrrr?r s rreduce_scatter_tensor_inplacersi  N     %TZ%E }3444 <<-eReSQQ R RRravgproductminmaxbandborbxorrc|rtd|ptjj}|td|t ||||Sr)rFrrrrr()rrwrrr s rall_reduce_inplacersf  N     %TZ%E }3444 << 62uc:: ; ;;rc |rtd|ptjj}|td|t |||||Sr)rFrrrrr)rrr}r~rrr s rall_to_all_inplacersx  N     %TZ%E }3444 <<          rr[cF|rtddkr*tfd|Dstd|ptjj}|tdt d||}g}d}|D]d}|dk} | rdn|d} | r||n |||| z} || || z }et||D]\} } | | |S)Nrrc3pK|]0}|ddkV1dS)rN)rE)rrrs rrz%all_gather_inplace..s<$V$VQQVVAYY&++a..%@$V$V$V$V$V$Vrz7Remapping variable size all_gather is not yet supportedrr ) rFrarrrrr;rErYrdr)r[rrrr r output_splitsoffsetr is_scalart_offsetrrrs ` rall_gather_inplacersW  N   zz||q$V$V$V$V+$V$V$V!V!VVWWW  %TZ%E }3444 vq% 5 5FM F EEGGqL !011qvvayy )QfVnnvfv?P6P/QS!!!( ]33S # rr#r group_dstcR|tjj}|td|dkr'|t dt j||}n|}t|}tj j ||||}tr|St|S)Nrr#z@Cannot specify both 'dst' and 'group_dst' args as per eager impl)rrrrFr&r-get_global_rankr;rrrrrNr%)rrr rr global_dstrs r isend_inplacers }   }3444B ?R )%;;  $U++J Y ' - -fj#z R RF f % %%r group_srcc2|tjj}|td|dkr'|t dt j||}n|}t|}tj j ||||}t|S)Nrr#z@Cannot specify both 'src' and 'group_src' args as per eager impl) rrrrFr&r-rr;rrrrur%)rrr rr global_srcrs r irecv_inplacer1s }   }3444B ?R )%;;  $U++J Y ' - -fj#z R RF f % %%rrxryrzr{rctjstd||dkrtj}t |}t |tr|n|j}tj j |||||}trdt||DStt!t"|S)Nz%torch.distributed must be initializedrcBg|]\}}|dkrt|n|S)ru)r%rvs rrz)batch_p2p_ops_inplace..ZsB   A&(7]] q ! ! !   r)ris_initializedrFr-_get_default_groupr"r0rrrrrrrNrdr2rZr%)rxryrzr{rresolveds rbatch_p2p_ops_inplacerIs   FDEEEZ2--,.. j))H'#66OHrHrKrRrTrGrWrYr^rir]rgr|rrrrlibraryregister_autogradrrrrrrrrrrrrrrrr1rr"r; custom_opr? register_fakerCrErGrrNr%contextmanagerrVr]rardrirkrprrr|rrrrrrrrrrrrrrLibrarylib_implimpllib_impl_autogradfxnodehas_side_effectrrrrrur legacy_liblegacy_lib_implops_defsmodulesr my_moduleop_defindexop_namegetattr backend_impldefineTagpt2_compliant_tagrrReduceOpSUMAVGPRODUCTMINMAXBANDBORBXORREDUCE_OP_TO_STRrrrrrrr$rlegacy_all_gather_baserlegacy_reduce_scatter_baserlegacy_all_gatherr/legacy_allgatherlegacy_allgather_singlelegacy_allreducelegacy_all_to_all_singlerlegacy_batch_p2p_ops legacy_irecv legacy_isendlegacy_reducescatter_singlelegacy_reducescatterrrrrrrrrtraceable_collective_remapsrrrr:s ++++++++++ 111111111......444444======::::::25555555222111111112 NNNNNNN   HMn   #J  I 49o  $c )* +  n  .-----:::: & &EL &s &: &C & & & & &&U\&S&&#&&&&8 '' ,'' ' '  \ ''''\ ;; ,;; ; ;;;;2 '' ,'''  ' ''''^ JJ ,JJJ  J JJJJ@ ; ; , ; ;  ; ;  \ ; ; ; ;( D D , D D  D D D D D* JJ ,JJJ  J JJJJ, SS ,SSS  S SSSS$LN66 u| 6(+64>6EH6 %,6666:=?66 u| 6%/6696 %,6666D -6-6  -6-6c-6  -6 -6  %, -6-6-6-6h=? 9 9 u|  9%/ 969 9 %, 9 9 9 9$ VV  VVcV  V V  %, VVVV& V V V* 2&2& ,2&S D(2&Cy4'2&  2& 2&  \ 2&2&2&2&t V V , VS D( VCy4' V  V V  \ V V V V& 5<        #+ +%,++++8 & & & "* +el++++8     .#6 1U\1111D & & & -"5 111118 . . . )1  7#6   6"5   21 =T%,5G====> & & & ,!4 =elAS====8     8-@ CU\@RCCCCD & & & 7,?  VV ,V #YV V V  \ VVVV>q#q#q#q#q#ELq#q#q#h L&L&L&#L&uS$s)S=P7QL&L&L&L&`#%8L8L 8L8L '8L8L8L8Lv  z  T^    - % ( (%, ( ( (  ( $#U\#el###%$# U\        ''"5( ( ( ( ( (' ''''  # # # # # # LFFF """III"""@@@""":::%%% ///... ) ) )::::::       = !3V < < l,f555 m.777 $&@&III %'BFKKK m.777 g{F+++ g{F+++ o2F;;;  "I6 &(KVTTT &1   %'I6RRR !G %0   !#:FCCC k?F333 l,f555 M))*EvNNA6?*,CVLLL ei8DLMMM ei8DEEE ei8>FGGG ei8>??? ei8>FGGG ei8>??? ei8FNOOO ei8FGGG ] " "#4e < < -''(96BB    K ! MMFQc***+G7<W77Lf59#>???,0KLLLL  XX<X,X X X  XXXX. SS LS <S S  S  S SSSS, MuMuM9MuMuMMuM   << L< < < <<<<,   L < @   ""el#" L" """"P&* && L& & &  t # &  &&&&<&* && L& & &  t # &  &&&&02 #Y2Cy23i2%,  2  22220  t~ -  '                                0/// 333((( ((((((*** *** 222)01!8)9+ 7///1 s$A AA A''BB