o TiC@slddlmZddlmZddlmZGdddZddZ Gdd d eZ Gd d d eZ Gd d d Z dS))comm) namedtuple)productc@steZdZdZddZddZddZdd gd d fd d ZddZddZ ddZ ddZ ddZ ddZ ddZdS)ProcessTopologyai Manages the mapping of n-dimensional Cartesian coordinates to linear indices. This mapping is used to map the rank of processes to the grid for various forms of parallelism. Each axis of the tensor is accessed by its name. The provided ordering of the axes defines the layout of the topology. ProcessTopology uses a "row-major" layout of the tensor axes, and so axes=['x', 'y'] would map coordinates (x,y) and (x,y+1) to adjacent linear indices. If instead axes=['y', 'x'] was used, coordinates (x,y) and (x+1,y) would be adjacent. Some methods return ProcessCoord namedtuples. csv|_|_td|_i_dd|D}tt|D]\}fddjD}jdi|}|j|<qdS)zCreate a mapping of n-dimensional tensor coordinates to linear indices. Arguments: axes (list): the names of the tensor axes dims (list): the dimension (length) of each axis of the topology tensor ProcessCoordcSsg|]}t|qS)range).0drrS/home/ubuntu/.local/lib/python3.10/site-packages/deepspeed/runtime/pipe/topology.py )sz,ProcessTopology.__init__..csi|] }|j|qSr)axesindex)r axiscoordselfrr ,sz,ProcessTopology.__init__..Nr)r dimsrrmapping enumeratecartesian_product)rr rranges global_rankkeyrrr __init__s  zProcessTopology.__init__cKsNt|t|jkr td|jdi|}||jvs"Jd|d|j|S)zReturn the global rank of a process via its coordinates. Coordinates are specified as kwargs. For example: >>> X = ProcessTopology(axes=['x', 'y'], dims=[2,3]) >>> X.get_rank(x=0, y=1) 1 z7get_rank() does not support slices. Use filter_match())zkey z invalidNr)lenr ValueErrorrr)r coord_kwargsrrrr get_rank1s   zProcessTopology.get_rankcC|jS)zAReturn a list of the axis names in the ordering of the topology. )r rrrr get_axis_namesAzProcessTopology.get_axis_namesdatapipe_-c s`tfdd|D}g}|D]}t|j|d|}||||dq||S)aReturn a string representation of a rank. This method is primarily used for checkpointing model data. For example: >>> topo = Topo(axes=['a', 'b'], dims=[2, 2]) >>> topo.get_rank_repr(rank=3) 'a_01-b_01' >>> topo.get_rank_repr(rank=3, omit_axes=['a']) 'b_01' Args: rank (int): A rank in the topology. omit_axes (list, optional): Axes that should not be in the representation. Defaults to ['data', 'pipe']. inner_sep (str, optional): [description]. Defaults to '_'. outer_sep (str, optional): [description]. Defaults to '-'. Returns: str: A string representation of the coordinate owned by ``rank``. csg|]}|vr|qSrrr a omit_axesrr r [z1ProcessTopology.get_rank_repr..rank02d) frozensetr"getattr get_coordappendjoin) rr.r+ inner_sep outer_sepr namesaxax_rankrr*r get_rank_reprEs zProcessTopology.get_rank_reprcCs ||jvrdS|j|j|S)zReturn the number of processes along the given axis. For example: >>> X = ProcessTopology(axes=['x', 'y'], dims=[2,3]) >>> X.get_dim('y') 3 r)r rr)rrrrr get_dimbs zProcessTopology.get_dimcCs4|jD] \}}||kr|Sqtd|d)aWReturn the coordinate owned by a process rank. The axes of the returned namedtuple can be directly accessed as members. For example: >>> X = ProcessTopology(axes=['x', 'y'], dims=[2,3]) >>> coord = X.get_coord(rank=1) >>> coord.x 0 >>> coord.y 1 zrank z not found in topology.)ritemsr)rr.ridxrrr r2ns  zProcessTopology.get_coordcsjvrgSfddjDg}fddD}t|D]1fddD}g}tD]}jdi||i}|j|q5||q |S)a Construct lists suitable for a communicator group along axis ``axis``. Example: >>> topo = Topo(axes=['pipe', 'data', 'model'], dims=[2, 2, 2]) >>> topo.get_axis_comm_lists('pipe') [ [0, 4], # data=0, model=0 [1, 5], # data=0, model=1 [2, 6], # data=1, model=0 [3, 7], # data=1, model=1 ] Returns: A list of lists whose coordinates match in all axes *except* ``axis``. csg|]}|kr|qSrrr()rrr r r,z7ProcessTopology.get_axis_comm_lists..csg|] }t|qSr)rr;r(r!rr r scsi|] }||qSr)rr()r other_axesrr rsz7ProcessTopology.get_axis_comm_lists..Nr)r rrr;rr3r)rrlistsr other_keyssub_listaxis_keyrr)rrr>rr get_axis_comm_listss   z#ProcessTopology.get_axis_comm_listsc s.fdd}t|j}fdd|DS)aVReturn the list of ranks whose coordinates match the provided criteria. Example: >>> X = ProcessTopology(axes=['pipe', 'data', 'model'], dims=[2, 2, 2]) >>> X.filter_match(pipe=0, data=1) [2, 3] >>> [X.get_coord(rank) for rank in X.filter_match(pipe=0, data=1)] [ProcessCoord(pipe=0, data=1, model=0), ProcessCoord(pipe=0, data=1, model=1)] Arguments: **filter_kwargs (dict): criteria used to select coordinates. Returns: The list of ranks whose coordinates match filter_kwargs. cs*D]\}}t|||krdSqdS)NFT)r<r1)xrval) filter_kwargsrr _filter_helpers z4ProcessTopology.filter_match.._filter_helpercsg|]}j|qSrr)r rr!rr r sz0ProcessTopology.filter_match..)filterrkeys)rrFrGcoordsr)rFrr filter_matchs zProcessTopology.filter_matchcs,j|fddjD}|S)a/Returns the list of global ranks whose coordinate in an axis is idx. For example: >>> X = ProcessTopology(axes=['x', 'y'], dims=[2,3]) >>> X.get_axis_list(axis='x', idx=0) [0, 1, 2] >>> X.get_axis_list(axis='y', idx=0) [0, 3] cs"g|] }|krj|qSrrH)r kaxis_numr=rrr r s"z1ProcessTopology.get_axis_list..)r rrrJ)rrr=ranksrrNr get_axis_lists zProcessTopology.get_axis_listcC t|jSN)rrr!rrr world_size zProcessTopology.world_sizecCrRrS)strrr!rrr __str__rUzProcessTopology.__str__N)__name__ __module__ __qualname____doc__rrr"r:r;r2rCrLrQrTrWrrrr r s  ( rcCs^|dkrtdg}|dkr-td|dD]}||dkr(||||}nq|dks|S)z8 Returns the prime factorization of positive integer N. rz!Values must be strictly positive.)rrr3)Nprimes candidaterrr _prime_factorss  rac eZdZdZfddZZS)PipeDataParallelTopologya& A topology specialization for hybrid data and pipeline parallelism. Uses data parallelism on the last dimension to encourage gradient reductions to use high-bandwidth intra-node links and lower-volume pipeline communications to use low-bandwidth inter-node links. cstjddg||gddS)Nr%r$r rsuperr)rnum_ppnum_dp __class__rr rsz!PipeDataParallelTopology.__init__rXrYrZr[r __classcell__rrrir rcsrccrb)PipeModelDataParallelTopologyz> A topology for hybrid pipeline, model, and data parallelism. cstjgd|||gddS)N)r%r$modelrdre)rrgnum_mprhrirr rsz&PipeModelDataParallelTopology.__init__rkrrrir rmsrmc@seZdZdZd+ddZddZddZd d Zd d Zd dZ ddZ ddZ ddZ ddZ ddZddZddZddZdd Zd!d"Zd#d$Zd%d&Zd'd(Zd)d*ZdS),PipelineParallelGridaImplements a grid object that stores the data parallel ranks corresponding to each of the model parallel stages The grid object organizes the processes in a distributed pytorch job into a 2D grid, of stage_id and data_parallel_id. self.stage_id and self.data_parallel_id stores the stage id and the data parallel id of current process. self.dp_group groups the processes by stage_id. self.dp_group[i], is a list containing all process ranks whose stage_id is i. self.p2p_groups stores a list of tuple, where each tuple stores process ranks of adjacent stages for a given data_parallel_id. For example if num_stage is 5 then a tuple [7,8] represents stages [3, 4], with data_parallel id = 1. A stage wrap around will appear as non-adjacent ranks, for example tuple [4,0] with representing wrap-around stage 4 and 0, for data_parallel_id = 0, or similarly [9,5] represents wrapped around stages [4,0] for data_parallel_id = 1. Nc Cst|_t|_|dur|jdkrtd|||_n&d}d}tt|jD]\}}|ddkr6||9}q'||9}q't ||d|_t |j dd|_ t |j dd|_ t |j dd|_|j|_|slJd ||_||_d|_d |_t|j D],}t|jjd|d }|jdkr tj|d } |j|vr| |_t||_||j|_q|jd ksJ|jdusJg|_|jd|_ |j D]} tj| d } |j| vr| |_| |_!q|jdk|_"|j|j dk|_#|$|_%g|_&d|_'|jd|_(|j(D]}|jdkr tj|d } |j|vr||_&| |_'q|j'dus&J|jdkrNt|jD]} | g} tj| d } | d|jkrJ| |_)| |_*q1dSg|_+|jd|_,|j,D]} tj| d } |j| vro| |_)| |_*q[dS) NrzUsing topology:r\r])rhrgr$r%rnz Invalid Grid)rr=)rP)-distrrget_world_sizerTprint_toporrarcmaxr;data_parallel_sizepipe_parallel_sizemodel_parallel_sizeslice_parallel_size_is_grid_valid get_stage_idstage_idget_data_parallel_iddata_parallel_idds_model_proc_group ds_model_rankrsortedrQ new_grouprds_model_world_sizerdp_grouprC dp_groups dp_proc_groupis_first_stage is_last_stage_build_p2p_groups p2p_groupspp_group pp_proc_group pipe_groups slice_groupslice_proc_groupmp_group model_groups) rtopology process_grouprgrhr=primedprP proc_groupg group_rankgrouprrr rs                           zPipelineParallelGrid.__init__cC|jj|jdjSNr-)rur2rr%r!rrr r|oz!PipelineParallelGrid.get_stage_idcCrr)rur2rr$r!rrr r~rrz)PipelineParallelGrid.get_data_parallel_idcCs|jd}g}t|jD]+}|D]&}t||jksJ||vr7||}||d|j}|||gnqq t||jksBJ|S)ziGroups for sending and receiving activations and gradients across model parallel stages. r%r\)rurCrrTrrxrr3)r comm_lists p2p_listsr.lr= buddy_rankrrr rus  z&PipelineParallelGrid._build_p2p_groupscCs0d}|jD] }||j|9}q|tkS)Nr\)rur"r;rrrs)rrPr8rrr r{s z#PipelineParallelGrid._is_grid_validcKs8|j|j}|jdd|i|}|jjdi|S)Nr%r)rur2r_replace_asdictr)rr}kwargsme transformrrr stage_to_globalsz$PipelineParallelGrid.stage_to_globalcCr rS)rur!rrr rzPipelineParallelGrid.topologycCr rS)rr!rrr get_global_rankrz$PipelineParallelGrid.get_global_rankcCs|S)z1 The stage of the pipeline this rank resides in. )r|r!rrr get_pipe_parallel_ranksz+PipelineParallelGrid.get_pipe_parallel_rankcCr )z' The number of stages in the pipeline. )rxr!rrr get_pipe_parallel_world_sizer#z1PipelineParallelGrid.get_pipe_parallel_world_sizecCr )z. The group of ranks within the same pipeline. )rr!rrr get_pipe_parallel_groupr#z,PipelineParallelGrid.get_pipe_parallel_groupcCr )z& Which pipeline this rank resides in. )rr!rrr get_data_parallel_rankr#z+PipelineParallelGrid.get_data_parallel_rankcCr )z The number of pipelines. )rwr!rrr get_data_parallel_world_sizer#z1PipelineParallelGrid.get_data_parallel_world_sizecCr )z< The group of ranks within the same stage of all pipelines. )rr!rrr get_data_parallel_groupr#z,PipelineParallelGrid.get_data_parallel_groupcCr rS)rr!rrr get_model_parallel_rankrz,PipelineParallelGrid.get_model_parallel_rankcCr rS)rr!rrr get_model_parallel_world_sizerz2PipelineParallelGrid.get_model_parallel_world_sizecCr rS)rr!rrr get_model_parallel_grouprz-PipelineParallelGrid.get_model_parallel_groupcCs$d|jvr|jj|jdjSdS)Nrnr-r)rur"r2rrnr!rrr get_slice_parallel_ranksz,PipelineParallelGrid.get_slice_parallel_rankcCr rS)rzr!rrr get_slice_parallel_world_sizerz2PipelineParallelGrid.get_slice_parallel_world_sizecCr rS)rr!rrr get_slice_parallel_grouprz-PipelineParallelGrid.get_slice_parallel_group)NN)rXrYrZr[rr|r~rr{rrrrrrrrrrrrrrrrrrr rps, ] rpN) deepspeedrrr collectionsr itertoolsrrrrarcrmrprrrr s   N