o ٷi&o@s dZddlmZddlZddlmZddlmZddlZddl m Z ddl m Z ddl mZmZmZmZmZmZddlmZmZdd lmZmZe eZd Zd ZeGd d d Zd-ddZGdddeZ GdddeZ!d.ddZ"d.ddZ#d/d$d%Z$d0d&d'Z% d1d2d)d*Z&d3d+d,Z'dS)4aSequence Parallelism hooks for non-intrusive SP support. This module implements the hook-based mechanism for applying sequence parallelism to models without modifying their forward() methods. Usage: 1. Define _sp_plan on your model class (corresponds to diffusers' _cp_plan) 2. Call apply_sequence_parallel(model, config, plan) to enable SP 3. Call remove_sequence_parallel(model, plan) to disable SP The hooks automatically shard inputs before forward and gather outputs after, based on the plan specification. ) annotationsN) dataclass)Any) init_logger)AnySequenceParallelInputSequenceParallelConfigSequenceParallelInputSequenceParallelModelPlanSequenceParallelOutputSequenceParallelPartialInput) sp_gathersp_shard) HookRegistry ModelHookz sp_input---{}zsp_output---{}c@s:eZdZUdZdZded<dZded<  ddddZdS)ModuleForwardMetadatazMetadata for mapping forward() parameter names to args/kwargs positions. This caches the inspection of a module's forward signature to efficiently locate parameters by name in subsequent calls. Nzdict[str, int] | Nonecached_parameter_indicesz type | None_cls identifierstrargstuplekwargs dict | Nonereturntuple[Any, bool, int | None]cCs|pi}||vr||ddfS|jdur9|j|d}|dur'td|d|t|kr4||d|fSdd|fS|jdurBtdtt|jjj }|dd}dd t |D|_||jvrktd|d |j|}|t|kr{dd|fS||d|fS) a(Get a parameter value from args or kwargs by name. Args: identifier: The parameter name to look up. args: Positional arguments passed to forward. kwargs: Keyword arguments passed to forward. Returns: Tuple of (value, is_kwarg, index). - value: The parameter value (or None if not found) - is_kwarg: True if found in kwargs - index: Position in args if found there Raises: ValueError: If parameter not found in signature. TN Parameter 'z' not found in cached indices.Fz$Model class is not set for metadata.cSsi|]\}}||qSrr).0iparamrr_/home/ubuntu/.local/lib/python3.10/site-packages/vllm_omni/diffusion/hooks/sequence_parallel.py uszIModuleForwardMetadata._get_parameter_from_args_kwargs..z"' not found in function signature.) rget ValueErrorlenrlistinspect signatureforward parameterskeys enumerate)selfrrrindexr*rrr!_get_parameter_from_args_kwargsJs*         z5ModuleForwardMetadata._get_parameter_from_args_kwargsrN)rrrrrrrr)__name__ __module__ __qualname____doc__r__annotations__rr/rrrr!r?s   rmodule nn.ModulercCsZt|dr+t|jdkr+tt|j}|dur|}n |St|dr+t|jdks |S)zUnwrap a module from any wrappers to get the original class. Args: module: Potentially wrapped module. Returns: The unwrapped module. _modulesrN)hasattrr%r8nextitervalues)r6innerrrr!_unwrap_modules r>csdeZdZdZd)fdd Zd*d d Zd+ddZd,ddZd-ddZ  d.d/d$d%Z d0d'd(Z Z S)1SequenceParallelSplitHookaFHook for splitting inputs before a module's forward pass. This hook is registered to modules that need their inputs sharded across sequence parallel ranks. It intercepts the forward call, shards specified inputs according to the plan, and passes the sharded inputs to the original forward. For split_output=True inputs, it shards the output instead. Supports both SequenceParallelInput (full split) and SequenceParallelPartialInput (partial split for text/image separation). Note: This corresponds to `ContextParallelSplitHook` in diffusers. metadataJdict[str | int, AnySequenceParallelInput | list[AnySequenceParallelInput]]configrrNonecs&t||_||_d|_i|_dSN)super__init__r@rBmodule_forward_metadata_text_len_cacher-r@rB __class__rr!rFs  z"SequenceParallelSplitHook.__init__r6r7cCst|j}t|d|_|S)N)r)r>rKrrG)r-r6clsrrr!initialize_hooks  z)SequenceParallelSplitHook.initialize_hookrrrtuple[tuple, dict]c Ost|}|j|jD]\}}t|ttfr|jrq|j |||\}}} |dur-qt|t j r<| ||||}nit|ttfrt|ttfsWtd|dt|jt|t|krptdt|d|dt|g} t|D]\} } t | r|| js| | || ||} | | qvt|| }n tdt|j|r|||<q| dur| t|kr||| <qtd|d ||_t||_t||fS) zShard inputs before forward.Nzq4t| t t frH| jsIq4| t|kr\td| dt|d|| } || | |j|j|| <|| | urtd }q4|r|r|jd 7_|r|dSt||S) z,Shard outputs for split_output=True entries.rget_forward_contextis_forward_context_availablecs|] }t|tjVqdSrDrRrTrUrrbrrr! z9SequenceParallelSplitHook.post_forward..FzIndex z$ out of bounds for output of length .Tr)#vllm_omni.diffusion.forward_contextrfrgrRrTrUr&rallr@rQintrr rSr%r$rVr[rZ_sp_shard_depthrW) r-r6rdrfrgrXis_tensor_list output_listactually_shardedr.r^originalrrr! post_forwards,      z&SequenceParallelSplitHook.post_forwardsp_inputr rdictrpc Cs|j}t|tr |S||jvr|j|Sz@|j|||\}}}|dur,td|dt|tjr8|j d}nt|tr@|}n td|dt |j ||j|<|WStyk}z td|d||d}~ww) z2Resolve text length from the source specification.Nrz(' is None, cannot determine text length.rz#Cannot determine text length from 'z ' of type z#Failed to resolve text_len_source 'z': ) text_len_sourcerRrprHrGr/r$rTrUshaperWr1) r-rwrrsourceval_text_lenerrr!_resolve_text_len s(       z+SequenceParallelSplitHook._resolve_text_lenrNrb torch.Tensorrrc Cs|pi}|jdur!||jkr!td|jd|d|St|tr8|jr0|||jSt ||jddSt|t rj| |||}|j}| |d|}| ||| ||}t ||dd} tj|| g|dStd t|j) z4Shard a tensor according to the input specification.NzExpected tensor with dims, got z. Skipping split.FvalidaterdimzUnsupported input config type: ) expected_dimsrlogger warning_oncerRrauto_pad_shard_with_auto_pad split_dimr r rnarrowsizerTcatr$rWr1) r-rbrwrrr~r text_part image_partimage_part_shardedrrr!rV0s   z+SequenceParallelSplitHook._prepare_sp_inputrc Csdddlm}ddlm}m}m}ddlm}m}|} | dkr!|S| |} | | } | dkr5t ||ddS|d} | j sLt d | d | d | d |dkr\t d | d | d | | } | | }t|j}| ||<tj||j|jd}tj||g|d}|r|}|jdur| |_| |_td| d|d| d| d|d |}|j| |d|S)aPShard tensor with automatic padding and attention mask creation. When sequence length is not divisible by SP world size, this method: 1. Pads the tensor to make it divisible 2. Creates an attention mask indicating valid vs padding positions 3. Stores the mask and padding info in ForwardContext r)get_attn_backend)get_ring_parallel_world_sizeget_sequence_parallel_rank get_sequence_parallel_world_sizererFrzSequence length (z%) is not divisible by SP world size (z). Cannot use zZ which does not support attention_mask. Please switch to SDPA or Ascend attention backend.ze). Cannot use Ring attention which does not support attention_mask. Please switch to Ulysses SP only.)dtypedevicerNzAuto-padded sequence from z to z (pad_size=z , world_size=z, dim=))&vllm_omni.diffusion.attention.selectorr.vllm_omni.diffusion.distributed.parallel_staterrrrnrfrgrr supports_attention_maskr$get_namer&rzrTzerosrrrsp_original_seq_lensp_padding_sizerdebugchunk)r-rbrrrrrrfrg world_sizeseq_len remainder attn_backendpad_sizepadded_seq_len pad_shapepaddingx_paddedctxrankrrr!rTsV     z.SequenceParallelSplitHook._shard_with_auto_pad)r@rArBrrrCr6r7rr7)r6r7rrrrrrNr6r7rdrrr)rwr rrrrxrrpr0) rbrrwrrrrrrr)rbrrrprr) r1r2r3r4rFrMrcrvrrVr __classcell__rrrJr!r?s  6 !' $r?cs6eZdZdZdfdd Zdd d ZdddZZS)SequenceParallelGatherHookaGHook for gathering outputs after a module's forward pass. This hook is registered to modules that need their outputs gathered from all sequence parallel ranks. It intercepts the output and gathers it according to the plan specification. Note: This corresponds to `ContextParallelGatherHook` in diffusers. r@5SequenceParallelOutput | list[SequenceParallelOutput]rBrrrCcs*tt|tr |g}||_||_dSrD)rErFrRr r@rBrIrJrr!rFs   z#SequenceParallelGatherHook.__init__r6r7cCs|SrDr)r-r6rrr!rMsz*SequenceParallelGatherHook.initialize_hookrdrc Csddlm}m}t|tj}|r|g}nt|ttfr$tdd|Ds.t dt |j t|}t |t |j krKt dt |j dt |dd }|rV|}|j}d }t|j D]X\} } | d urfq]|| } | jd ur| | jkrtd | jd | d q]t| | jd d} |d ur| | j|kr| | jd|} td| jd|d| || <d}q]|r|r|}td|jd|_|r|dSt ||S)z;Gather outputs after forward and remove padding if applied.rrecsrhrDrirjrrr!rkrlz:SequenceParallelGatherHook.post_forward..z.Expected tensor or list/tuple of tensors, got rOz outputs, got rmNFzExpected output tensor with rz. Skipping gather.rz Removed padding: gathered shape z (original_seq_len=rTr)rnrfrgrRrTrUr&rror$rWr1r%r@rr,rrrrr gather_dimrrrrzmaxrq) r-r6rdrfrgrXoriginal_seq_lenractually_gatheredrr^rbgatheredrrr!rvsB    z'SequenceParallelGatherHook.post_forward)r@rrBrrrCrr)r1r2r3r4rFrMrvrrrrJr!rs  rmodelr]rnn.Module | list[nn.Module]cCs |ddkr tdt||S)aaGet a submodule by dotted name, supporting wildcards. Args: model: The root module. name: Dotted path to submodule. Use "*" to match all children of a ModuleList. Returns: The submodule or list of submodules if wildcard used. Raises: ValueError: If the path is invalid or module not found. *rz.Wildcard '*' can only be used once in the name)countr$_find_submodule_by_name)rr]rrr!_get_submodule_by_names rcCs|dkr|Sd|vr|ddn|df\}}|dkr?t|tjs$tdg}|D]}t||}t|ts7|g}||q(|St||rNt ||}t||Std|d|j j d)z,Recursive helper for _get_submodule_by_name.rmrrz-Wildcard '*' can only be used with ModuleList'z' is not a submodule of ') splitrRnn ModuleListr$rr&extendr9getattrrKr1)rr] first_atomremaining_name submodules submodule subsubmodulesrrr!rs"        rrBrplanr rCc Cstd|jd|jdt||D]t\}}t||}t|ts)|g}td|dt |d|D]R}t|t rKt ||}t |}n5t|tttfrvt|tr[|g}tdd|Dsktd |t||}t |}n td t|jt|} | ||q9qd S) aApply sequence parallel hooks to a model according to the plan. This function registers hooks on the specified submodules to automatically shard inputs and gather outputs for sequence parallelism. Note: This corresponds to `apply_context_parallel` in diffusers. The complete SP flow is: 1. Input sharding (SequenceParallelSplitHook): Split sequence across SP ranks 2. Attention parallelism (handled by vLLM-Omni's Attention layer): - Ulysses: All-to-All over Q/K/V heads - Ring: K/V circulation in ring topology - Hybrid: Both (Ulysses handles head redistribution, Ring handles K/V) 3. Output gathering (SequenceParallelGatherHook): Gather sequence from SP ranks Args: module: The model to apply SP to. config: The sequence parallel configuration. plan: Dictionary mapping module names to input/output specifications. Example: config = SequenceParallelConfig(ulysses_degree=2) plan = { "": {"hidden_states": SequenceParallelInput(split_dim=1, expected_dims=3)}, "proj_out": SequenceParallelOutput(gather_dim=1, expected_dims=3), } apply_sequence_parallel(model, config, plan) Note: vLLM-Omni's Attention layer automatically handles the internal parallelism (Ulysses All-to-All or Ring attention) based on the forward_context configuration. This function only handles input/output sharding for the model as a whole. z0Applying sequence parallel with config: ulysses=z, ring=z , plan keys: zApplying SP hooks to 'z' (z module(s))css"|] }t|tp |duVqdSrD)rRr rjrrr!rkRs z*apply_sequence_parallel..z.Expected SequenceParallelOutput elements, got zUnsupported plan type: N)rrulysses_degree ring_degreer&r+rQrrRr%rxr?_SP_INPUT_HOOK_TEMPLATEformatr rror$r_SP_OUTPUT_HOOK_TEMPLATErWr1r get_or_create register_hook) r6rBr module_id sp_model_planrmhook hook_nameregistryrrr!apply_sequence_parallels8'           rcCs|D]@\}}t||}t|ts|g}|D],}t|dd}|dur$qt|tr/t|}nt|ttt fr=t |}nq| |qqdS)zRemove sequence parallel hooks from a model. Note: This corresponds to `remove_context_parallel` in diffusers. Args: module: The model to remove SP from. plan: The same plan used when applying SP. _hook_registryN) rQrrRr&rrxrrr rr remove_hook)r6rrrrrrrrrr!remove_sequence_parallel]s        rSequenceParallelConfig | Nonec Csddlm}m}ddlm}||}|dur td|jjd|durR|}|}t||d}|dkr;|dkr;d }n |dkrBd }nd }t d |d |d|t |||t d|jjdS)aEnable sequence parallelism for a model using its _sp_plan. This is a convenience function that reads the model's _sp_plan attribute and applies sequence parallelism automatically. Note: This corresponds to `enable_context_parallel_for_model` in diffusers, but uses vLLM-Omni's _sp_plan instead of diffusers' _cp_plan. The function performs two main tasks: 1. Applies _sp_plan hooks to shard inputs and gather outputs 2. Ensures Attention layers are configured for the correct parallel mode (handled automatically by vLLM-Omni's forward_context mechanism) Args: model: The model to enable SP for. Must have a _sp_plan attribute. config: Optional config. If None, uses default based on current parallel state. Raises: ValueError: If model has no _sp_plan defined. Note: vLLM-Omni supports Ulysses + Ring hybrid parallelism: - ulysses_degree > 1: Uses All-to-All communication over Q/K/V heads - ring_degree > 1: Uses Ring attention with K/V passing - Both > 1: Hybrid mode (Ulysses handles head redistribution, Ring handles K/V circulation) r)rget_ulysses_parallel_world_sizeget_sp_plan_from_modelNzModel zY has no _sp_plan defined. Define _sp_plan as a class attribute or pass a plan explicitly.)rrrhybridulyssesringz6Created SP config from parallel state: ulysses_degree=z, ring_degree=z, mode=z!Enabled sequence parallelism for ) rrr'vllm_omni.diffusion.distributed.sp_planrr$rKr1rrinfor) rrBrrrrrrmoderrr!"enable_sequence_parallel_for_model}s<  rcCs.ddlm}||}|durt||dSdS)zDisable sequence parallelism for a model. Note: This corresponds to `disable_context_parallel_for_model` in diffusers. Args: model: The model to disable SP for. rrN)rrr)rrrrrr!#disable_sequence_parallel_for_models rr)rr7r]rrr)r6r7rBrrr rrC)r6r7rr rrCrD)rr7rBrrrC)rr7rrC)(r4 __future__rr' dataclassesrtypingrrTtorch.nnr vllm.loggerrrrrrr r r +vllm_omni.diffusion.distributed.sp_shardingr r vllm_omni.diffusion.hooks.baserrr1rrrrr>r?rrrrrrrrrrr!s8       B N   G" E