o ۷iE@sdZddlmZddlmZddlmZer ddlZddlm Z eGdddZ edd Gd d d Z edd Gd d d Z edd GdddZ e e BZeeeBeeeBeedfBfZe ee Bee dfBZeeeeBfZd#ddZd#ddZd$ddZd%d!d"ZdS)&a$Sequence Parallelism configuration and plan type definitions. This module provides: 1. SequenceParallelConfig: Configuration for SP (ulysses_degree, ring_degree) 2. SequenceParallelInput/Output: Type definitions for _sp_plan declarations 3. Validation utilities for _sp_plan A _sp_plan is a dictionary that specifies how to shard/gather tensors at different points in a model's forward pass. This allows automatic handling of sequence parallelism without modifying the model's forward() method. NOTE: Our "Sequence Parallelism" (SP) corresponds to "Context Parallelism" (CP) in diffusers. We use "Sequence Parallelism" to align with vLLM-Omni terminology. Example: class MyTransformer(nn.Module): _sp_plan = { # Split inputs before model forward "": { "hidden_states": SequenceParallelInput(split_dim=1, expected_dims=3), "encoder_hidden_states": SequenceParallelInput(split_dim=1, expected_dims=3), }, # Split RoPE embeddings after pos_embed layer "pos_embed": { 0: SequenceParallelInput(split_dim=0, expected_dims=2, split_output=True), }, # Gather output after proj_out layer "proj_out": SequenceParallelOutput(gather_dim=1, expected_dims=3), } ) annotations) dataclass) TYPE_CHECKINGNc@seZdZUdZdZded<dZded<dZded<d Zd ed <d Z d ed <d Z d ed<d)ddZ e d*ddZ d*ddZd*ddZd*ddZd*ddZd*ddZd*dd Zd+d%d&Zd,d'd(Zd S)-SequenceParallelConfigaConfiguration for Sequence Parallelism using vLLM-Omni's parallel state. This class provides a unified interface for SP configuration that integrates with vLLM-Omni's existing SequenceParallelGroupCoordinator. Unlike diffusers' DeviceMesh-based approach (ContextParallelConfig), this uses the existing parallel state management. Note: This corresponds to `ContextParallelConfig` in diffusers library. Args: ulysses_degree: Number of devices for Ulysses (All-to-All) attention. Sequence is split across devices, with Q/K/V redistributed via All-to-All communication. Best for moderate sequences with good interconnect bandwidth. ring_degree: Number of devices for Ring attention. Sequence is split across devices, with K/V passed in a ring topology. Best for long sequences with limited memory/bandwidth. convert_to_fp32: Whether to convert output and LSE to float32 for numerical stability in ring attention. Note: ulysses_degree * ring_degree = sequence_parallel_size vLLM-Omni supports hybrid Ulysses-Ring attention (both > 1). intulysses_degree ring_degreeTboolconvert_to_fp32N int | None_rank _world_sizeztorch.device | None_devicereturnNonecCs@|jdks |jdkrtd|jdkr|jdkrtddSdS)Nrz0`ulysses_degree` and `ring_degree` must be >= 1.zZAt least one of `ulysses_degree` or `ring_degree` must be > 1 to use sequence parallelism.)rr ValueErrorselfr]/home/ubuntu/vllm_env/lib/python3.10/site-packages/vllm_omni/diffusion/distributed/sp_plan.py __post_init__Wsz$SequenceParallelConfig.__post_init__cCs |j|jS)z#Total sequence parallel world size.)rr rrrrsequence_parallel_size`s z-SequenceParallelConfig.sequence_parallel_sizecCddlm}|S)zGet the sequence parallel world size from parallel state. Returns: The world size for sequence parallelism. Raises: RuntimeError: If parallel state is not initialized. r) get_sequence_parallel_world_size).vllm_omni.diffusion.distributed.parallel_stater)rrrrrget_world_sizee z%SequenceParallelConfig.get_world_sizecCr)zGet the current rank in the sequence parallel group. Returns: The rank within the sequence parallel group. Raises: RuntimeError: If parallel state is not initialized. r)get_sequence_parallel_rank)rr)rrrrrget_rankrrzSequenceParallelConfig.get_rankcCr)zGet the Ulysses parallel world size. Returns: The world size for Ulysses (All-to-All) parallelism. r)get_ulysses_parallel_world_size)rr )rr rrrget_ulysses_world_size z-SequenceParallelConfig.get_ulysses_world_sizecCr)zGet the current rank in the Ulysses parallel group. Returns: The rank within the Ulysses parallel group. r)get_ulysses_parallel_rank)rr#)rr#rrrget_ulysses_rankr"z'SequenceParallelConfig.get_ulysses_rankcCr)zwGet the Ring parallel world size. Returns: The world size for Ring attention parallelism. r)get_ring_parallel_world_size)rr%)rr%rrrget_ring_world_sizer"z*SequenceParallelConfig.get_ring_world_sizecCr)zGet the current rank in the Ring parallel group. Returns: The rank within the Ring parallel group. r)get_ring_parallel_rank)rr')rr'rrr get_ring_rankr"z$SequenceParallelConfig.get_ring_rankrank world_sizedevice torch.devicec CsX||_||_||_|j|j}|}||kr*td|jd|jd|d|d dS)a!Initialize the config with runtime parallel state. This is called automatically when sequence parallelism is enabled. Args: rank: The global rank of this process. world_size: Total world size. device: The device for this rank. z(Configuration mismatch: ulysses_degree (z) * ring_degree (z) = z-, but actual sequence parallel world size is .N)r rrrr rr)rr)r*r+expected_sp_sizeactual_sp_sizerrrsetups    zSequenceParallelConfig.setupcCs |jduS)zCheck if the config has been initialized with runtime state. Returns: True if setup() has been called, False otherwise. N)r rrrris_initializeds z%SequenceParallelConfig.is_initialized)rr)rr)r)rr*rr+r,rr)rr )__name__ __module__ __qualname____doc__r__annotations__r r r rrrpropertyrrrr!r$r&r(r0r1rrrrr3s&           rT)frozenc@sHeZdZUdZded<dZded<dZded <dZded <dd dZdS)SequenceParallelInputaaConfiguration for splitting an input tensor across sequence parallel ranks. This specifies how to shard a tensor in the pre-forward or post-forward hook of a layer. The tensor will be split along the specified dimension. Note: This corresponds to `ContextParallelInput` in diffusers library. Args: split_dim: The dimension along which to split the tensor. expected_dims: Expected number of dimensions. If provided, validates that the tensor has this many dimensions before splitting. If the tensor has a different number of dimensions, splitting is skipped with a warning. split_output: If True, split the output of the layer instead of the input. This is useful for layers whose outputs should be split after preprocessing (e.g., RoPE embeddings). auto_pad: If True, automatically pad the tensor if its size along split_dim is not divisible by world_size. Creates an attention mask to indicate valid vs padding positions. The mask is stored in ForwardContext. Note: Ring attention does not support attention mask, so auto_pad should only be used with Ulysses SP. Example: # Split hidden_states along sequence dimension (dim 1) SequenceParallelInput(split_dim=1, expected_dims=3) # Split RoPE output along sequence dimension (dim 0) SequenceParallelInput(split_dim=0, expected_dims=2, split_output=True) # Split with auto-padding for variable-length sequences SequenceParallelInput(split_dim=1, expected_dims=3, auto_pad=True) r split_dimNr expected_dimsFr split_outputauto_padrstrc Cs&d|jd|jd|jd|jd S)Nz SequenceParallelInput(split_dim=, expected_dims=, split_output=z , auto_pad=))r:r;r<r=rrrr__repr__ zSequenceParallelInput.__repr__rr>) r2r3r4r5r6r;r<r=rBrrrrr9s    r9c@s0eZdZUdZded<dZded<d d d ZdS) SequenceParallelOutputaConfiguration for gathering an output tensor across sequence parallel ranks. This specifies how to gather a tensor in the post-forward hook of a layer. The tensor will be gathered along the specified dimension from all ranks. Note: This corresponds to `ContextParallelOutput` in diffusers library. Args: gather_dim: The dimension along which to gather the tensor. expected_dims: Expected number of dimensions. If provided, validates that the tensor has this many dimensions before gathering. Example: # Gather output along sequence dimension (dim 1) SequenceParallelOutput(gather_dim=1, expected_dims=3) r gather_dimNr r;rr>cCsd|jd|jdS)Nz"SequenceParallelOutput(gather_dim=r?rA)rFr;rrrrrBszSequenceParallelOutput.__repr__rD)r2r3r4r5r6r;rBrrrrrEs  rEc@sDeZdZUdZded<ded<dZded<d Zd ed <dddZdS)SequenceParallelPartialInputaConfiguration for partially splitting a tensor (e.g., split image part, keep text part). This is designed for models like LongCat/Qwen where RoPE embeddings need special handling: - Text portion: kept full across all ranks (for joint attention) - Image portion: split across ranks The tensor is assumed to be concatenated as [text_part, image_part] along split_dim. Note: This is an extension beyond diffusers' standard ContextParallelInput, designed for vLLM-Omni's dual-stream attention models. Args: split_dim: The dimension along which to split the image portion. text_len_source: How to determine text length: - str: Name of a forward parameter that contains text length - int: Fixed text length value expected_dims: Expected number of dimensions for validation. split_output: If True, split the output instead of input. Example: # Split RoPE: text portion (from txt_ids.shape[0]) kept full, image portion split SequenceParallelPartialInput( split_dim=0, text_len_source="txt_ids", # Get text length from txt_ids.shape[0] expected_dims=2, split_output=True, ) # Or with fixed text length SequenceParallelPartialInput( split_dim=0, text_len_source=512, # Fixed text length expected_dims=2, split_output=True, ) rr:z str | inttext_len_sourceNr r;Fr r<rr>c Cs&d|jd|jd|jd|jd S)Nz'SequenceParallelPartialInput(split_dim=z, text_len_source=r?r@rA)r:rHr;r<rrrrrBBrCz%SequenceParallelPartialInput.__repr__rD)r2r3r4r5r6r;r<rBrrrrrGs %  rG.valueobjectrr cCst|ttfS)z5Check if a value is a valid input configuration type.) isinstancer9rGrIrrr_is_valid_input_configwsrMcCs$t|ttfs dStdd|DS)z?Check if a value is a list/tuple of valid input configurations.Fcss|]}t|VqdSN)rM.0xrrr sz._is_valid_input_config_list..)rKlisttupleallrLrrr_is_valid_input_config_list|srVplanSequenceParallelModelPlanrc Cszt|tstdt|j|D]\}}t|ts&tdt|jt|tr,qt|tt frBt dd|Dr=qt |rBqt|tr|D]`\}}t|tt fsdtdt|jd|dt|t r~t |s~tdt|jd|d |d t |rt|t r|jstd |d |d qKt |rqKtd t|jd|d |dqtdt|jd|ddS)zValidate a _sp_plan dictionary for correctness. Args: plan: The _sp_plan dictionary to validate. Raises: ValueError: If the plan is invalid. z_sp_plan must be a dict, got z#_sp_plan keys must be strings, got css|]}t|tVqdSrN)rKrErOrrrrRsz#validate_sp_plan..z(Input spec keys must be str or int, got z for module ''zRInteger keys (output indices) must map to SequenceParallelInput/PartialInput, got z'[]z\Integer keys (output indices) require split_output=True, got split_output=False for module 'zRInput spec values must be SequenceParallelInput/PartialInput or list thereof, got z'['z']zI_sp_plan values must be dict (input spec) or SequenceParallelOutput, got N)rKdictrtyper2itemsr>rErSrTrUrVrrMr<)rW module_id module_plankeyrIrrrvalidate_sp_planst    ramodel nn.Module SequenceParallelModelPlan | NonecCs t|dd}|durt||S)zGet the _sp_plan from a model if it exists. Args: model: The model to get the plan from. Returns: The _sp_plan dictionary, or None if not defined. _sp_planN)getattrra)rbrWrrrget_sp_plan_from_models rg)rIrJrr )rWrXrr)rbrcrrd)r5 __future__r dataclassesrtypingrtorchtorch.nnnnrr9rErGAnySequenceParallelInputr[r>rrSrTSequenceParallelInputTypeSequenceParallelOutputTyperXrMrVrargrrrrs8    .8   :