o Tiq@s*dZddlmZddlmZddlmZmZddl m Z ddl m Z ia iadadadadaiadZdrd d Zd d Zdadadadadadrd dZddZddZddZddZ ddZ!ddZ"ddZ#ddZ$dd Z%d!d"Z&d#d$Z'd%d&Z(d'd(Z)dsd*d+Z*  )dtd,d-Z+dsd.d/Z,d0d1Z-d2d3Z.d4d5Z/d6d7Z0d8d9Z1d:d;Z2dd?Z4d@dAZ5dBdCZ6dDdEZ7dFdGZ8dHdIZ9dJdKZ:dLdMZ;dNdOZdTdUZ?dVdWZ@dXdYZAdZd[ZBd\d]ZCd^d_ZDd`daZEdbdcZFdddeZGdfdgZHdhdiZIdjdkZJdldmZKdndoZLdpdqZMdS)ual Support different forms of parallelism in DeepSpeed using multiple process groups. Given that there are multiple scenarios and use-cases, this file is going to be updated frequently. For now, the group creation needed for the training scenario is being implemented. For inference and other new scenarios, the code will be either reused or added to this file. )comm)log_dist)$bwc_tensor_model_parallel_world_size bwc_pipeline_parallel_world_sizeDeprecatedException)get_acceleratorNcCstd)z3 Deprecated function. Retained to inform the users.zPlease do not use the groups.initialize() API as it is deprecated. Instead, pass the desired ep_size to deepspeed.moe.layer.MoE(..,ep_size,..)r)ep_sizempur J/home/ubuntu/.local/lib/python3.10/site-packages/deepspeed/utils/groups.py initialize5srcCs ||dksJd||dS)z6Ensure that numerator is divisible by the denominator.rz{} is not divisible by {}N)format) numerator denominatorr r r _ensure_divisibility<s rcCsTtdurdS|durt|}t||fd}|jdda|jddatattfS)z&Initialize model data parallel groups.N) data_paralleltensor_parallelrmesh_dimr)_TENSOR_MODEL_PARALLEL_GROUPdistget_world_sizeinitialize_mesh_device get_group_DATA_PARALLEL_GROUP_MODEL_PARALLEL_GROUP)tensor_model_parallel_sizedata_parallel_size mesh_devicer r r _init_tp_mesh_devicePs    r!cCtdusJdtS)z?Get the tensor model parallel group the caller rank belongs to.Nz3intra_layer_model parallel group is not initialized)rr r r r get_tensor_model_parallel_groupj r#cCr")z8Get the model parallel group the caller rank belongs to.Nz'model parallel group is not initialized)rr r r r get_model_parallel_grouprr$r%cCr")7Get the data parallel group the caller rank belongs to.Nz&data parallel group is not initialized)rr r r r get_data_parallel_groupzs r'cC|adS)z"Set the tensor model parallel sizeN)%_MPU_TENSOR_MODEL_PARALLEL_WORLD_SIZE) world_sizer r r $set_tensor_model_parallel_world_sizer+cCtdurtStjtdS)z6Return world size for the tensor model parallel group.Ngroup)r)rrr#r r r r $get_tensor_model_parallel_world_sizer0cCtSN)r0r r r r get_model_parallel_world_sizer4cCr()zSet tensor model parallel rank.N)_MPU_TENSOR_MODEL_PARALLEL_RANK)rankr r r set_tensor_model_parallel_rankr,r8cCr-)z3Return my rank for the tensor model parallel group.Nr.)r6rget_rankr#r r r r get_tensor_model_parallel_rankr1r:cCr2r3)r:r r r r get_model_parallel_rankr5r;cCst}t}|||S)zgCalculate the global rank corresponding to the first local rank in the tensor model parallel group.)rr9r0) global_ranklocal_world_sizer r r "get_tensor_model_parallel_src_ranks r>cCtjtdS).Return world size for the data parallel group.r.)rrr'r r r r get_data_parallel_world_sizerAcCr?+Return my rank for the data parallel group.r.)rr9r'r r r r get_data_parallel_rankrBrEc Cstd|dgdtsJt}t||}t||t}d}d}t|D]}t|||}t|}|||kr?|}q*t||D]}t|||d|}t|}|||kr`|}qF||fS)aA Initialize model data parallel groups. Arguments: model_parallel_size: number of GPUs used to parallelize model. Returns: Tuple of data parallel group and model parallel group Let's say we have a total of 8 GPUs denoted by g0 ... g7 and we use 2 GPUs to parallelize the model. The present function will create 4 model parallel groups and 2 data parallel groups as: 4 model parallel groups: [g0, g1], [g2, g3], [g4, g5], [g6, g7] 2 data parallel groups: [g0, g2, g4, g6], [g1, g3, g5, g7] Note that for efficiency, the caller should make sure adjacent ranks are on the same DGX box. For example if we are using 2 DGX-1 boxes with a total of 16 GPUs, rank 0 to 7 belong to the first box and ranks 8 to 15 belong to the second box. z(Creating model parallel group with size rranksNr ) rris_initializedrminrr9range new_group) model_parallel_size_r*model_parallel_sizer7rrirGr/r r r _create_model_parallels*         rOFc CstsJtd|dgdt}tdurdntt}t}||}t||d|}||}|tvrt d||D]@}t |D]9} |rXt || ||| d|} n t || |||} t | } td|dt | dg|| vr}| t|<qDq>|t vr|rt d||D].}t |D]'} t || |||} t | } td |dt | dg|| vr| t |<qqdSt ||D]*} t | || d|} t | } td |dt | dg|| vr| t |<qdSdS) a Create expert and data parallel groups. Note: Caller of this function is responsible to check if the groups already exist. Example - E + D parallel world_size = 16 expert_parallel_size = 2 # number of experts in same group expert_data_parallel_group = [0,2,4,6,8,10,12,14], [1,3,5,7,9,11,13,15] - all reduce is only on MoE params expert_parallel_group = [0, 1], [2,3], [4,5], [6,7], [8,9] - no all reduce, but all to all data_parallel_group = [0,1,...,15] - all reduce is only on non-MoE use_data_before_expert_parallel_ (bool): Use the D + E instead of E + D topology z3Creating expert and data parallel groups with size rrFNr ep_size_z2Creating expert data parallel process group named z with ranks: z-creating expert parallel process group named ) rrHrrr rr9r_EXPERT_DATA_PARALLEL_GROUPrJrKlist_EXPERT_PARALLEL_GROUP) expert_parallel_size_ use_data_before_expert_parallel_r* pp_world_sizer7 pp_stride group_name ep_stridepp_stage_startrNrGr/r r r _create_expert_and_data_parallelsp            r[c Csdt||||||}t||g}|}||}|r[||||} td||D]/} | |} t|D]$} |tt| D]} |dtt| | | || | |q@q4q*n!td||D]} | |} t|D]} |tt| | | |qkqag}g}|D]+}g}td||D] } ||| | |q||t|D] }|t|qq||fS)aGenerate expert parallel and expert data parallel group ranks list. Example - E + M + D parallel world_size = 16 model_degree = 2 expert_degree = 4 # number of experts in same group mp_group = [0, 1], [2,3], [4,5] ... data_parallel_group =[0,2,4,6,8,10, 12,14], [1,3,5,7,9,11,13,15] expert_parallel_group = [0,2,4,6], [8,10,12,14] [1,3,5,7], [9,11,13,15] expert_data_parallel_group = [0,8],[2,10],[4,12],[6,14], [1,9],[3,11],[5,13],[7,15] Args: world_size (int): Distributed world size. tensor_parallel_size_ (int): Tensor parallel group size. expert_parallel_size_ (int): Expert parallel group size. pipeline_parallel_size_ (int): Pipeline parallel group size use_data_before_expert_parallel_ (bool): Use the D + E instead of E + D topology Returns: Expert parallel group ranks and Expert data parallel group ranks list. r)rrJappendrRextendzip)r*tensor_parallel_size_rTpipeline_parallel_size_rU dp_world_sizedata_parallel_groups dp_group_sizerW dp_striderZ pp_stage_nextrNdsexpert_parallel_groupsexpert_data_parallel_groupsdp_rankspart_ep_groupsexpert_dp_ranksr r r _get_expert_parallel_ranks3sN         rmc CstsJdt|}|at}t}|}|dur dnt|}t||t||t d|d|d|d|d| d gd |}|t vr|t vrt |||||\} } | D]} t | } |t| vrm| t |<q\| D]} t | } |t| vr| t |<qpdSdSdS) a Create expert and data parallel groups based on MPU (model parallel) group. Note: Caller of this function is responsible to check if the groups already exist. Example - E + M + D parallel world_size = 16 model_degree = 2 expert_degree = 4 # number of experts in same group mp_group = [0, 1], [2,3], [4,5] ... data_parallel_group =[0,2,4,6,8,10, 12,14], [1,3,5,7,9,11,13,15] expert_parallel_group = [0,2,4,6], [8,10,12,14] [1,3,5,7], [9,11,13,15] expert_data_parallel_group = [0,8],[2,10],[4,12],[6,14], [1,9],[3,11],[5,13],[7,15] dist is not initializedNr z3Creating deepspeed groups with model parallel size z, pipeline parallel size z, expert parallel size z , world size z, dp world size rrP)rrHr!expert_tensor_parallel_world_sizerr9rArrrrQrSrmrKrR) rTr rUr`r*r7rbrVrXrhrirGr/r r r &_create_expert_data_and_model_parallelxsL        rpcCsVtdusJdg}tD]}d}|t|d|qt|dkr)t|SdS)z4Get the maximum ep_size from all the created groups.Nz&Warning! Process group not initialized_r)rSkeysr]intsplitlenmax)keylistkeyindexr r r _get_max_expert_sizes  r{cCs dtS)z+Get the name of the group with max. ep_sizerP)r{r r r r _get_max_expert_size_names r|cCs ttS)z!Get the max expert parallel size.)_get_expert_parallel_groupr|r r r r _get_max_expert_parallel_groups r~cC|tvsJdt|S)z9Get the expert parallel group the caller rank belongs to.z(expert parallel group is not initializedrSrXr r r r} r}cCtS)z#Get the expert parallel group dict.rr r r r _get_expert_parallel_group_dictrcCr)z>Get the expert data parallel group the caller rank belongs to.z-expert data parallel group is not initializedrQrr r r _get_expert_data_parallel_grouprrcCr)z(Get the expert data parallel group dict.rr r r r $_get_expert_data_parallel_group_dictrrcCs0tsJdtdurtjttdatS)aCreate a clone of the world group Note: We need to clone the dist world group because we use dist.get_global_rank() utility function in DeepSpeed at many places. As that function does not work on dist.group.WORLD, we need to keep a clone of it. rnNrF)rrH _WORLD_GROUPrKrJrr r r r _clone_world_groupsrcs\tsJdtt}|dkrAtdkrAtdks'Jdg}ttD]|q/tj|dtd<tS|dkr`tksOJdtjdd tDdtd<tStksjJd t|D]fd d tD}tj|dtd <qntD]g}t|D] }||qtj|dtd <qtS)Nrnrr z/num_gpus must >=1, cannot initialize All-To-AllrFlocal_0zCnum_gpus not equal to device per node, cannot initialize All-To-AllcSsg|]}|qSr r ).0rNr r r sz/_get_local_all_to_all_group..z(num_nodes<2 cannot initialize All-To-Allcsg|]}|qSr r )rjdevice_per_noderNr r rslocal_global_) rrHr device_countrrJr]rK_ALL_TO_ALL_GROUP) num_localcur_rank local_rankrr rr _get_local_all_to_all_groups8       rcCs:tsJdtdurtjddStdurtStS)r&rnNrr)rrHr rr r'rr r r r _get_data_parallel_groups  rcCsttdSNr)rget_global_rank!_get_sequence_data_parallel_groupr r r r _get_broadcast_src_ranksrcCstt|dSr)rrrrr r r _get_expert_broadcast_src_ranksrcCtjt|dS)z0Return world size for the expert parallel group.r.)rrr}rr r r _get_expert_parallel_world_sizercCr)z5Return world size for the expert data parallel group.r.)rrrrr r r $_get_expert_data_parallel_world_size!rrcCr)z-Return my rank for the expert parallel group.r.)rr9r}rr r r _get_expert_parallel_rank&rrcCst}t|}|||S)z^Calculate the global rank corresponding to a local rank zero in the expert parallel group.)rr9r)rXr<r=r r r _get_expert_parallel_src_rank+s rcCr)z2Return my rank for the expert data parallel group.r.)rr9rrr r r _get_expert_data_parallel_rank3rrcCs8tdur ttjddStdurtStjtdS)r@Nrrr.)r rrrr rArr r r r _get_data_parallel_world_size8s rcCstdurtSdS)/Return world size for the model parallel group.Nr )r r4r r r r _get_model_parallel_world_sizeBsrcCr?rC)rr9rr r r r _get_data_parallel_rankJrBrcCs8tdur ttjddStdurttdrtSdS)z2Return world size for the sequence parallel group.Nsequence_parallelr get_sequence_parallel_world_sizer )r rrrr hasattrrr r r r !_get_sequence_parallel_world_sizeOs rcCs8tdur ttdr tStdurttjddSdS)z/Return my rank for the sequence parallel group.Nget_sequence_parallel_rankrrr)r rrr rr9rr r r r _get_sequence_parallel_rankYs rcCs6tdus ttdstdurtdtjddStS)Nget_sequence_parallel_groupz No sequence parallel group foundrr)r rr KeyErrorrrr r r r _get_sequence_parallel_groupcs  rcC tdur ttdr tStS)rN%get_sequence_data_parallel_world_size)r rrrr r r r &_get_sequence_data_parallel_world_sizelrcCr)rDNget_sequence_data_parallel_rank)r rrrr r r r _get_sequence_data_parallel_ranktrrcCs tdur ttdr tStS)N get_sequence_data_parallel_group)r rrrr r r r r|srcCrr3)ror r r r %_get_expert_model_parallel_world_sizerrcCstsJtdusJdt}t}t||}t||t||D]}t|||d|}t|}|||kr@|aq&dS)a Create parameter partitioning group within ZeRO data parallel groups. Example - ZP + D parallel world_size = 16 zero_hpz_partition_size = 2 # number of ranks with replicated params (dual partitioning) zero_param_intra_parallel_group = [0, 1], [2,3], [4,5], [6,7], [8,9] - segmented (subgroup) with rep partition data_parallel_group = [0,1,...,15] - all reduce is on ZeRO model Nz:ZeRO parameter intra parallel group is already initializedr ) rrH _ZERO_PARAM_INTRA_PARALLEL_GROUPrr9rIrrJrK) group_sizer*r7zero_param_parallel_size_rNrGr/r r r !_create_zero_param_parallel_groups     rcCr)zTGet the ZeRO parameter partitioning intra parallel group the caller rank belongs to.)rr r r r $_get_zero_param_intra_parallel_groupsrcCstdur tdur dSdSdS)zQCheck if ZeRO data parallel with parameter partititioning groups are initialized.NF)rrr r r r #_zero_param_parallel_is_initializedsrcCr?)z;Return my rank for the ZeRO parameter inter parallel group.r.)rr9rr r r r ._get_zero_param_intra_parallel_rank_in_mygrouprBrcCr?)z8Return world size for the ZeRO parameter parallel group.r.)rrrr r r r /_get_zero_param_intra_parallel_group_world_sizerBrcCr?)z=Return all ranks for the ZeRO parameter intra parallel group.r.)rget_all_ranks_from_grouprr r r r *_get_zero_param_intra_parallel_group_ranksrBr)r N)F)r F)N__doc__ deepspeedrrdeepspeed.utilsrdeepspeed.utils.bwcrrdeepspeed.utils.exceptionsrdeepspeed.acceleratorrrSrQrrr rorr rrrrrr)r6r!r#r%r'r+r0r4r8r:r;r>rArErOr[rmrpr{r|r~r}rrrrrrrrrrrrrrrrrrrrrrrrrrrrrr r r r s       1J E6