Vh6|ddlZddlZddlmZddlmZddlmZddlZddl m cm Z ddl m cmcmZddlmZddlmZmZddlmZmZmZmZmZmZmZej<eZ ejBs"ejDjGdd Z$nddl%Z%e%jLd d Z' d.dd d ejPde)ejPdede*de+de*deefdZ, d.dd dejPdede*de+de*deef dZ-dejPde*de*dejPfdZ.dejPde*de*dejPfdZ/de)ejPde*de*de)ejPfdZ0 d/dedfd Z1d!d"de*fd#Z2eGd$d%Z3d&e4d'e3de*de4fd(Z5d&e4d'e3de*de4fd)Z6d&e4d'e3de*de4fd*Z7d+d"d,d"de4fd-Z8y)0N) dataclass) lru_cache)Optional)_resolve_process_group)_mesh_resources DeviceMesh)_get_group_size_by_name broadcastget_group_rankget_rank ProcessGroupscatterWorkz_dtensor::shard_dim_alltoallc(t|}t|Dcgc]}tj|}}t |}t |t }tj||j|||jScc}w)Ndim) r rangetorch empty_likerr r catchunk contiguous) input gather_dim shard_dim group_name group_size_ stacked_listgroup group_ranks Z/home/dcms/DCMS/lib/python3.12/site-packages/torch/distributed/tensor/_collective_utils.py_shard_dim_alltoall_metar#s~,Z8 9>z9JKA((/K K&z2#E8:6  IIl 3 U:9U -j : Z\ LsBzJPyTorch Distributed functional collectives do not work with torch::deploy.c|jdk(rtjdtj||||f}t |tj r|j}tj||j|||j|}|jStj||f}tjjj!||||S)NcpuzUCPU process group does not support alltoall yet, falling back with allgather + chunk!r) device_typeloggerwarningfuncolall_gather_tensor isinstanceAsyncCollectiveTensorwaitrrsizeget_local_rankr_resolve_group_nameops_dtensorshard_dim_alltoall)rrrmeshmesh_dimoutrs r"r3r32s 5   c &&uj4:JK c677 8((*Ckk#tyy2 B    ) ~~++T8,<=J 99   0 0 z9j ) group_srcoutput scatter_listr4r5async_opr8returnc|jry|j|}t|tsJ|t |k(rt |||||}|St |d|||}|S)aZ scatter a list of tensors to a device mesh dimension. We by default use the first rank of the mesh dimension as the source of truth, i.e for a 2d mesh [[0, 1], [2, 3]], if we scatter on mesh_dim = 1, we will scatter the tensor list on rank 0 to rank 0/1, and tensor list on rank 2 to rank 2/3. Args: output (torch.Tensor): the tensor to receive the scattered list. scatter_list (List[torch.Tensor]): the tensor list to be scattered. mesh_dim (int, optional): indicate which mesh dimension we want to scatter on, we by default choose the first rank on the mesh dimension as source of truth. Keyword args: group_src (int, optional): the group rank of the source data for the logical/global tensor, on the specific mesh dimension. By default, we use ``group_rank=0`` on each DeviceMesh dimension as the source data to preserve the single-device semantic. If passing ``None`` explicitly, this method simply uses its local data with no communication. Returns: A :class:`Work` object N)r:r r;r8)is_meta get_groupr+r r r)r9r:r4r5r;r8 dim_groupfuts r" mesh_scatterrBJs|J~~x(I i .. .HY'' %   J    Jr7tensorc|jry|j|}t|tsJt ||||S)a broadcast the tensor to a device mesh dimension. We by default use the first rank of the mesh dimension as the source of truth, i.e for a 2d mesh [[0, 1], [2, 3]], if we broadcast on mesh_dim = 1, we will broadcast the tensor on rank 0 to rank 0/1, and tensor on rank 2 to rank 2/3. Args: tensor (torch.Tensor): tensor to broadcast. mesh_dim (int, optional): indicate which mesh dimension we want to scatter on, we by default choose the first rank on the mesh dimension as source of truth. Keyword args: group_src (int, optional): the group rank of the source data for the logical/global tensor, on the specific mesh dimension. By default, we use ``group_rank=0`` on each DeviceMesh dimension as the source data to preserve the single-device semantic. If passing ``None`` explicitly, this method simply uses its local data with no communication. Returns: A :class:`Work` object N)r r;r8)r>r?r+r r )rCr4r5r;r8r@s r"mesh_broadcastrEs>F~~x(I i .. . V9x9 UUr7pad_dimpad_sizec|dk(r|Sddg|j|z z}||d<tjjj ||S)Nr)ndimrnn functionalpad)rCrFrGrMs r" pad_tensorrNsK1} a&FKK') *CCG 88   " "63 //r7c\|dk(r|S|j|d|j||z S)Nr)startlength)narrowr.)rCrFrGs r" unpad_tensorrSs;1} =={{7#h.  r7shardsrnum_empty_tensorsc|dk(r|St|dj}t|Dcgc]\}}||k7r|nd}}}|dj||j fdt |D|Scc}}w)Nrc3"K|]}ywN).0rrCs r" z.fill_empty_tensor_to_shards..s;Q&;s )listr. enumerate new_zerosextendr)rTrrU tensor_sizeidxr.rCs @r"fill_empty_tensor_to_shardsrbsA vay~~'(K9B;9O,5Cy a'KAY  -F MM;%(9":;; M sBzdtensor_spec.TensorMetac|j|jd|r,j|j|j dt t jjDcgc]}d}}t jj|tfd|Ds tdycc}w)N)dtype requires_grad)shapestridec3(K|] }|k( ywrXrY)rZmetalocal_metadatas r"r[z$check_tensor_meta..sD$t~%DszGInconsistent tensor metadata (including shape and stride) across ranks.) rdreupdaterfrgrr distributedget_world_sizeall_gather_objectall ValueError) local_tensorcheck_shape_stridergathered_metadatarjs @r"check_tensor_metarts##%33N "((L4G4G4I J (-U->->-M-M-O'PQ!QQ ''(9>J D2CD D U   Rs2 B=speczdtensor_spec.DTensorSpecc|jJd|jjjtj|j zS)Nz%spec should have tensor meta defined!) tensor_metarditemsizemathprodrf)rus r" spec_to_bytesr{sD    'P)PP '    ! ! * *TYYtzz-B BBr7c|eZdZUdZeed<eeed<eeed<eeed<e e ddeddfdZ y) MeshTopoInfoz9 Mesh information for collective cost estimation r4mesh_dim_devicesmesh_dim_bandwidthmesh_dim_latencyNr<cptj|j}d}|g|jz}dg|jz}dg|jz}d}t t |jD]5}|j |}|||<||z}||kDs$||xxdzcc<d||<7t||||S)NgU@g333333?g)\(?g@)rnum_devices_per_hostr&rJreversedrr.r}) r4rbase_bwrrr~total_num_devicesr5 num_devicess r"build_from_meshzMeshTopoInfo.build_from_meshs /CCDDTDTU%Y25499,3? tyy!12 1H))H-K)4 X &  ,  #77 #8,4,-0 * 1 "$68H  r7) __name__ __module__ __qualname____doc__r__annotations__r\intfloat staticmethodrrrYr7r"r}r}sY 3iU #5k!t_ j ^  r7r}bytes_gb mesh_topoc|j|}|j|}|dz }d||j|zz}||z|z |z }||dzzSNrffffff@.Ar~rrrrr5num_devices_on_mesh_dimrnum_hopslatencybws r"allgather_costrsm'88B"55h?&*HHy99(CCCG X  7 7;M MB R#X r7c|j|}|j|}d|dz z}d||j|zz}||z|z |z }||dzzS)Nrrrrrs r"allreduce_costr(ss'88B"55h?+a/0HHy99(CCCG X  7 7;M MB R#X r7c|j|}|j|}|dz }d||j|zz}||z|z |z }||dzzSrrrs r"reduce_scatter_costr3so (88B"55h?&*HHy99(CCCG X  7 7;M MB R#X r7 current_spec target_specc|j|jk7r tdS|jrytj |j}d}t ||j z dz dz dz }tt|j|jD]\}\}}||k(r|j|}|jr&|jr||z}|t|||z }U|jr$|jr|t|||dzz }|jr!|jr|t|||z }|jr&|jr|t!|||z }||z}|js|jstdcS|S)a This function returns the cost of redistribute from current to target DTensorSpec. NOTE: 1. Only consider communication cost here, since computation costs for redistribute are quite trival (i.e. we only need to narrow or simple division) 2. Only consider redistribute cost on same mesh, cross mesh communication cost is not quite needed for operator strategy estimation/selection. infgig?)r4r is_replicatedr}rr{ num_shardsr]zip placementsr~is_shard is_replicater is_partialrr) rrrcost comm_bytes_gbicurrenttargetrs r"redistribute_costrAsK,,,U|!!#,,\->->?I Dl#l&=&==DtKdR !* L # #[%;%;<!  GV f  "+"<"rs ! ::===E   8 $"u!!# ]]  !?@  A  HMMT 8 ;; LL;u||$; ; ;  ;;d^;B (V  (V LL(V (V(V (V  (Vd^(VV0u||0c0S0U\\0su||    +. CF  %,, &+ '(0C2CsC  * *  * ZU|suU|su     <,<+< <r7