From 42cb7d96bbcf48bda9a7909d31dd767d8a30bd09 Mon Sep 17 00:00:00 2001 From: Jinzhe Pan Date: Mon, 17 Mar 2025 08:31:45 +0000 Subject: [PATCH] feat: sp for wan --- .../xdit_context_parallel.cpython-310.pyc | Bin 0 -> 3754 bytes .../distributed/xdit_context_parallel.py | 127 ++++++++++++++++++ diffsynth/models/wan_video_dit.py | 2 + diffsynth/pipelines/wan_video.py | 19 ++- examples/wanvideo/wan_14b_text_to_video.py | 38 ++++-- 5 files changed, 175 insertions(+), 11 deletions(-) create mode 100644 diffsynth/distributed/__pycache__/xdit_context_parallel.cpython-310.pyc create mode 100644 diffsynth/distributed/xdit_context_parallel.py diff --git a/diffsynth/distributed/__pycache__/xdit_context_parallel.cpython-310.pyc b/diffsynth/distributed/__pycache__/xdit_context_parallel.cpython-310.pyc new file mode 100644 index 0000000000000000000000000000000000000000..844c90076f202adc0be23a9f0c4326ec2fff0bdb GIT binary patch literal 3754 zcmb7HTW=f372cV>aJeEWN|r54b{wGyTC-^6I!M!7;N}JuMGN)81==apLHpwN5A=b2DUkk@KJ~4@f@jrpzev>q`-7-_Xp?0%LDV25%XRV)PQTfi>+Rn2h?Lk_QqN1D7 ztoz#TMI}g9XuF70&9MYoRmV{+;_>sWh)EdQ$MB@cVxigbrFr;0>l;NAnfs_~@vi80CC`N@b0#kL@XhIdv*l@vGL>vMr{-*?2$Y5&<~|m9u&Z#ESG2bx5vwLh$J1^xjB=^Sv9yz=vM{nT z(o|Q7#n-EuiU%>wHiQ)wnd$~*q{!wHnd%zM)>L?LTl3;Ah&MX3%NA)Ry9G z6(jBixj~!PXo ztW!RPqf}mHZ$Qez-n6Lp2?zR|(?0E3!;S&8Q`@kx-wIa6nw2vneJz}!w*O?~-cGS| zGu;uLoj337Py=7B^nMD5kf57#8=m8s%vzi>*-`14rL^YZcv1d~b0M zzl_JSC}pl|XHKRYy(G@VzKqI=lA4|9ZQvc?FA_0eF6<3u?;y`&AYMAq-}w5mFzXYP zh#m=d`HOv;h6|=5V5GvT8lUhjYu2n9 zSdYP&oLQ@CRL!bIbuuZz3}17>2N!654WDQDT1!4~nl&!2H?W0Er=G3S9x(LQ2Am2z z*|RpVd=0l)6*b{Mw5DyGbG>5arBmWSd?+_gEto$^?ea&~@Ajr^s9jP2t~`{Q`X~4| zM^|TCbL%gg%tC8>bZvC4YFBHg+Nm|VQIRy4`gt8=My7q6I4X)*EBnS1AA7I~ z1Dp$={0=qAc6x-U%ZI`mdFUmSq7Cv#t4|y7hyK}e_I?lD=MPP$Nt-AIE%ztb+@CG% z%l3^AA*OqAnor6CsNjt=F-hbFL*xLy+?#%rMg4JfcNFETy78J1%Xb3IsC!y|96%> z+`37=t0bHn*iv?Obc?{?WiT~iQ~LxtxaC-{n&Dh!BiS<%zCKBp*k)pXi?X4Y^Y!Ww zuo;h|feej+4WOF1r)%+eK%5lN%HA;UpQzioigpM(wI`4BC>3g#L}2jN-*Y?%hb zcqROkXsDR*|5sGt;Z)$ebG(GcGoUmud(j3`A^_8fTT=2w2=1NQNDo|u>SYU|zp{ag z7&B>vjW+Q9yiMcog|~+_pu{<+TB+joKcaukovnw`qhcH z59EndPt!|%Qcj_NI1!z2QfC93a&XqI(6^&4tZ1Yany zPqiVDn21r|OC7dJhlu^@5+D*V6+0kp)`sd@yoE2l4qxP9oOw?JFW~9`Ovotr_x3vL zi#)_bb`A?yggtPOIL-3H98qr`_vs(WZZA`E_d@Qx$ViO082i3)kuSmjm#|kJ)A<(v zUX-~irOGCYF@n(EuS`;^oqJi97~?QUyKvK&*Y)3Bfae{8+=AT02R1d>25ZnflEJkZ2rtAdv*C2+~ZGpE%?0g!G8f^oyg<> literal 0 HcmV?d00001 diff --git a/diffsynth/distributed/xdit_context_parallel.py b/diffsynth/distributed/xdit_context_parallel.py new file mode 100644 index 0000000..a144e08 --- /dev/null +++ b/diffsynth/distributed/xdit_context_parallel.py @@ -0,0 +1,127 @@ +import torch +from typing import Optional +from einops import rearrange +from xfuser.core.distributed import (get_sequence_parallel_rank, + get_sequence_parallel_world_size, + get_sp_group) +from xfuser.core.long_ctx_attention import xFuserLongContextAttention + +def sinusoidal_embedding_1d(dim, position): + sinusoid = torch.outer(position.type(torch.float64), torch.pow( + 10000, -torch.arange(dim//2, dtype=torch.float64, device=position.device).div(dim//2))) + x = torch.cat([torch.cos(sinusoid), torch.sin(sinusoid)], dim=1) + return x.to(position.dtype) + +def pad_freqs(original_tensor, target_len): + seq_len, s1, s2 = original_tensor.shape + pad_size = target_len - seq_len + padding_tensor = torch.ones( + pad_size, + s1, + s2, + dtype=original_tensor.dtype, + device=original_tensor.device) + padded_tensor = torch.cat([original_tensor, padding_tensor], dim=0) + return padded_tensor + +def rope_apply(x, freqs, num_heads): + x = rearrange(x, "b s (n d) -> b s n d", n=num_heads) + s_per_rank = x.shape[1] + + x_out = torch.view_as_complex(x.to(torch.float64).reshape( + x.shape[0], x.shape[1], x.shape[2], -1, 2)) + + sp_size = get_sequence_parallel_world_size() + sp_rank = get_sequence_parallel_rank() + freqs = pad_freqs(freqs, s_per_rank * sp_size) + freqs_rank = freqs[(sp_rank * s_per_rank):((sp_rank + 1) * s_per_rank), :, :] + + x_out = torch.view_as_real(x_out * freqs_rank).flatten(2) + return x_out.to(x.dtype) + +def usp_dit_forward(self, + x: torch.Tensor, + timestep: torch.Tensor, + context: torch.Tensor, + clip_feature: Optional[torch.Tensor] = None, + y: Optional[torch.Tensor] = None, + use_gradient_checkpointing: bool = False, + use_gradient_checkpointing_offload: bool = False, + **kwargs, + ): + t = self.time_embedding( + sinusoidal_embedding_1d(self.freq_dim, timestep)) + t_mod = self.time_projection(t).unflatten(1, (6, self.dim)) + context = self.text_embedding(context) + + if self.has_image_input: + x = torch.cat([x, y], dim=1) # (b, c_x + c_y, f, h, w) + clip_embdding = self.img_emb(clip_feature) + context = torch.cat([clip_embdding, context], dim=1) + + x, (f, h, w) = self.patchify(x) + + freqs = torch.cat([ + self.freqs[0][:f].view(f, 1, 1, -1).expand(f, h, w, -1), + self.freqs[1][:h].view(1, h, 1, -1).expand(f, h, w, -1), + self.freqs[2][:w].view(1, 1, w, -1).expand(f, h, w, -1) + ], dim=-1).reshape(f * h * w, 1, -1).to(x.device) + + def create_custom_forward(module): + def custom_forward(*inputs): + return module(*inputs) + return custom_forward + + # Context Parallel + x = torch.chunk( + x, get_sequence_parallel_world_size(), + dim=1)[get_sequence_parallel_rank()] + + for block in self.blocks: + if self.training and use_gradient_checkpointing: + if use_gradient_checkpointing_offload: + with torch.autograd.graph.save_on_cpu(): + x = torch.utils.checkpoint.checkpoint( + create_custom_forward(block), + x, context, t_mod, freqs, + use_reentrant=False, + ) + else: + x = torch.utils.checkpoint.checkpoint( + create_custom_forward(block), + x, context, t_mod, freqs, + use_reentrant=False, + ) + else: + x = block(x, context, t_mod, freqs) + + x = self.head(x, t) + + # Context Parallel + x = get_sp_group().all_gather(x, dim=1) + + # unpatchify + x = self.unpatchify(x, (f, h, w)) + return x + + +def usp_attn_forward(self, x, freqs): + q = self.norm_q(self.q(x)) + k = self.norm_k(self.k(x)) + v = self.v(x) + + q = rope_apply(q, freqs, self.num_heads) + k = rope_apply(k, freqs, self.num_heads) + q = rearrange(q, "b s (n d) -> b s n d", n=self.num_heads) + k = rearrange(k, "b s (n d) -> b s n d", n=self.num_heads) + v = rearrange(v, "b s (n d) -> b s n d", n=self.num_heads) + + x = xFuserLongContextAttention()( + None, + query=q, + key=k, + value=v, + ) + x = x.flatten(2) + + return self.o(x) \ No newline at end of file diff --git a/diffsynth/models/wan_video_dit.py b/diffsynth/models/wan_video_dit.py index 32a79e3..618e7d8 100644 --- a/diffsynth/models/wan_video_dit.py +++ b/diffsynth/models/wan_video_dit.py @@ -90,6 +90,8 @@ def rope_apply(x, freqs, num_heads): x = rearrange(x, "b s (n d) -> b s n d", n=num_heads) x_out = torch.view_as_complex(x.to(torch.float64).reshape( x.shape[0], x.shape[1], x.shape[2], -1, 2)) + + print(f"x_out.shape: {x_out.shape}, freqs.shape: {freqs.shape}") x_out = torch.view_as_real(x_out * freqs).flatten(2) return x_out.to(x.dtype) diff --git a/diffsynth/pipelines/wan_video.py b/diffsynth/pipelines/wan_video.py index 76e1fa0..b79352f 100644 --- a/diffsynth/pipelines/wan_video.py +++ b/diffsynth/pipelines/wan_video.py @@ -1,3 +1,4 @@ +import types from ..models import ModelManager from ..models.wan_video_dit import WanModel from ..models.wan_video_text_encoder import WanTextEncoder @@ -12,6 +13,10 @@ import numpy as np from PIL import Image from tqdm import tqdm from typing import Optional +import torch.distributed as dist +from xfuser.core.distributed import (get_sequence_parallel_rank, + get_sequence_parallel_world_size, + get_sp_group) from ..vram_management import enable_vram_management, AutoWrappedModule, AutoWrappedLinear from ..models.wan_video_text_encoder import T5RelativeEmbedding, T5LayerNorm @@ -135,11 +140,19 @@ class WanVideoPipeline(BasePipeline): @staticmethod - def from_model_manager(model_manager: ModelManager, torch_dtype=None, device=None): + def from_model_manager(model_manager: ModelManager, torch_dtype=None, device=None, use_usp=False): if device is None: device = model_manager.device if torch_dtype is None: torch_dtype = model_manager.torch_dtype pipe = WanVideoPipeline(device=device, torch_dtype=torch_dtype) pipe.fetch_models(model_manager) + if use_usp: + from xfuser.core.distributed import get_sequence_parallel_world_size + from ..distributed.xdit_context_parallel import usp_attn_forward, usp_dit_forward + + for block in pipe.dit.blocks: + block.self_attn.forward = types.MethodType(usp_attn_forward, block.self_attn) + pipe.dit.forward = types.MethodType(usp_dit_forward, pipe.dit) + pipe.sp_size = get_sequence_parallel_world_size() return pipe @@ -375,11 +388,15 @@ def model_fn_wan_video( x = tea_cache.update(x) else: # blocks + if dist.is_initialized() and dist.get_world_size() > 1: + x = torch.chunk(x, get_sequence_parallel_world_size(), dim=1)[get_sequence_parallel_rank()] for block in dit.blocks: x = block(x, context, t_mod, freqs) if tea_cache is not None: tea_cache.store(x) x = dit.head(x, t) + if dist.is_initialized() and dist.get_world_size() > 1: + x = get_sp_group().all_gather(x, dim=1) x = dit.unpatchify(x, (f, h, w)) return x diff --git a/examples/wanvideo/wan_14b_text_to_video.py b/examples/wanvideo/wan_14b_text_to_video.py index 654565d..d67e1d5 100644 --- a/examples/wanvideo/wan_14b_text_to_video.py +++ b/examples/wanvideo/wan_14b_text_to_video.py @@ -1,29 +1,47 @@ import torch from diffsynth import ModelManager, WanVideoPipeline, save_video, VideoData from modelscope import snapshot_download +import torch.distributed as dist # Download models -snapshot_download("Wan-AI/Wan2.1-T2V-14B", local_dir="models/Wan-AI/Wan2.1-T2V-14B") +# snapshot_download("Wan-AI/Wan2.1-T2V-14B", local_dir="models/Wan-AI/Wan2.1-T2V-14B") # Load models model_manager = ModelManager(device="cpu") model_manager.load_models( [ [ - "models/Wan-AI/Wan2.1-T2V-14B/diffusion_pytorch_model-00001-of-00006.safetensors", - "models/Wan-AI/Wan2.1-T2V-14B/diffusion_pytorch_model-00002-of-00006.safetensors", - "models/Wan-AI/Wan2.1-T2V-14B/diffusion_pytorch_model-00003-of-00006.safetensors", - "models/Wan-AI/Wan2.1-T2V-14B/diffusion_pytorch_model-00004-of-00006.safetensors", - "models/Wan-AI/Wan2.1-T2V-14B/diffusion_pytorch_model-00005-of-00006.safetensors", - "models/Wan-AI/Wan2.1-T2V-14B/diffusion_pytorch_model-00006-of-00006.safetensors", + "/demo-huabei2/models/dit/Wan2.1-T2V-14B/diffusion_pytorch_model-00001-of-00006.safetensors", + "/demo-huabei2/models/dit/Wan2.1-T2V-14B/diffusion_pytorch_model-00002-of-00006.safetensors", + "/demo-huabei2/models/dit/Wan2.1-T2V-14B/diffusion_pytorch_model-00003-of-00006.safetensors", + "/demo-huabei2/models/dit/Wan2.1-T2V-14B/diffusion_pytorch_model-00004-of-00006.safetensors", + "/demo-huabei2/models/dit/Wan2.1-T2V-14B/diffusion_pytorch_model-00005-of-00006.safetensors", + "/demo-huabei2/models/dit/Wan2.1-T2V-14B/diffusion_pytorch_model-00006-of-00006.safetensors", ], - "models/Wan-AI/Wan2.1-T2V-14B/models_t5_umt5-xxl-enc-bf16.pth", - "models/Wan-AI/Wan2.1-T2V-14B/Wan2.1_VAE.pth", + "/demo-huabei2/models/dit/Wan2.1-T2V-14B/models_t5_umt5-xxl-enc-bf16.pth", + "/demo-huabei2/models/dit/Wan2.1-T2V-14B/Wan2.1_VAE.pth", ], torch_dtype=torch.float8_e4m3fn, # You can set `torch_dtype=torch.bfloat16` to disable FP8 quantization. ) -pipe = WanVideoPipeline.from_model_manager(model_manager, torch_dtype=torch.bfloat16, device="cuda") + +dist.init_process_group( + backend="nccl", + init_method="env://", +) +from xfuser.core.distributed import (initialize_model_parallel, + init_distributed_environment) +init_distributed_environment( + rank=dist.get_rank(), world_size=dist.get_world_size()) + +initialize_model_parallel( + sequence_parallel_degree=dist.get_world_size(), + ring_degree=1, + ulysses_degree=dist.get_world_size(), +) +torch.cuda.set_device(dist.get_rank()) + +pipe = WanVideoPipeline.from_model_manager(model_manager, torch_dtype=torch.bfloat16, device=f"cuda:{dist.get_rank()}", use_usp=True if dist.get_world_size() > 1 else False) pipe.enable_vram_management(num_persistent_param_in_dit=None) # You can set `num_persistent_param_in_dit` to a small number to reduce VRAM required. # Text-to-video