קgJ;BddlZddlmZddlmZddlmZmZejfdZ dejfdZ dejfdZ ej ejfdZ ej ejfdZejfd Zejfd Zejfd Zddejfd Zej ejfd ZGddeZGddeZGddeZGddeZGddeZGddeZGddeZGddeZGddeZGd d!eZdS)"N)Function)groupReduceOpc:t|||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)tensorsrcrs [/var/www/html/ai-engine/env/lib/python3.11/site-packages/torch/distributed/nn/functional.py broadcastr s"   C / //c:t|||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)r dstrs r gatherr s ==eV , ,,r c(tj||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. )_Scatterr)tensorsr rs r scatterr/s" >#u /w / / //r c<t||||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)r roprs r reducerCs$ ==b% 0 00r c*tj|||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_Scatterr)output input_listrrs r reduce_scatterrXs!  UF @Z @ @ @@r c8t||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)r rs r all_gatherr!ks   E6 * **r c:t|||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_tensorrs r _all_gather_baser&zsB    |U C CCr c(tj||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. ) _AlltoAllr)output_tensor_listinput_tensor_listrs r all_to_allr+s  ?5"4 I7H I I IIr c>t|||||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_sizesrs r all_to_all_singler1s)4   v)+rEr r rrsH\  \   r rc:eZdZedZedZdS)rcV||_||_fdttj|D}tj||krtj|||ntjd||t|S)Nc8g|]}tjSrK)torch zeros_like).0ir s r z#_Gather.forward..s1   )*E V $ $   r r8) rrranger9get_world_size contiguousr:rtuple)r=rrr tensor_lists ` r r>z_Gather.forwards     .3D4Ge4T4T4T.U.U   ""$$ =u % % % , , K S > > > > > Kc 7 7 7 7[!!!r cDdtj|j|jg|RfzSNNN)rrrr)r= grad_outputss r rEz_Gather.backward s(x~cgsyP<PPPRRRr NrFrKr r rrsM""\"$SS\SSSr rc:eZdZedZedZdS)rc@||_||_tfdDsJtjd}t j||kr&t j|t||nt j|d|||S)Nc3xK|]4}|dkV5dS)rNsize)rQtrs r z#_Scatter.forward..s>BBQ16688wqz000BBBBBBr rr8) r rallrOrPr9r:rlist)r=r rrrs ` r r>z_Scatter.forwards BBBB'BBBBBBBB!'!*-- =u % % % , , Lg5 A A A A A Ls% 8 8 8 8 r cTdt|j|j|zSrZ)rrr rr=rCs r rEz_Scatter.backwards!gmmCGSY LLLLr NrFrKr r rrsM  \ MM\MMMr rc:eZdZedZedZdS)rcz||_||_|}tj|||||SNrr)r rr<r9r)r=r rrr s r r>z_Reduce.forward"s;  FCBe4444 r cVdt|j|j|fzSN)NNN)rrr rrgs r rEz_Reduce.backward*s&!Z%5%5cgsy+%V%V$XXXr NrFrKr r rr!sM\YY\YYYr rc:eZdZedZedZdS)rc||_|}td|D}tj|t ||||S)Nc3>K|]}|VdSr@rVrQrbs r rcz*_Reduce_Scatter.forward..5s*!L!LQ!,,..!L!L!L!L!L!Lr rk)rrVrWr9rre)r=rrr r*s r r>z_Reduce_Scatter.forward0s` ""$$!!L!L:K!L!L!LLL FD):$;$;%PPPP r cHdt|j|zSrm)r rrrgs r rEz_Reduce_Scatter.backward9s!J$4$4SY $L$LLLr NrFrKr r rr/sM\MM\MMMr rc:eZdZedZedZdS)r c||_fdttj|D}tj||t |S)Nc8g|]}tjSrKrO empty_like)rQ_r s r rSz&_AllGather.forward..Es1   )*E V $ $   r r8)rVrrTr9rUr!rW)r=rr out_tensor_lists ` r r>z_AllGather.forward?s""$$     .3D4Ge4T4T4T.U.U    u====_%%%r ctj|jtjjurXtj|j}t j||}tj tj |j|g|R}nLd|D}tj |j|g|R}t j t j|d}d|fS)Nr8c6g|]}tj|SrKrw)rQr s r rSz'_AllGather.backward..Us#OOO5+F33OOOr r)dim)r9 get_backendrBackendNCCLr:rOrxrrrrAr(sumstack)r=r\r;rDrXgxss r rEz_AllGather.backwardLs  #) , , , 0A A A=sy111D!,t"455B &x|SYR\RRRBBPO,OOOK/#)[H<HHHC5;s++333Bbzr NrFrKr r r r >sH & &\ &  \   r r c:eZdZedZedZdS)r#cf||_tj||||Sr7)rr9r&rV)r=r$r%rs r r>z_AllGatherBase.forward\s3  m\-D-D-F-FeTTTTr ctj|jtjjurtj|j}t |}|d|zdkrtd|d||dtj|jz|d<tj ||j |j }tj ||tj|jntdd|dfS)Nr8rzTensor with dimensions: z8 does not have first dimension divisible by world_size: devicedtypezBackend not supported!)r9r~rrrrUrera RuntimeErrorrOemptyrr_reduce_scatter_baserrA)r=rC world_sizeout_sizerDs r rEz_AllGatherBase.backwardbs  #) , , , 0A A A,39===JK,,..//H{Z'1,,"VxVVISVV#1+)<39)M)M)MMHQK!3;;LB  %b+x|SY O O O O788 8b$r NrFrKr r r#r#[sH\   \   r r#c:eZdZedZedZdS)r(c<||_fdttj|D|_tj|}t dDtj|tjj ur[ttj|D]7}d}||krt}tj |||||8n$tj |t|t |S)NcDg|]}|SrKr`)rQrRrs r rSz%_AlltoAll.forward..zs4& & & "#GAJOO  & & & r r8c3>K|]}|VdSr@rqrrs r rcz$_AlltoAll.forward..~s*881 888888r ) rrTr9rUinput_tensor_size_listr:rWr~rGLOOrerr+)r=rrzrmy_rankrRto_sends ` r r>z_AlltoAll.forwardws: & & & & ',T-@u-M-M-M'N'N& & & "-e,,,8888888  % ( ( (DL,= = =4.U;;;<< J J<<"7mmG _Q/!5IIIII  J OW      _%%%r c`fd|jD}dtj|j|gRzS)Ncjg|]/}tj|djdj0S)rr)rOrrr)rQrar\s r rSz&_AlltoAll.backward..sQ    K\!_3<?;P      r r[)rr(rr)r=r\rXs ` r rEz_AlltoAll.backwardsS    2    iociT|TTTTTr NrFrKr r r(r(vsM&&\&,UU\UUUr r(c:eZdZedZedZdS)r-c||_||_||_||_t j||||||S)N)r/r0r)rra input_sizer/r0r9r1)r=rrr/r0r.s r r>z_AlltoAllSingle.forwardsX !2 2   1/      r c tj|j|j|j}dt |j||j|j | fzS)Nr)NNNN) rOrrrrr-rrr/r0rV)r=rCr s r rEz_AlltoAllSingle.backwardsq N;#5[=N   (  ! ! &%&&((   +   r NrFrKr r r-r-sH  \    \    r r-c:eZdZedZedZdS)r3cx||_||_|}tj||||Srj)rrr<r9r4)r=rrr s r r>z_AllReduce.forwards9  2U3333 r cVdt|j|j|fzSrZ)r3rrrrgs r rEz_AllReduce.backwards&z// ;OOQQQr NrFrKr r r3r3sM\RR\RRRr r3)rOtorch.distributed distributedr9torch.autogradrrrWORLDr rrrArrr!r&r+r1r4rrrrrr r#r(r-r3rKr r rs` ###### .-------"'0000( - - - -%+0000($ !D!D!D!DH=BK J J J J& + >#,ek////*        (SSSSShSSS2MMMMMxMMM$ Y Y Y Y Yh Y Y Y M M M M Mh M M M:     X   6 U U U U U U U UF     h   @ R R R R R R R R R Rr