U *-e:@stddlZddlmZddlmZddlmZmZejfddZ dejfddZ dejfdd Z ej ejfd d Z ej ejfd d ZejfddZejfddZejfddZddejfddZej ejfddZGdddeZGdddeZGdddeZGdddeZGd d!d!eZGd"d#d#eZGd$d%d%eZGd&d'd'eZGd(d)d)eZGd*d+d+eZdS),N)Function)groupReduceOpcCst|||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. ) _Broadcastapply)tensorsrcrr `/var/www/html/Darija-Ai-Train/env/lib/python3.8/site-packages/torch/distributed/nn/functional.py broadcast sr cCst|||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. )_Gatherr)rdstrr r r gathers rcCstj||f|S)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. )_Scatterr)tensorsrrr r r scatter,srcCst||||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. )_Reducer)rr oprr r r reduce@srcCstj|||f|S)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_Scatterr)outputZ input_listrrr r r reduce_scatterUsrcCs t||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. ) _AllGatherr)rrr r r all_gatherhs rcCst|||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. )_AllGatherBaser) output_tensor input_tensorrr r r _all_gather_basevs!rcCstj||f|S)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. ) _AlltoAllr)Zoutput_tensor_listinput_tensor_listrr r r all_to_allsr cCst|||||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. )_AlltoAllSingler)rinputoutput_split_sizesinput_split_sizesrr r r all_to_all_singlesr%cCst|||S)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 ) _AllReducer)rrrr r r all_reducesr'c@s$eZdZeddZeddZdS)rcCs2||_||_t|_|}tj|||d|SNr)rrdistget_rankrankcloner )ctxrrrr r r forwards  z_Broadcast.forwardcCs4t|jtj|j|}|j|jkr*|dd|fSN)rrrrSUMrr,Zzero_)r. grad_outputgxr r r backwards z_Broadcast.backwardN__name__ __module__ __qualname__ staticmethodr/r4r r r r rs rc@s$eZdZeddZeddZdS)r csp||_||_fddttj|dD}tj|d|krVtj|||dntjd||dt|S)Ncsg|]}tqSr )torch zeros_like.0irr r sz#_Gather.forward..r)) r rranger*get_world_size contiguousr+rtuple)r.r rr tensor_listr r?r r/s z_Gather.forwardcGsdtj|j|jf|fSNNN)rrr r)r. grad_outputsr r r r4sz_Gather.backwardNr5r r r r r s r c@s$eZdZeddZeddZdS)rcsr||_||_tfddDs&ttd}tj|d|kr\tj|t ||dntj|d||d|S)Nc3s"|]}|dkVqdS)rNsizer=trr r sz#_Scatter.forward..rr)) rrallAssertionErrorr:r;r*r+rlist)r.rrrrr rMr r/sz_Scatter.forwardcCsdt|j|j|SrF)r rrrr.r2r r r r4sz_Scatter.backwardNr5r r r r r s rc@s$eZdZeddZeddZdS)rcCs*||_||_|}tj||||d|SNrr)rrr-r*r)r.rrrrr r r r/ s z_Reduce.forwardcCsdt|j|j|fSN)NNN)rrrrrRr r r r4(sz_Reduce.backwardNr5r r r r rs rc@s$eZdZeddZeddZdS)rcGs:||_|}tdd|D}tj|t|||d|S)Ncss|]}|VqdSr0rCrKr r r rN3sz*_Reduce_Scatter.forward..rT)rrCrDr*rrQ)r.rrrrr r r r/.s z_Reduce_Scatter.forwardcCsdt|j|SrU)rrrrRr r r r47sz_Reduce_Scatter.backwardNr5r r r r r-s rc@s$eZdZeddZeddZdS)rcsD||_fddttj|dD}tj||dt|S)Ncsg|]}tqSr r: empty_like)r=_r?r r r@Csz&_AllGather.forward..r))rCrrAr*rBrrD)r.rrout_tensor_listr r?r r/=s z_AllGather.forwardcGstj|jdtjjkrFt}t||}tj t j |j|f|n6dd|D}t j |j|f|}tj t|dd}d|fS)Nr)cSsg|]}t|qSr rW)r=rr r r r@Ssz'_AllGather.backward..r)dim)r* get_backendrBackendNCCLr+r:rXrrrr1rsumstack)r.rHr,r3rEZgxsr r r r4Jsz_AllGather.backwardNr5r r r r r<s rc@s$eZdZeddZeddZdS)rcCs||_tj|||d|Sr()rr*rrC)r.rrrr r r r/Ysz_AllGatherBase.forwardcCstj|jdtjjkrtj|jd}t|}|d|dkrTtd|d||dtj|jd|d<t j ||j |j d}t ||tj|jntdd|dfS)Nr)rzTensor with dimensions: z8 does not have first dimension divisible by world_size: devicedtypezBackend not supported!)r*r\rr]r^rBrQrJ RuntimeErrorr:emptyrbrcZ_reduce_scatter_baserr1)r.r2Z world_sizeZout_sizer3r r r r4_s z_AllGatherBase.backwardNr5r r r r rXs rc@s$eZdZeddZeddZdS)rcs||_fddttj|dD|_tj|d}tddDtj|dtjj krttj|dD].}d}||krt }tj |||||dqhntj |t |dt|S)Ncsg|]}|qSr rIr<rMr r r@tsz%_AlltoAll.forward..r)css|]}|VqdSr0rVrKr r r rNxsz$_AlltoAll.forward..) rrAr*rBinput_tensor_size_listr+rDr\r]ZGLOOrQrr )r.rrZrZmy_rankr>to_sendr rMr r/qs$  z_AlltoAll.forwardcs,fdd|jD}dtj|j|fS)Ncs(g|] }tj|djdjdqS)rra)r:rerbrc)r=rJrHr r r@sz&_AlltoAll.backward..rG)rfrrr)r.rHrEr rhr r4s z_AlltoAll.backwardNr5r r r r rps rc@s$eZdZeddZeddZdS)r!cCs4||_||_||_||_tj|||||d|S)N)r#r$r)rrJ input_sizer#r$r*r%)r.rrr#r$r"r r r r/s z_AlltoAllSingle.forwardc Cs8tj|j|j|jd}dt|j||j|j | fS)Nra)NNNN) r:rerirbrcr!rrr#r$rC)r.r2rr r r r4sz_AlltoAllSingle.backwardNr5r r r r r!s r!c@s$eZdZeddZeddZdS)r&cCs(||_||_|}tj|||d|SrS)rrr-r*r')r.rrrr r r r/s z_AllReduce.forwardcCsdt|j|j|fSrF)r&rrrrRr r r r4sz_AllReduce.backwardNr5r r r r r&s r&)r:Ztorch.distributed distributedr*Ztorch.autogradrrrZWORLDr rrr1rrrrr r%r'rr rrrrrrr!r&r r r r s4  $ !