o }oiy @sddlZddlZddlZddlmZddlmZddlmZm Z m Z m Z m Z m Z mZmZmZddlZddlmZddlmZddlmZdZerTdd lmZdd lmZGd d d eZ d;dedededdddf ddZdded(d)de e)ej*ffd*d+Z+d?d-e e)e fd.e,ddfd/d0Z-d@d1ej*d2e,fd3d4Z.   5dAd6ee d7ee d8e/fd9d:Z0dS)BN) defaultdict)contextmanager) TYPE_CHECKINGAnyCallableDict GeneratorMappingOptionalProtocolTypeVar)nn)"initialize_model_parallel_for_nemo)logging.NEMO_MEGATRON_MODEL_PARALLEL_APPSTATE_OVERRIDE) Optimizable)ModelParallelConfigc@seZdZdZdddZdS)SharedStateDictProtocol cCsdS)rN)selfprefixrrP/home/ubuntu/.local/lib/python3.10/site-packages/nemo/lightning/_strategy_lib.pysharded_state_dict'sz*SharedStateDictProtocol.sharded_state_dictN)r__name__ __module__ __qualname____doc__rrrrrr$srF world_size global_rank local_rankparallel_configrreturnc Csddlm}|}tjtddkr!|j|j}|j } |j } n|}|jp'd} || kr7t d|d| d|} |} t d%id |d | d | d |jd |j d|jd|jd|jd|jd|jdt|dddt|ddd|dt|ddd|dt|ddot|dddkdt|dddt|ddd t|d dd!t|d!dd"t|d"dd#t|d#d$dSdt|dddt|ddd t|d dd!t|d!dd"t|d"dd#t|d#d$dS)&aw Initializes the parallel ranks for distributed training. This function sets up the parallel ranks based on the provided world size, global rank, local rank, and parallel configuration. It also sets the seed for random number generation and determines whether to use fp8 precision. Args: world_size (int): The total number of processes participating in the distributed training. global_rank (int): The rank of the current process in the distributed training setup. local_rank (int): The rank of the current process within its machine. parallel_config (ModelParallelConfig): The configuration object containing settings for model parallelism. seed (int, optional): The seed for random number generation. Defaults to 1234. fp8 (bool, optional): Whether to use fp8 precision for model parameters. Defaults to False. rAppStatefalsetruezExpected world_size (z-) to be greater than/equal to pipeline size ()r!r"r#tensor_model_parallel_sizeexpert_model_parallel_sizeexpert_tensor_parallel_sizepipeline_model_parallel_size$pipeline_model_parallel_comm_backend$virtual_pipeline_model_parallel_sizecontext_parallel_size"encoder_tensor_model_parallel_size$encoder_pipeline_model_parallel_sizeseed"pipeline_model_parallel_split_rankNuse_fp8init_mpi_proc_grouptp_comm_overlapFtp_comm_bootstrap_backendmpiuse_te_rng_tracker use_sharpuse_tp_pp_dp_mapping#num_distributed_optimizer_instancesnccl_communicator_config_pathuse_gloo_process_groupsTr) nemo.utilsr'osenvirongetrlowerr,r/r"r# ValueErrorrr-r.r0r1r2getattr) r!r"r#r$r5fp8r' app_stateinit_world_sizeinit_global_rankinit_local_rankpprrrinit_parallel_ranks,s                      rOmodelcCsddl}ddlm}ddlm}|}|jdur||jr|j |j |j |j |j |j|j|j|j|j|j|j|jr>dnd|j|j|jd|j|ksQJ|j|ksZJ|j|kscJ||_ |!|_"|#|_$|%|_&|'|_(|j)r|jj*dd dSdSdSdS) zBInitializes Megatron-LM model parallel if using model parallelism.rNparallel_stater&ztp-cp-ep-pp-dpztp-cp-ep-dp-pp)r,r/r1r6r0r4r3r2r-r.r=orderr?r@create_gloo_process_groupsr;)backend)+torch.distributed megatron.corerRrBr'model_parallel_sizedestroy_model_parallel distributedis_initializedinitialize_model_parallelr,r/r1r6r0r4r3r2r-r.r=r>r?r@rAtensor_model_parallel_rankget_tensor_model_parallel_rankpipeline_model_parallel_rank get_pipeline_model_parallel_rankexpert_tensor_parallel_rankget_expert_tensor_parallel_rankget_tensor_model_parallel_grouptensor_model_parallel_groupget_data_parallel_groupdata_parallel_groupget_data_parallel_rankdata_parallel_rankget_data_parallel_world_sizedata_parallel_size!get_pipeline_model_parallel_grouppipeline_model_parallel_groupr8 new_group)rPtorchrRr'rJrrrinit_model_parallelosJ          rocCsddlm}tt|dd|}|rNt|drN|j}tddt|D] }t||s*q"t||t||t|drBt|j |t||q"t|d rL| |SdS) rr)TransformerConfigconfigNconfigure_modelcSs |d S)N__) startswith)xrrrs z/set_model_parallel_attributes..__io__ __post_init__) ,megatron.core.transformer.transformer_configrp isinstancerHhasattrrqfilterdirsetattrrwrx)rP parallelismrphas_mcore_configrq attr_namerrrset_model_parallel_attributess    r)NNNc#szddlm}|jfdd}||_Wn tyYnw|j}|j}d|_d|_dVz ddlm}|_Wn tyBYnw||_||_dS)rr)transformer_enginecs|}d|d<|S)zForces device to metametadevicer)ckwargsoriginalrr_get_extra_te_kwargs_metasz=megatron_lazy_init_context.._get_extra_te_kwargs_metaFTN)megatron.core.extensionsr_get_extra_te_kwargs ImportErrorperform_initializationuse_cpu_initialization)rq_ter_orig_perform_initialization_orig_use_cpu_initializationrrrmegatron_lazy_init_contexts,        rccs|j}d|_dV||_dS)rTN)r)rqrrrrmegatron_cpu_init_contexts  rModelT)boundcs\eZdZdZ      dfdd Zfd d Zd d ZdddZddZddZ Z S) GradScalerz Gradient sclaer for model-parallel inf check. The inf in gradients are checked across tensor-parallel ranks in (1) executing optimizer step and (2) gradient scaler update. @@?Tr*cs.tj|||||dd|_||_|j|_dS)N) init_scale growth_factorbackoff_factorgrowth_intervalenabled)super__init__optimizer_update_skipped hysteresis_hysteresis_tracker)rrrrrrr __class__rrrs  zGradScaler.__init__cs*t|ddr |j|Stj|g|RS)N_custom_amp_unscale_gradsF)rH unscale_gradsr_unscale_grads_)r optimizerargsrrrr s  zGradScaler._unscale_grads_cOsddlm}d}tjtdd|dDg}tjj|tjj j | d| dkr;|j |i|}d|_|Sd|_|S) NrrQcss|]}|VqdSN)item).0vrrr z-GradScaler._maybe_opt_step..found_inf_per_deviceopgroupFT)rWrRrncuda FloatTensorsumvaluesrZ all_reduceReduceOpMAXget_model_parallel_grouprstepr)rroptimizer_staterrrRretval found_infrrr_maybe_opt_steps $ zGradScaler._maybe_opt_stepNc sddlm}|js dS|d\}|durIt|tr"|j|nd}t|tj j s/J|| dks9J||j dusBJ||j |nfdd |jD}t|dks_Jd |d}tjj|tjjj|d t|dkrtdt|D]}||}tjj|tjjj|d ||7}q~|dkr|jd8_|jdkrt|||j|j|jn|d nt|||j|j|j|j|_ttj jjj |_dS) z Updates to native grad scaler update function. 1. Check inf across model-parallel ranks. 2. Update hysteresis tracker. 3. Apply hysteresis to grad scale update. rrQNupdatez[new_scale should be a float or a 1-element torch.cuda.FloatTensor with requires_grad=False.r*Fcs.g|]}|dD] }|jjddq qS)rT)r non_blocking)rtor)rstater_scalerr Bs z%GradScaler.update..z,No inf checks were recorded prior to update.rg)!rWrR_enabled_check_scale_growth_trackerrzfloatrfill_rnrrnumel requires_gradcopy__per_optimizer_statesrlenrZrrrrranger_amp_update_scale__growth_factor_backoff_factor_growth_intervalrramp grad_scaler_refresh_per_optimizer_state) r new_scalerR_growth_trackerreason found_infsfound_inf_combinedirrrrr%sp         zGradScaler.updatecCs,|jr||j|j|j||jdSiS)zM Add hysteresis_tracker to the native functions' state_dict. )scalerrrrr)r get_scalerrr_get_growth_trackerr)rrrr state_dicts  zGradScaler.state_dictcCs|jsdSt|dkrtd|d|_|jdur!|j|d|d|_|d|_|d|_|d|_ |j durB|j |dd |vrM|d |_ dSd |_ dS) z_ Load hysteresis_tracker in addition to the state dict of the native function. NrzeThe source state dict is empty, possibly because it was saved from a disabled instance of GradScaler.rrrrr_hysterisis_trackerr*) rr RuntimeError _init_scalerrrrr_init_growth_trackerrr)rrrrrload_state_dicts$         zGradScaler.load_state_dict)rrrrTr*r) rrrrrrrrrr __classcell__rrrrrs  \rcCstdd}|durnz t|dd}Wn ty d}Ynwz t|dd}Wn ty7d}Ynw|dksD|dkrj|dkrltjdtjdtj d tj d tj d tj d dSdSdS dS) zAThese optimizations are present in NVIDIA NGC PyTorch Containers.NVIDIA_PYTORCH_VERSIONN.rr* TF) rCgetenvintsplit Exceptionrn_C_jit_set_profiling_executor_jit_set_profiling_mode_jit_override_can_fuse_on_cpu_jit_override_can_fuse_on_gpu_jit_set_texpr_fuser_enabled%_debug_set_autodiff_subgraph_inlining)nvidia_torch_versionNVIDIA_TORCH_MAJORNVIDIA_TORCH_MINORrrrenable_nvidia_optimizationss*         rfully_sharded_model_spacerrc s8ddlm}mm}ddlm}ddlm}|}dd| D}t |dr0|j|||dSt ||sT||| } ||t jd d |jDd || | S| } ||t jd d |jDd t| d t| ddksxJfddfddt| d | ddD| d <|| d| S)aR Sharded state dictionary for an MainParamsOptimizerWrapper. Used to save and load the optimizer state when training with distributed_checkpoint. Returns ------- dict: The sharded state dictionary for the optimizer Raises: ValueError: If a parameter ID does not match any model sharded parameter. r)!get_param_id_to_sharded_param_mapmake_sharded_optimizer_tensoroptim_state_to_sharding_state)MainParamsOptimizerWrapper)init_optimizer_statescSs i|] \}}|ds||qS) _extra_stateendswith)rkeyvaluerrr s z0optimizer_sharded_state_dict..r) is_loading sharding_typecss|]}|dVqdS)paramsNrrgrrrrrz/optimizer_sharded_state_dict..)model_sharded_state_dictoptim_params_itercss|]}|VqdSrrrrrrrsfp32_from_fp16_paramsr param_groupsc s6z|WSty}z td|d|d}~ww)Nz Param id z' does not match any model sharded param)KeyErrorrG)param_ide)id_to_sharded_param_maprrget_safes  z.optimizer_sharded_state_dict..get_safecs.g|]\}}fddt|d|DqS)cs"g|] \}}||ddqS)zoptimizer.state.fp32_param)rr)rr fp32_paramrrrrr sz;optimizer_sharded_state_dict...r )zip)r fp32_group state_grouprrrr s  z0optimizer_sharded_state_dict..)*megatron.core.dist_checkpointing.optimizerrrrnemo.core.optimrnemo.core.optim.optimizersrritemsr{rzr itertoolschain from_iterablerfloat16_groupsrr) rPrr r rrrrroptimizer_state_dictr)rrrroptimizer_sharded_state_dictsF           r'T checkpointstrictc CsTddlm}ddlm}m}|dur't|ts'||}|j|j|j g}||v}z ddl m }d}Wn t p6t y>d}Ynwt|D]\} } |durcd|vr[|dd | } n|d | } n d|vrl|d} n|} d} |j} t| d r| j} | d 7} t| d sxi}| D]K\}}d|}}|d r|td d}|d 7}|d s|| krd | |}||||<q|| krd || }|||t|d<q|||<q|rt| d rt| j|r| jj||d qCz | j||d WqCty'}z%| j| dd \}}tdd|Drtd|dn|WYd}~qCd}~wwdS)rrrQ)StrictHandlingparse_strict_flagN)FullyShardedDataParallelTFrmodel_moduler*zmodule.)r)css|]}|dVqdS)rNr)rsrrrr^sz(load_model_state_dict..zZLoding checkpoint created with Transformer Engine version lower than 1.13. Missing layers z will be ignored.)rWrR+megatron.core.dist_checkpointing.validationr*r+rzboolASSUME_OK_UNEXPECTEDRAISE_UNEXPECTED RAISE_ALL%megatron.core.distributed.custom_fsdpr,rModuleNotFoundError enumerate.get_virtual_pipeline_model_parallel_world_sizer.r{r!rtrrrallrwarning)megatron_parallelr(r)rRr*r+strict_optionsr,have_custom_fsdpindexr.checkpoint_state_dict n_nesting mcore_model _state_dictrr count_keyto_add to_remover missing_keys expected_keysrrrload_model_state_dictsz             rIr  broadcastcCsddlm}|dkrM|}|s@tj|}tj|kr.d|vr.tj |ddStjdkr>tj ||dSdStjj |||ddSdS)a When pipeline parallelism is enabled, casts a tensor defined on the last pipeline stage to other ranks. Args: value (torch.Tensor): A tensor to be casted from the final pipeline stage of a pipeline parallelism group (e.g. loss). Note that this tensor should already be defined on the target rank(s) to fill with received data. broadcast (bool): When True, broadcasts value from the final pipeline stage rank to all ranks in its group. When False, only rank zero receives value from the final pipeline stage rank in its group. This mode exists to avoid slow one-to-many communication when not necessary. Defaults to False. rrQr*)rN) rWrR&get_pipeline_model_parallel_world_size%get_pipeline_model_parallel_last_rankrnrZget_process_group_ranksrkget_ranksendrecvrJ)r rJrRsrc_rankpp_ranksrrr_sync_from_last_pipeline_stagegs  rS?no_weight_decay_cond scale_lr_condlr_multcCs|ddlm}m}ddlm}ddlm}|} t||s%Jdt|Gddd|} dd |D} ||| |||| j d } | j D]M} d | vsLd | vrMqB| d | d<| d| d <| d | dd}| d| dd}|| d <|| d<d| vr| d| d <| d=d| vr| d| d<| d=qBt |j ddrt |j ddrdd |D}t|dkr|dn|}|D]}||j_q| | S)rr)OptimizerConfigget_megatron_optimizer)McoreDistributedOptimizerr&zExpected OptimizerConfig, got c@s eZdZdZ   dddZdS)z*setup_megatron_optimizer..McoreOptrNFrcSs@t|jjj}i}d|vr||d<|jj|fd|i|}|S)Nr r )inspect signaturemcore_optimizerr parameters)rrr&r r mcore_optimizer_sigdistrib_optim_kwargsrrrrrsz=setup_megatron_optimizer..McoreOpt.sharded_state_dict)NFrrrrrrMcoreOpts racSg|]}|jqSr)r.)rmrrrrz,setup_megatron_optimizer..)rUrVrWrA pre_lr_mult pre_mult_wdlr pre_mult_lr weight_decayrWrTwd_mult pre_wd_multoverlap_param_gatherFalign_param_gathercSrbr)start_param_sync)r model_chunkrrrrrdr*)megatron.core.optimizerrXrYrrZrBr'rztyperArrErH ddp_configrrqparam_sync_func)rPrqrUrVrWrXrYrZr'rJra ddp_modules mcore_optpgnew_lrnew_wdrsr.rrrsetup_megatron_optimizersN        ry)r Fr)r%N)Fr)T)F)NNrT)1r[r"rC collectionsr contextlibrtypingrrrrrr r r r rnr nemo.lightning.megatron_initrrBrr lightning.fabric.utilities.typesr#megatron.core.model_parallel_configrrrrOModulerorrrrrrrrstrTensorr'r1rIrSrryrrrrsv  ,       C2$  ?   QK#