o پi@sdZddlmZmZddlZddlmZddlmZz ddl m m Z Wne y7ddlm mZ YnwddlmZmZmZdejdedee jd d fd d ZGd ddeZGdddeZ ddejjdeefddZdS)z/ Common utilities for torch model parallelism. )OptionalSequenceN) DeviceMesh)ColwiseParallelRowwiseParallelparallelize_module full_tensor device_mesh placementsreturnz dt.DTensorcCsBtj|j||\}}ddt||D}||}tj|||S)aD Locally shards a full tensor based on indicated sharding arrangement, and returns a DTensor containing the local shard. .. warning:: This is a private API that is subject to change. It skips the communication otherwise required by `distribute_tensor`. It is only applicable to cases where all ranks have the same `full_tensor`. For example, in distributed inference all ranks load from the same checkpoint. This API will not check for data equality between ranks, it is thus user's responsibility to ensure the `full_tensor` is the same across ranks. Args: full_tensor (torch.Tensor): the full tensor to be sharded. device_mesh (:class:`DeviceMesh`): DeviceMesh to place the DTensor. Must have same dimension as the number of placements. placements (Sequence[:class:`Shard`]): the placements that describes how to place the local tensor on DeviceMesh. Returns: A :class:`DTensor` object with the shard as its local tensor. Examples: >>> # xdoctest: +SKIP("need world_size and rank") >>> device_mesh = dist.init_device_mesh("cuda", (world_size,)) >>> full_tensor = torch.arange(world_size, device=f"cuda:{rank}") >>> dtensor = _shard_tensor(full_tensor, device_mesh, [Shard(1)]) cSsg|] \}}t|||qS)slice).0 cur_shape cur_offsetr r T/home/ubuntu/.local/lib/python3.10/site-packages/sglang/srt/layers/model_parallel.py <s z!_shard_tensor..)dt_utils%compute_local_shape_and_global_offsetshapezipDTensor from_local)rr r roffsetslices local_tensorr r r _shard_tensors!rc@seZdZdZddZdS)ColwiseParallelShardedz A version of ColwiseParallel where the local weight has been already sharded. This is used for the fused wqkv case, where during loading, we already sharded wq, wk, wv before fusing them. cCsJ|D]\}}tj||tdg}tjj|dd}|||qdS)NrF) requires_grad) named_parametersrrrShardtorchnn Parameterregister_parameter)selfnamemoduler paramdtensor dist_paramr r r_partition_linear_fnLs z+ColwiseParallelSharded._partition_linear_fnN)__name__ __module__ __qualname____doc__r,r r r rrDs rcs,eZdZdZddZefddZZS)RowwiseParallelMaybeWaita5 A version of RowwiseParallel that waits for the output (establish dependency between comm stream and compute stream in CUDA sense) before going into the next op. This is needed to workaround the current interaction between AsyncCollectiveTensor and multi-platform ops, such as `RMSNorm`. c Cs`|dtt|j|tdgt|dddur.|dtt|j |t gdSdS)Nweightbias) r%r#r$rr2rr!getattrdistribute_tensorr4 Replicate)r&r'r(r r r rr,^sz-RowwiseParallelMaybeWait._partition_linear_fncs&ttt|||||}tjj|SN)superr1_prepare_output_fnr" distributed_functional_collectives wait_tensor)output_layoutsuse_local_outputmodoutputsr  __class__r rr:os z+RowwiseParallelMaybeWait._prepare_output_fn)r-r.r/r0r, staticmethodr: __classcell__r r rBrr1Vs r1r(cs(dtjjddffdd }||dS)a Tensor parallelize the model across the given device mesh. Args: module (`torch.nn.Module`): The module to tensor parallelize. device_mesh (`torch.distributed.DeviceMesh`): The device mesh to use for tensor parallelism. r@r Ncst|dd}|dur dS|D]3\}}||}|dkr%t|tq|dkr1t|tq|dkr=t|tqtd|dS)N_tp_planColwiseRowwiseColwise_ShardedzUnknown TP style )r5items get_submodulerrr1r ValueError)r@tp_plan child_nametp_stylesubmodr r rtplizes  ztensor_parallel..tplize)r"r#Moduleapply)r(r rRr rQrtensor_parallelysrUr8)r0typingrrr"torch.nnr#torch.distributed.device_meshrtorch.distributed.tensorr;tensorr ImportErrortorch.distributed._tensor_tensor!torch.distributed.tensor.parallelrrrTensorr!rrr1rSrUr r r rs8    ,%