o ۷i& @sdZddlmZddlmZddlmZmZddlZddl m Z ddl m Z m Z mZe eZ ddddZ ddddZ d d!ddZeGdddZeZd"ddZdS)#zSequence Parallelism sharding utilities. This module provides low-level sharding and gathering functions for Sequence Parallelism. These can be used directly in model forward methods for semi-intrusive SP support, or internally by the SP hooks. ) annotations)contextmanager) dataclassfieldN) init_logger)get_sequence_parallel_rank get_sequence_parallel_world_size get_sp_groupTtensor torch.Tensordimintvalidateboolreturnc Cst}|dkr |St}||}|r'||dkr'td|d|d|d||kr= world_size (z). Tensor shape: r )rrsize ValueErrorshapechunk)r r r world_sizerankrra/home/ubuntu/vllm_env/lib/python3.10/site-packages/vllm_omni/diffusion/distributed/sp_sharding.pysp_shards  rcCs&t}|dkr |St}|j||dS)aGather a tensor along the specified dimension from all sequence parallel ranks. The sharded tensors from all ranks are concatenated along dim. Args: tensor: The local shard to gather. dim: The dimension along which to gather. validate: If True, validate tensor consistency (currently unused). Returns: The full tensor gathered from all ranks. Example: # At end of model forward: output = sp_gather(output, dim=1) rr)rr all_gather)r r rrsp_grouprrr sp_gatherJs r pad_valuefloattuple[torch.Tensor, int]c Cst}|dkr |dfS||}||}|dkr!t||dddfS||}t|j}|||<tj|||j|jd}tj ||g|d}t||dd|fS)aShard a tensor with automatic padding if not divisible by world_size. This is useful for variable-length sequences where padding may be needed. Args: tensor: The tensor to shard. dim: The dimension along which to split. pad_value: Value to use for padding. Returns: Tuple of (sharded_tensor, padding_size). The padding_size indicates how much padding was added to the original tensor before sharding. Example: sharded, pad_size = sp_shard_with_padding(hidden_states, dim=1) # ... process ... output = sp_gather(output, dim=1) if pad_size > 0: output = output[..., :-pad_size] # Remove padding rrFr)dtypedevicer) rrrlistrtorchfullr$r%cat) r r r rr remainderpad_size pad_shapepaddingrrrsp_shard_with_paddinghs  r.c@szeZdZUdZeedZded<eedZded<dZ ded<dd d Z e d dZ dd ddZ d!ddZdddZdS)"ShardingValidatora2Validator for tracking and verifying sharding operations. This class helps ensure that sharding and gathering operations are correctly paired in model forward passes. It tracks which tensors have been sharded and verifies that they are properly gathered. Usage: validator = ShardingValidator() with validator.track(): hidden_states = validator.shard(hidden_states, "hidden_states", dim=1) # ... model computation ... output = validator.gather(output, "hidden_states", dim=1) validator.validate() # Raises if any shard was not gathered Attributes: _sharded: Set of tensor names that have been sharded. _gathered: Set of tensor names that have been gathered. _enabled: Whether tracking is currently enabled. )default_factoryzset[str]_sharded _gatheredFr_enabledrNonecCs|j|jdS)z1Reset the validator state for a new forward pass.N)r1clearr2selfrrrresets zShardingValidator.resetccs,d|_|z dVWd|_dSd|_w)z6Context manager to enable tracking for a forward pass.TNF)r3r8r6rrrtracks zShardingValidator.trackTr r namestrr r validate_divisiblecCs<|jr||jvrtd|d|j|t|||dS)aPShard a tensor and track the operation. Args: tensor: The tensor to shard. name: A name to identify this tensor for validation. dim: The dimension along which to split. validate_divisible: If True, validate divisibility. Returns: The sharded tensor. Tensor 'z' sharded multiple timesr#)r3r1loggerwarningaddr)r7r r:r r<rrrshards   zShardingValidator.shardcCs8|jr||jvrtd|d|j|t||S)aGather a tensor and track the operation. Args: tensor: The local shard to gather. name: The name used when sharding (for validation). dim: The dimension along which to gather. Returns: The gathered tensor. r=z ' gathered without being sharded)r3r1r>r?r2r@r)r7r r:r rrrgathers    zShardingValidator.gathercCs$|j|j}|rtd|ddS)zValidate that all sharded tensors were gathered. Raises: ValueError: If any sharded tensor was not gathered. z5The following tensors were sharded but not gathered: z;. This may indicate a bug in the model's SP implementation.N)r1r2r)r7 unmatchedrrrrs  zShardingValidator.validateN)rr4T) r r r:r;r r r<rrr )r r r:r;r r rr )__name__ __module__ __qualname____doc__rsetr1__annotations__r2r3r8rr9rArBrrrrrr/s     r/cCstS)zdGet the global sharding validator instance. Returns: The global ShardingValidator. )_global_validatorrrrrget_sharding_validator srLrD)r r r r rrrr )r)r r r r r r!rr")rr/)rH __future__r contextlibr dataclassesrrr' vllm.loggerr.vllm_omni.diffusion.distributed.parallel_staterrr rEr>rrr.r/rKrLrrrrs$    2 ! 8g