o ۷in, @sddlmZddlZddlmZddlmZddlmZgdZ dd ej d e d e d e d ej f ddZ GdddejjZ dd ej d e d e d e d ej f ddZGdddejjZGdddZdS))AnyN)Tensor)current_omni_platform) all_to_all_4D all_to_all_5D SeqAllToAll4D SeqAllToAll5DRingCommFinput scatter_idx gather_idxuse_syncreturncCs|dksJd|d|jt|}|dkrp|dkrp|j\}}}} ||} ||} ||||| | dd} t| } |dkrVtj | | |d|rUt n| } | | || | } | dd|| | | } | S|dkr|dkr|j\}} } } | |}| |}t|}||||| | dddd|| ||| } t| } |dkrtj | | |d|rt n| } | |||| } | dd|||| } | St d ) a all-to-all for QKV Args: input (torch.tensor): a tensor sharded along dim scatter dim scatter_idx (int): default 1 gather_idx (int): default 2 group (torch.distributed.ProcessGroup): torch process group use_sync (bool): whether to synchronize after all-to-all Returns: torch.tensor: resharded tensor (bs, seqlen/P, hc, hs) zinput must be 4D tensor, got and shape r r rgroupz8scatter_idx must be 1 or 2 and gather_idx must be 1 or 2 dimshapedistget_world_sizereshape transpose contiguoustorch empty_likeall_to_all_singler synchronize RuntimeError)r r rrrseq_world_sizebs shard_seqlenhchsseqlenshard_hcinput_toutputr,Z/home/ubuntu/vllm_env/lib/python3.10/site-packages/vllm_omni/diffusion/distributed/comm.pyrsN&    rc@s<eZdZe d dedejdededede defd d Z d S) rFctxrr r rrrcC*||_||_||_||_t|||||dSN)rr)rr rrrr.rr r rrr,r,r-forwardhs  zSeqAllToAll4D.forwardN)F __name__ __module__ __qualname__ staticmethodrr ProcessGrouprintboolr2r,r,r,r-rgs$rrcCs|dksJd|d|jt|}|dkr|dkr|j\}}}} } |dks.J||} | |} |||d|| | dd} t| }|dkr^tj || |d|r]t n| }|| d|| | }|dddd}||| d| | S|dkr|dkr|j\}} }} } | |} | |}t|}||||d| | dd dd|| |d|| } t| }|dkrtj || |d|rt n| }|| |d|| }|dd}|||d| | St d ) a all-to-all for QKV forward (bs, seqlen/N, 3, hc, hs) -> (bs, seqlen, 3, hc/N, hs) Args: input (torch.tensor): a tensor sharded along dim scatter dim scatter_idx (int): default 1 gather_idx (int): default 2 group (torch.distributed.ProcessGroup): torch process group use_sync (bool): whether to synchronize after all-to-all Returns: torch.tensor: resharded tensor (bs, seqlen/P, 3, hc, hs) zinput must be 5D tensor, got rrr rrr rz8scatter_idx must be 1 or 3 and gather_idx must be 1 or 3r)r r rrrr#r$r%t_cntr&r'r(r)r*r+_r,r,r-rxsP&      rc@s@eZdZe   ddedejdededed e d efd d Z d S)rrr Fr.rr r rrrcCr/r0)rr rrrr1r,r,r-r2s  zSeqAllToAll5D.forwardN)rr Fr3r,r,r,r-rs(rc@sReZdZdZdejfddZddejdejdBdejfd d Z d d Z d dZ dS)r z@Ring communication utility for Ring Attention P2P communication. process_groupcCs||_g|_t|j|_t|j|_d|_|jd|j|_|jd|j|_ |durAt |j|j|_t |j|j |_ dSdS)Nr ) _process_group_opsrget_rankrankr world_size_reqs send_rank recv_rankget_global_rank)selfr>r,r,r-__init__szRingComm.__init__Nto_send recv_tensorrcCs|s|}|durtj|tjd}n |}|s|}tjtj||j|j d}tjtj ||j |j d}|j ||j ||S)N) memory_formatr) is_contiguousrrrcontiguous_formatrP2POpisendrEr?irecvrFr@append)rHrJrKressend_oprecv_opr,r,r- send_recvs  zRingComm.send_recvcCs$|jdur tdt|j|_dS)Nzcommit called twice)rDr"rbatch_isend_irecvr@)rHr,r,r-commit s zRingComm.commitcCs6|jdur td|jD]}|q d|_g|_dS)Nzwait called before commit)rDr"waitr@)rHreqr,r,r-rYs    z RingComm.wait)N) r4r5r6__doc__rr8rIrrrVrXrYr,r,r,r-r s " r )r r NF)rr NF)typingrrtorch.distributed distributedrrvllm_omni.platformsr__all__tensorr9r:rautogradFunctionrrrr r,r,r,r-sB     W Z