o j9:jG@sddlZddlZddlmZddlmZddlmZmZddddZ ddZ ej fd d Z dej fd d Z dej fd dZejej fddZejej fddZej fddZej fddZej fddZddej fddZejej fddZGdddeZGdd d eZGd!d"d"eZGd#d$d$eZGd%d&d&eZGd'd(d(eZGd)d*d*eZGd+d,d,eZGd-d.d.eZGd/d0d0eZ dS)1N)Function)groupReduceOp suggestioncCs(d|d}|r|d|d7}t|)N torch.distributed.nn.functional.z& is not supported under torch.compile.z Use instead.) RuntimeError)namermsgr f/home/nk/hobo-godmode/plappi-mvp/.venv/lib/python3.10/site-packages/torch/distributed/nn/functional.py_not_supported_under_compiles rcCs"tjd|d|dtdddS)Nrz is deprecated, use r)category stacklevel)warningswarn FutureWarning)r rr r r _deprecateds rcC.tjr tdddtddt|||S)a Broadcasts the tensor to the whole group. ``tensor`` must have the same number of elements in all processes participating in the collective. Arguments: tensor (Tensor): Data to be sent if ``src`` is the rank of current process. src (int): Source rank. group (ProcessGroup, optional): The process group to work on. Returns: Tensor: Received tensor from the broadcast op. broadcastz3torch.distributed._functional_collectives.broadcastr)torchcompiler is_compilingrr _Broadcastapply)tensorsrcrr r r r s  rcCs tjr tdt|||S)aT Gathers a list of tensors in a single process. Arguments: tensor (Tensor): Input tensor. dst (int, optional): Destination rank (default is 0). group (ProcessGroup, optional): The process group to work on. Returns: tuple[Tensor]: List of appropriately-sized tensors with the gathered data. gather)rrrr_Gatherr)rdstrr r r r:s rcCs&tjr tdtj||g|RS)a Scatters a list of tensors to all processes in a group. Each process will receive exactly one tensor and store its data in the ``tensor`` argument. Arguments: tensors (list[Tensor]): List of tensors to scatter on the source rank. Receivers must pass ``None`. src (int, optional): Source rank (default is 0). group (ProcessGroup, optional): The process group to work on. Returns: Tensor: Output tensor from the scatter operation. scatter)rrrr_Scatterr)tensorsrrr r r r"Ks r"cCs"tjr tdt||||S)a Reduces the tensor data across all machines. Only the process with rank ``dst`` is going to receive the final result. Arguments: tensor (Tensor): Input of the collective. dst (int): Destination rank. op (optional): One of the values from ``torch.distributed.ReduceOp`` enum. Specifies an operation used for element-wise reductions. group (ProcessGroup, optional): The process group to work on. Returns: Tensor: Output of the collective. reduce)rrrr_Reducer)rr!oprr r r r%as r%cCs6tjr tdddtddtj|||g|RS)a Reduces, then scatters a list of tensors to all processes in a group. Arguments: output (Tensor): Output tensor. input_list (list[Tensor]): List of tensors to reduce and scatter. op (optional): One of the values from ``torch.distributed.ReduceOp`` enum. Specifies an operation used for element-wise reductions. group (ProcessGroup, optional): The process group to work on. Returns: Tensor: Output of the collective. reduce_scatterz?torch.distributed._functional_collectives.reduce_scatter_tensorr)rrrrr_Reduce_Scatterr)output input_listr'rr r r r(xs r(cCs,tjr tdddtddt||S)a Gathers tensors from the whole group in a list. Arguments: tensor (Tensor): Tensor to be broadcast from current process. group (ProcessGroup, optional): The process group to work on. Returns: tuple([Tensor]): Output of the collective. all_gatherz;torch.distributed._functional_collectives.all_gather_tensorr)rrrrr _AllGatherr)rrr r r r,s  r,cCs tjr tdt|||S)a Single tensor all gather. Gathers a single tensor from all ranks, and puts them in a single output tensor. Args: output_tensor (Tensor): Output tensor. It should contain correctly-sized tensors to be used for output of the collective. input_tensor (Tensor): Tensor to be broadcast from current process. group (ProcessGroup, optional): The process group to work on. If None, the default process group will be used. Examples: >>> # All tensors below are of torch.int64 dtype. >>> # We have 2 process groups, 2 ranks. >>> # xdoctest: +SKIP("incorrect want text") >>> output_tensor = torch.zeros(2, dtype=torch.int64) >>> output_tensor [tensor([0, 0])] # Rank 0 and 1 >>> tensor = torch.arange(1, dtype=torch.int64) + 1 + rank >>> tensor tensor([1]) # Rank 0 tensor([2]) # Rank 1 >>> dist.all_gather_base(output_tensor, tensor) >>> output_tensor tensor([1,2]) # Rank 0 tensor([1,2]) # Rank 1 .. warning:: `_all_gather_base` is experimental and subject to change. It is the caller's responsibility to ensure the output_tensor is correctly sized. _all_gather_base)rrrr_AllGatherBaser) output_tensor input_tensorrr r r r.s !r.cCs&tjr tdtj||g|RS)a Each process scatters list of input tensors to all processes in a group and return gathered list of tensors in output list. Arguments: output_tensor_list (list[Tensor]): list of tensors to gather one per rank. input_tensor_list (list[Tensor]): List of tensors to scatter one per rank. group (ProcessGroup, optional): The process group to work on. Returns: tuple([Tensor]): Output of the collective. all_to_all)rrrr _AlltoAllr)output_tensor_listinput_tensor_listrr r r r2s r2cCs2tjr tdddtddt|||||S)a 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. Arguments: output (Tensor): Gathered concatenated output tensor. input (Tensor): Input tensor to scatter. output_split_sizes: (list[Int], optional): Output split sizes for dim 0 if specified None or empty, dim 0 of ``output`` tensor must divide equally by ``world_size``. input_split_sizes: (list[Int], optional): Input split sizes for dim 0 if specified None or empty, dim 0 of ``input`` tensor must divide equally by ``world_size``. Returns: Tensor: Output of the collective. all_to_all_singlez;torch.distributed._functional_collectives.all_to_all_singler)rrrrr_AlltoAllSingler)r*inputoutput_split_sizesinput_split_sizesrr r r r6s  r6cCr)a& Reduces the tensor data across all machines in such a way that all get the final result. After the call the returned tensor is going to be bitwise identical in all processes. Arguments: tensor (Tensor): Input of the collective. op (optional): One of the values from ``torch.distributed.ReduceOp`` enum. Specifies an operation used for element-wise reductions. group (ProcessGroup, optional): The process group to work on. Returns: Tensor: Output of the collective all_reducez4torch.distributed._functional_collectives.all_reducer)rrrrr _AllReducer)rr'rr r r r; s  r;c@$eZdZeddZeddZdS)rcCs6||_||_tj|d|_|}tj|||d|SNr)rrdistget_rankrankcloner)ctxrrrr r r forward's z_Broadcast.forwardcCs4t|jtj|j|}|j|jkr|dd|fSN)r&rrrSUMrrBzero_)rD grad_outputgxr r r backward3s  z_Broadcast.backwardN__name__ __module__ __qualname__ staticmethodrErKr r r r r&s   rc@r=)r csv||_||_fddttj|dD}tj|d|kr.tj|||dt|Stjd||dt|S)Ncg|]}tqSr )r zeros_like.0irr r F z#_Gather.forward..r?) r!rranger@get_world_size contiguousrArtuple)rDr!rr tensor_listr rVr rE=s z_Gather.forwardcGsdtj|j|jg|RfSNNN)r#rr!r)rD grad_outputsr r r rKQsz_Gather.backwardNrLr r r r r <s  r c@r=)r#cst||_||_tfddDsttd}tj|d|kr/tj|t ||d|Stj|d||d|S)Nc3s$|] }|dkVqdS)rNsizerTtr$r r \s"z#_Scatter.forward..rr?) rrallAssertionErrorrrRr@rAr"list)rDrrr$r*r rer rEWsz_Scatter.forwardcCsdt|j|j|Sr^)r rrrrDrIr r r rKesz_Scatter.backwardNrLr r r r r#Vs   r#c@r=)r&cCs*||_||_|}tj||||d|S)Nr'r)rrrCr@r%)rDrr'rrr r r rEls z_Reduce.forwardcCdt|j|j|fSN)NNN)rrrrrjr r r rKuz_Reduce.backwardNrLr r r r r&k  r&c@r=)r)cGs:||_|}tdd|D}tj|t|||d|S)Ncs|]}|VqdSrFr[rcr r r rfz*_Reduce_Scatter.forward..rk)rr[r\r@r(ri)rDr'rrr5r r r rE|s z_Reduce_Scatter.forwardcCsdt|j|Srm)r-rrrjr r r rKsz_Reduce_Scatter.backwardNrLr r r r r){s  r)c@r=)r-csD||_fddttj|dD}tj||dt|S)NcrQr r empty_like)rT_rVr r rWrXz&_AllGather.forward..r?)r[rrYr@rZr,r\)rDrrout_tensor_listr rVr rEs z_AllGather.forwardcGstj|jdtjjtjjfvr.tj|jd}t||}t j t j |j|g|R}d|fSdd|D}t j |j|g|R}tjt|dd}d|fS)Nr?cSsg|]}t|qSr rs)rTrr r r rWsz'_AllGather.backward..r)dim)r@ get_backendrBackendNCCLXCCLrArrtr)rrrGr3sumstack)rDr`rBrJr]gxsr r r rKsz_AllGather.backwardNrLr r r r r-s   r-c@r=)r/cCs||_tj|||d|Sr>)rr@r.r[)rDr0r1rr r r rEsz_AllGatherBase.forwardcCstj|jdtjjtjjfvrPtj|jd}t|}|d|dkr.t d|d||dtj|jd|d<t j ||j |j d}t||tj|jnt dd|dfS)Nr?rzTensor with dimensions: z8 does not have first dimension divisible by world_size: devicedtypezBackend not supported!)r@rxrryrzr{rZrirbr remptyrr_reduce_scatter_baserrG)rDrI world_sizeout_sizerJr r r rKs    z_AllGatherBase.backwardNrLr r r r r/s  r/c@r=)r3cs||_fddttj|dD|_tj|d}tddDtj|dtjj urPttj|dD]}d}||kr@t }tj |||||dq4t|Stj |t |dt|S)Ncsg|]}|qSr rarSrer r rWs z%_AlltoAll.forward..r?csrprFrqrcr r r rfrrz$_AlltoAll.forward..) rrYr@rZinput_tensor_size_listrAr\rxryGLOOrir"r2)rDrrvr$my_rankrUto_sendr rer rEs&  z_AlltoAll.forwardcs.fdd|jD}dtj|j|gRS)Ncs(g|]}tj|djdjdqS)rr)rrrr)rTrbr`r r rWs z&_AlltoAll.backward..r_)rr3rr)rDr`r]r rr rKs z_AlltoAll.backwardNrLr r r r r3s  r3c@r=)r7cCs4||_||_||_||_tj|||||d|S)N)r9r:r)rrb input_sizer9r:r@r6)rDrr*r9r:r8r r r rEs z_AlltoAllSingle.forwardc Cs8tj|j|j|jd}dt|j||j|j | fS)Nr)NNNN) rrrrrr7rrr9r:r[)rDrIrr r r rKs z_AlltoAllSingle.backwardNrLr r r r r7s  r7c@r=)r<cCs.||_||_|jtjd}tj|||d|S)N) memory_formatrk)rr'rCrcontiguous_formatr@r;)rDr'rrr r r rEs z_AllReduce.forwardcCrlr^)r<rr'rrjr r r rKrnz_AllReduce.backwardNrLr r r r r< ror<)!rrtorch.distributed distributedr@torch.autogradrrrrrWORLDrrr"rGr%r(r,r.r2r6r;rr r#r&r)r-r/r3r7r<r r r r s<    & ($"