o }oiM @s"ddlZddlZddlZddlZddlZddlZddlmZmZddl m Z ddl m Z ddl mZmZmZmZmZmZmZmZmZmZmZddlmZddlZddlmZddlm Z ddl!m"Z"dd l#m$Z$dd l%m&Z&dd l'm(Z(dd l)m*Z*dd l+m,Z,ddl-m.Z.ddl/m0Z0ddl1m2Z2ddl3m4Z4ddl5m6Z6m7Z7ddl8m9Z9ddl:m;Z;ddlm?Z?ddl@mAZAddlBmCZCmDZDddlBmEZFddlBmGZGmHZHmIZImJZJmKZKddlLmMZMmNZNddlOmPZPddlQmRZRddlSmTZTzddlUmVZVWneWy ddlXmVZVYnwdd lYmZZZdd!l[m\Z\m]Z]dd"l^m_Z_dd#l`maZadd$lbmcZcdd%ldmeZedd&lfmgZgmhZhdd'limjZjmkZkmlZlzdd(lmmnZndd)lompZpd*ZqWneWerfycd+ZqYnwzddlsZsd*ZtWneWerfyzd+ZtYnwz@dd,lumvZvmwZwdd-lxmyZydd.lzm{Z{dd/l|m}Z}dd0l~mZmZmZdd1lmZdd lmZZdd2lmZdd3lmZd*ZWneWerfyd+ZYnwzdd4lmZWneWerfyehd5dd4lmZYnwz dd6lmZmZd*ZWn eyd+ZYnwd7Zd8Zd9ed:Z ;d]ded?dfd@dAZGdBdCdCe6ZGdDdEdEeZd^dFejjdGed?edHfdIdJZd^dFejjdGed?edHfdKdLZGdMdNdNe7ZGdOdPdPeaZGdQdRdRe2ZGdSdTdTe4ZGdUdVdVejjjZGdWdXdXe2ZGdYdZdZe*ZGd[d\d\e$ZdS)_N) OrderedDict defaultdict)contextmanager)Path) AnyCallableDict GeneratorIteratorListLiteralMappingOptionalSizedUnion)TorchCheckpointIO)get_filesystem)_optimizer_to_device)TQDMProgressBar) _update_n)LightningOptimizer) _DataFetcher)ClusterEnvironment) CheckpointIO)_WrappingCheckpointIO)MixedPrecisionPlugin) FSDPPrecision) DDPStrategy FSDPStrategy) TrainerFn)Trainer) OmegaConf)ReduceOp) noop_hook)BackwardPrefetchFullStateDictConfig)FullyShardedDataParallel)MixedPrecisionOptimStateKeyTypeShardedStateDictConfigShardingStrategy StateDictType)FullOptimStateDictConfigShardedOptimStateDictConfig)transformer_auto_wrap_policy)DistributedDataParallel)is_global_rank_zero)_refresh_per_optimizer_state) Float16Module)AutocastTransformerLayerParallelTransformerLayer) utils_funcs)SaveRestoreConnector)MainParamsOptimizerWrapper)init_optimizer_states)AppStatelogging) ckpt_to_dirinject_model_parallel_rankuninject_model_parallel_rank)MegatronDistributedFusedAdam)McoreDistributedOptimizerTF)dist_checkpointingparallel_state)CheckpointingException)dict_list_map_outplace)LocalNonpersistentObject)!get_param_id_to_sharded_param_mapmake_sharded_optimizer_tensoroptim_state_to_sharding_state)¶m_is_not_tensor_parallel_duplicate)TransformerLayer)DistributedCheckpointIO)get_num_microbatcheszCMegatron num_microbatches_calculator not found, using Apex version.)restore_sharded_modelopt_statesave_sharded_modelopt_state.NEMO_MEGATRON_MODEL_PARALLEL_APPSTATE_OVERRIDEzIhttps://docs.nvidia.com/nemo-framework/user-guide/latest/knownissues.htmlz (1) To resolve this issue, try to set `model.dist_ckpt_load_strictness` to `log_all`. This setting enables loading older checkpoints. (2) For more details and troubleshooting guidance, please refer to the framework documentation: z. sharpnccl_communicator_config_pathdistributed_timeout_minutesreturncCst}|jdurttjrtj|j|j |j |j |j |j |||j|jr(dnd|j|d |jtks8J|jtksAJt|_t|_t|_t|_t|_|jrddl m!}ddl"}|j!#|d}||j!#dkrtjj$d d dSdSdSdSdS) zInitializes Megatron-LM model parallel if using model parallelism. Args: sharp: Apply SHARP to NCCL data-parallel communication. nccl_communicator_config_path: Path to the yaml NCCL communication process group config file. Nztp-cp-ep-pp-dpztp-cp-ep-dp-pp) tensor_model_parallel_sizepipeline_model_parallel_size$virtual_pipeline_model_parallel_size"pipeline_model_parallel_split_rank$pipeline_model_parallel_comm_backendcontext_parallel_sizerQ use_sharpexpert_model_parallel_sizeorder#num_distributed_optimizer_instancesrRr)versiontransformer_enginez1.9mpi)backend)%r9model_parallel_sizerAdestroy_model_paralleltorch distributedis_initializedinitialize_model_parallelrTrUrVrWrXrYr[use_tp_pp_dp_mappingr]tensor_model_parallel_rankget_tensor_model_parallel_rankpipeline_model_parallel_rank get_pipeline_model_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_groupinit_mpi_proc_groupimportlib.metadatar^ packagingVersion new_group)rPrQrR app_stater^ry te_versionr~\/home/ubuntu/.local/lib/python3.10/site-packages/nemo/collections/nlp/parts/nlp_overrides.pyinit_model_parallelsF         rcs&eZdZdZ       d?deeejdedee de dee d e d e d e e ee e ffd dffd d Zd@fdd ZdAdeded dffdd ZfddZdBddZ dCdee e fde e efdee d dfddZdDdee e fd!e d dfd"d#Zd$ed efd%d&Zd'ee e ffd(d)Zd*e e efd+ee e ffd,d-Zd*e e efd+ee e fd ee e ffd.d/ZdDd*e e efd0e d ee e ffd1d2Zdee e fd ee e ffd3d4Zde e efd dfd5d6Ze d7d8Z!e fd9d:Z"e d e fd;d<Z#e d e fd=d>Z$Z%S)ENLPDDPStrategya}DDP plugin for Pytorch Lightning. Needed to customize DDP for model parallel models. Args: no_ddp_communication_hook: Disable DDP communication hook when using AMP-O2 with FP32 gradient accumulation. nccl_communicator_config_path: Path to the yaml file with NCCL communicator options sharp: Apply SHARP to NCCL data-parallel communication. NFparallel_devicescluster_environment checkpoint_iono_ddp_communication_hookrQrPdist_ckpt_parallel_savekwargsrSc  sNtstdts tdtjd|||d|||_||_||_||_dS)NzApex was not found. Please see the NeMo README for installation instructions: https://github.com/NVIDIA/NeMo#megatron-gpt.megatron-core was not found. Please see the NeMo README for installation instructions: https://github.com/NVIDIA/NeMo#megatron-gpt.)rrrr~) HAVE_APEX ImportErrorHAVE_MEGATRON_COREsuper__init__rrQrP_dist_ckpt_parallel_save) selfrrrrrQrPrr __class__r~rrs&  zNLPDDPStrategy.__init__trainer pl.Trainercs^|jj}|tjkrt|dS|jdusJ|j||||j dus-JdS)z Override setup() of DDPStrategy to avoid _sync_module_states(self.model) during eval as it can cause PP > 1 to hang due to assumption in DDPStrategy class that the same model is replicated across GPUs N) statefnrFITTINGrsetup acceleratormodel_to_devicesetup_precision_pluginmodel)rr trainer_fnrr~rrs  zNLPDDPStrategy.setup global_rank world_sizecsJtts!t}|jdur#t|j|j|j dddSdSdS)zSet up distributed environment.N<)rR) rsetup_distributedrAmodel_parallel_is_initializedr9rbrrPrQ_timeout total_seconds)rrrr|rr~rrs    z NLPDDPStrategy.setup_distributedcst|jdr |jjst|jdr#|jjr#|jjr|j|j|_dSt}|jdurSt dt |jfdt j ddi|j|_|jrQd|j_|jdtdSdStdS) zOverride LightningModule ddp if using model parallel. Sets find_unused_parameters to False to use activation-checkpoint-recomputation. megatron_amp_O2with_distributed_adamNz&Configuring DDP for model parallelism. process_groupTwith_context_parallelF)hasattrrrruse_mcore_dist_optim setup_mcore_distributed_parallel_modelr9rbr:infor/rAro _ddp_kwargsrrequire_backward_grad_syncregister_comm_hookr#r configure_ddp)rr|rr~rrs.        zNLPDDPStrategy.configure_ddpcs>|jjdd}|j}dd|D}t|tr$|j||||jdSt|tr/|||St|tsSt || }t |t j dd|jDdt||St |j| }t |t j d d|jDdt|d t|d d ks|Jfd dfddt|d |d d D|d <t|d |S)ab 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. F)use_pl_optimizercSs i|] \}}|ds||qS) _extra_state)endswith).0keyvaluer~r~r Os z?NLPDDPStrategy.optimizer_sharded_state_dict..) is_loadingrcss|]}|dVqdS)paramsNr~rgr~r~r asz>NLPDDPStrategy.optimizer_sharded_state_dict..)model_sharded_state_dictoptim_params_itercss|]}|VqdSNr~rr~r~rrmsfp32_from_fp16_params optimizer param_groupsc s6z|WSty}z td|d|d}~ww)Nz Param id z' does not match any model sharded param)KeyError ValueError)param_ide)id_to_sharded_param_mapr~rget_safeus  z=NLPDDPStrategy.optimizer_sharded_state_dict..get_safecs,g|]\}}fddt|d|DqS)cs"g|] \}}t||ddqS)zoptimizer.state.fp32_paramprefix)rF)rr fp32_paramrr~r |szJNLPDDPStrategy.optimizer_sharded_state_dict...r)zip)r fp32_group state_grouprr~rr{s  z?NLPDDPStrategy.optimizer_sharded_state_dict..)lightning_module optimizerssharded_state_dictitems isinstancer?rr>r7r8 state_dictrE itertoolschain from_iterablerrGrfloat16_groupslenr)runsharded_optim_staterrroptimizer_state_dictr~)rrroptimizer_sharded_state_dict@sP             z+NLPDDPStrategy.optimizer_sharded_state_dict checkpointfilepathstorage_optionscCst}|jr`|s dn|dd}|r.t|ddksJd|j|ddd}|g|d<nd|d<tg|d <|jj|t||d t r\t |j d r^t |j t||jjd d dSdSdSt|}|jsl|jdkrw|jj|||d dSdS)a(PTL method which we override to accomodate distributed checkpoints and the legacy model parallel checkpoints. When using megatron core, the distributed checkpointing library expects save functions to be called on every rank and internally does the rank checking. Tinclude_optimizeroptimizer_stateszFCurrently only support checkpointing 1 distributed optimizer per time!r)rNrrget_model_module_listmodel.r)r9use_distributed_checkpointinggetrrrrsave_checkpointr; HAVE_MODELOPTrrrMrunwrapped_checkpoint_iosave_sharded_strategyr<is_global_zerorr)rrrrr|rsharded_optim_stater~r~rrs4      zNLPDDPStrategy.save_checkpointTstrictcCs|jrdSd}d}t|jdrd}|jj}n t|jdr"d}|jj}|durTt|ts0t|trTi}|dD]}| |d|dd}|d|||<q8||d<|jj |d|ddS) z!Load lightning module state dict.Nr enc_dec_modelr..module.r)r) rrrrrrr2MCoreFloat16Modulekeysreplaceload_state_dictrrr model_key model_attrnew_state_dictrnew_keyr~r~rload_model_state_dicts$   z$NLPDDPStrategy.load_model_state_dictckptcsLtjsJtjtjftjdtjdfdd}t||S)z4Ensure checkpoint tensors are on the correct device.cuda)indexcs*t|tjr|jr|jkr|}|Sr)rrdTensoris_cudadeviceto)tcur_devr~r _fix_devices z7NLPDDPStrategy._fix_tensors_device.._fix_device)rdrrf is_availablercurrent_devicerC)rrr r~rr_fix_tensors_devices"  z"NLPDDPStrategy._fix_tensors_devicercCs4d|ddvr|dddS|ddddS)z)Return the param groups in the state dictrrrrr~rrr~r~r_get_param_groups zNLPDDPStrategy._get_param_groupcheckpoint_pathrcCs\t|}|ddurdS|dddddurdS||}||}t|t|kS)z Check if the number of param groups in the checkpoint not match with the sharded_state_dict Returns: bool: True if the number of param groups does not match rNFrr)r@load_common_state_dictrrr)rrrcommon_state_dictmodel_param_groupscheckpoint_param_groupsr~r~r_check_param_groups_mismatchs   z+NLPDDPStrategy._check_param_groups_mismatchc Cs*t|}||}||}tdd|D}tdd|D}d}|re|setdtddt|Dd}|ra||t gddd |d d vrX||d d d d <n ||d d d <nt d |j j ||d} |durd | d d vr| d d d d  || S| d d d  || S)ad Try to fix the param groups in the checkpoint. This is to fix the bug that in 24.03, all checkpoints store EP param group regardless of using EP or not. This function makes sure all checkpoints are compatible for loading. Returns: sharded_state_dict: Loaded dictionary for the distributed load function cs|] }|ddVqdS is_expertFNrrparamr~r~rrz3NLPDDPStrategy._fix_param_groups..csrrrrr~r~rrrNzCurrently training the model without expert parallelism while restored checkpoint has EP params. Ignoring the EP params for restoring.css$|] \}}|ddr|VqdSrr)rrentryr~r~rrs"T)rrrrrrz+Cannot find expert param in the checkpoint.r)r@rranyr:warningnext enumerateinsertrDrrload_checkpointpop) rrrrrrmodel_has_expert_paramcheckpoint_has_expert_param expert_indexloaded_state_dictr~r~r_fix_param_groupss6  z NLPDDPStrategy._fix_param_groupsload_optimizer_statesc Cst|}t}|jr|||std|dtr)t|jdr)t|j |dd|j }i}||d<|r?|j ddg|d <| ||rL| ||}n"z |jj||d }Wntym}z |d t}t|d }~wwt|jd drz||}|St|}d} td| dD]} ||rntd|d| d| dt| dqtd|dtjt} |j|}t} | | } t d|j!d| dd| dd|S)a}PTL method which we override to integrate distributed checkpoints for model parallel models. In order to load distributed checkpoints we need to provide the sharded_state_dict to the distributed load function. We get the sharded_state_dict from self.lightning_module which makes it convenient to have the loading logic happen at the strategy level. z6Distributed checkpoints should be a directory. Found: rrrrrT)rrr Ncontinue_trainingF rCheckpoint at z not found. Fail count z out of  not found. Aborting training.z Global Checkpoint Load : Rank : z : Start time : z.3fz#s : Time spent in load_checkpoint: s)"rr9risdirrrrrrLrrrrr*rr$rB LOAD_ERRORgetattr#_integrate_original_checkpoint_datar<rangeexistsr:r timesleepFileNotFoundErrorrdr empty_cache monotonicrr)rrr+fsr|rrr error_messagemax_error_count error_count start_timeend_timedurationr~r~rr$'sf        zNLPDDPStrategy.load_checkpointcCs|jjj}|D] }|dvr||||<q d|ddvr2|dddd|dddd<|S|ddd|ddd<|S)z Ensures that model and optimizer weights are loaded from the checkpoint. All other metadata are reinitialized. )rrrrrr)rr_checkpoint_connectordump_checkpoint)rroriginal_checkpointrr~r~rr6ns$  z2NLPDDPStrategy._integrate_original_checkpoint_datacCsf|jr|jr|jt|dSdSt}t|}|js!|jdkr1t d||j|dSdS)z$Delete checkpoint saved at filepath.rRemoving checkpoint: N) rrrremove_checkpointr;r9r<rrr:rrrr|r~r~rrIsz NLPDDPStrategy.remove_checkpointcCsjtot|jt}t|jdo|jdu}|r&|s&tdt |jj |_ |s3|r3tdt |_ |S)znemo.collections.multimodal.modules.stable_diffusion.attentionr|MCoreTransformerLayerr3r4fsdp_wrap_moduleupdate functoolspartialr.r* FULL_SHARD HYBRID_SHARD SHARD_GRAD_OPlistrrvrnrpstate_dict_contextrQrPrtrr) rrtrurvrwrQrPrxryrr|fsdp_sharding_strategyrr~rrsJ       zNLPFSDPStrategy.__init__cCs|dvr tj}}}n |dvrtj}}}n|dkr$tj}}}ntd|d|dur6t|d}|dur@t|d}t|||dS)a Set FSDP mixed precision recipe. `param_dtype` sets the data type for computation in forward and backpropagation, and the parameter data type for optimizer execution is maintained in the full precision. `buffer_dtype` is only valid when a module has buffers by `register_buffer` method, which is not shared by FSDP. `reduce_dtype` sets gradient reduction data type. )z16-true16-mixed)z bf16-truersbf16 z-Was unable to infer precision type, received rN) param_dtype reduce_dtype buffer_dtype)rdfloat16bfloat16floatrr5torch_dtype_from_precisionr')rrwrurxrrrr~r~rr>s    z+NLPFSDPStrategy._set_mixed_precision_recipecstts+t}|jdksJd|jtjkr$|j dks$Jdt |j |j tj dd|jd<g|jd<tdkr~|jD];}t|dd rU|jd|qDtjtt|tjtjd }tjj|tjtd |d kr}|jd|qDd Sd S)z4 Overriding to set parallel states. rz*FSDP does not support pipeline parallelismzHFSDP hybrid sharding cannot be used when tensor_model_parallel_size > 1.Trrignored_statessequence_parallelFdtyperopgrouprN) rsetup_environmentrArr9rUrtr*rrTrrPrQror$get_tensor_model_parallel_world_sizer parametersr5appendrdtensorr]rHint8rr re all_reducer"MINrm)rr|pis_not_tp_duplicaterr~rr\s8        z!NLPFSDPStrategy.setup_environmentcCsh|jdusJ|jstdkrdnd}|j|j|d|j}Wd|S1s-wY|S)zN Store the model state dict in one of full or sharded format. NrTF)rd)rrvrArrrr)rrdrr~r~rlightning_module_state_dicts   z+NLPFSDPStrategy.lightning_module_state_dictrcCsPt|tr|j}||jt|j|}Wd|S1s!wY|S)zW Store the full optimizer state dict in one of full or sharded format. N)rr _optimizerrrrkoptim_state_dict)rrrr~r~roptimizer_states  zNLPFSDPStrategy.optimizer_statercCsd}d}t|jdrd}|jj}n t|jdrd}|jj}|durOt|ts+t|trOi}|dD]}||d|dd}|d|||<q3||d<| |j|j |dWddS1siwYdS)z-Load lightning module states from checkpoint.Nrrrrrr) rrrrrr2rrrrrrr~r~rrs$   "z%NLPFSDPStrategy.load_model_state_dictc Cs2dd}|d}t|j|D]\}}||jv||}tt|ddtrsz+tj |jdddt |t j |j}Wd n1sHwYt ||j}Wntyr}ztd |td WYd }~nd }~wwtj||j|d } || t||jWd n1swYqd S) zV Re-key the full optimizer state dict to sharded optimizer state dict cSs:|} d|vr |St|tsJd|t|d}q)NTrz"Fail to find optimizer state dict.r)rrQrr) opt_statetemp_opt_stater~r~r_get_osdsz;NLPFSDPStrategy.load_optimizer_state_dict.._get_osdrrrTF) writebackrdNz3Failed to load optimzier state dicts. Errored with r)rroptim)rrrrrrrr]rksummon_full_paramsrekey_optim_state_dictr( PARAM_NAMEshard_full_optim_state_dict Exceptionprintexitoptim_state_dict_to_loadrr root_device) rrrrrrtemp_osd rekeyed_osdr sharded_osdr~r~rload_optimizer_state_dicts< z)NLPFSDPStrategy.load_optimizer_state_dictrrcCsTt}t||jd}|js|jdkr|jj|||ddSdS|jj|||ddS)zStore checkpoints 1. In case of sharded checkpoint, all ranks store unique checkpoints. 2. In case of non-sharded checkpoint, all data-parallel rank 0 store checkpoints. fsdp_sharded_ckptrrN)r9r<rvrrrr)rrrrr|r~r~rrs zNLPFSDPStrategy.save_checkpointrcCst|}t||jd}||std|dtjddlm }|t d|j |}Wd|S1srrr~r~rr$s    zNLPFSDPStrategy.load_checkpointcCsht}t||jd}|jrtd||j|dS|jdkr2td||j|dSdS)zRemove checkpointsrrHrN)r9r<rvr:rrrIrrrJr~r~rrIs z!NLPFSDPStrategy.remove_checkpointcCrS)zWhen loading FSDP-sharded checkpoint, need to restore checkpoint after configuring FSDP sharding to match FSDP-sharded format between the checkpoint and the current model and optimizer. Tr~rTr~r~rrU rVz.NLPFSDPStrategy.restore_checkpoint_after_setup)rrNFrsNFNNrSNr) rWrXrYrZr\rr]r[rsetrrrr'rrrrdr Optimizerrrr rrrrr$rIr^rUr_r~r~rrrqsv      G    &  *   "rqcseZdZdZdfdd Zdeffdd Zd d Zdfd d Z     ddede e e efde e j dedededede effdd ZZS)NLPSaveRestoreConnectorz8Custom connector to support saving and restoring states.rSNcs*tstdtstdtdS)NzApex was not found. Please see the NeMo README for installation instructions: https://github.com/NVIDIA/apex Megatron-based models require Apex to function correctly.rrr:r rrrrTrr~rrs z NLPSaveRestoreConnector.__init__ save_pathc st}|jddr d}n t|do|du}d}|jdur$|jdks'|rtj|}|rt tj ||j }t sYdd}|jjjdurS|jjjj||jd|jj|}tj|jdd } | ||trt|d rt| tr| j} t| tswt||| jd d n8|jd kr|jdkrtj |d|jdd|j } ntj |d|jdd|j dd|j } |!|"| t#j$ rt#j$%|rt&} n|j d ko|jd ko|jd k} | rt'(} |rt)*t+|| n|jdkr3t,|j-D]1} t.tj | d| dtj |d| dd|j } t)*| tj | d| d|j qnLt/0t,|j-t,|jD]?\} }t.tj | d| dd|dtj |d| dd|dd|j } t)*| tj | d| dd|d|j q?tj | |j1}|j2|dt|dr|j3dur|j4|| d|j5||d|j6r|7|| ntj|}t8| D]}t)*tj | ||qWdn 1swYt#j$ rt#j$%dSdSt9:||S)zSave model to save path.fsdpFrNrcSdSrr~r~r~r~rdummy?z.NLPSaveRestoreConnector.save_to..dummyr) async_saverrrrmp_rank_02d_tp_rank_ _pp_rank_03d)path2yaml_file artifacts)nemo_file_folder);r9rLrrrrbospathdirnamer;joinmodel_weights_ckptrArfrstrategylauncherlaunchrrJrKrrrrrrMrrrrrUrirk_save_state_dict_to_diskrrdrebarrierr0tempfileTemporaryDirectoryshutilmover\r7rTmakedirsrproductmodel_config_yamlto_config_filer_handle_artifacts_update_artifact_pathspack_nemo_file_make_nemo_file_from_folderlistdirrsave_to)rrrr| dist_ckpt dist_ckpt_dirdir_namerrrmp_model_weightsshould_move_datatmpdirtp_rankpp_rank config_yaml folder_pathfilerr~rr(s              $   0zNLPSaveRestoreConnector.save_toc Cs|ddr)i}|D]}|dd}|dd}|dd}||||<q |}|d drGi}|D]}|d d d }||||<q5|}i}|D]}|d dd }||||<qM|}|dds}i}|D]}|ddd }||||<qk|}|ddkri}|D] }|dd}|dd}|dd}|dd}||||<q|}|}d|vri}dd}|D]:}|dkrq||r|d} t| d } d| d!d tt| d g| d"g} |||| <q||||<q|}|d#rD|d#d$dkrDi}|D].}d%|vrq|d&d'}|d(d)}|d*d+}|d,d-}|d.d/}||||<q|}|S)0zRemap keys in state dict.megatron_legacyFzbert_model.language_modelzbert_model.model.language_model transformerencoderz .attention.z.self_attention.rrz model.module.rz6word_embeddings.adapter_layer.mm_linear_adapter.linearz?word_embeddings.adapter_layer.mm_projector_adapter.mm_projectorinductorz ._orig_modtargetzbnemo.collections.multimodal.models.text_to_image.stable_diffusion.ldm.ddpm.MegatronLatentDiffusionunetzmodel.diffusion_modelvaefirst_stage_model text_encodercond_stage_modelz.noise_schedulerz?model.model.diffusion_model.input_blocks.1.0.in_layers.2.weightc Ssd}gd}|D]\}dD]W}dD]R}dD]M}|dkr=dD]}||d|d|d|d| }||kr;d Sqqtd D]}||d|d |d|d| }||kr`d SqAqqq qd S) Nzmodel.model.diffusion_model.) input_blocks middle_block output_blocks) in_layers out_layers))weightbiasr )rr rT z.0.F)r7) rbase_strblocksblock layer_typerrnumtemplater~r~rshould_processs.$ $ zANLPSaveRestoreConnector.modify_state_dict..should_processzEmodel.cond_stage_model.transformer.text_model.embeddings.position_idsrN unet_config use_te_fp8 extra_stateznorm_to_q.layer_norm_znorm.znorm_to_q.weightz to_q.weightzff.net.layer_norm_z ff.net.0.z ff.net.fc1_zff.net.1.proj.z ff.net.fc2_z ff.net.3.)rrrsplitr]rr\) rconfrrrr loaded_keysrkey_r2idxnew_key_r~r~rmodify_state_dicts                  ."       z)NLPSaveRestoreConnector.modify_state_dictcsNt|}tj|rt||Stjtj|drdStd|d)Nr Expected z to be a file or directory.) r=rrisfiler_load_state_dict_from_diskr3splitextr)r model_weights map_locationuninject_model_weightsrr~rr's  z2NLPSaveRestoreConnector._load_state_dict_from_diskTF restore_pathoverride_config_pathr*r return_configrvalidate_access_integrityreplace_sharded_tensor_keyc  st||||||||} t| tr|dur| S| \} } } | durts=dd}|jjdur8|jjj||d|j t }|j dur\t j|j r\td|j d|j }nd}|rddd }|j||d }|j|||d t j||j}t j|d }t j|sJd |dtrt| drt| |ddi}| }||d<| r|dD]}t|dr|jd| |_qt | }|j!||||d}| "|t| dr| #Wdn1swYn|$| | } t%| | |td| j&j'd|d| S)ay Restores model instance (weights and configuration) into .nemo file Args: restore_path: path to .nemo file from which model should be instantiated override_config_path: path to a yaml config that will override the internal config file or an OmegaConf / DictConfig object representing the model config. map_location: Optional torch.device() to map the instantiated model to a device. By default (None), it will select a GPU if available, falling back to CPU otherwise. strict: Passed to load_state_dict. By default True return_config: If set to true, will return just the underlying config of the restored model as an OmegaConf DictConfig object without instantiating the model. Example: ``` model = nemo.collections.nlp.models.TextClassification.restore_from('asr.nemo') assert isinstance(model, nemo.collections.nlp.models.TextClassification) ``` Returns: An instance of type cls or its underlying config (if return_config is set). TNcSrrr~r~r~r~rrarz3NLPSaveRestoreConnector.restore_from..dummyrz9Restoration will occur within pre-extracted directory : `z`.cSsd|vS)Nz.yamlr~)namer~r~rwsz6NLPSaveRestoreConnector.restore_from..) filter_fn) path2file out_foldermembersrr%z to be a directory.rrrrrr)rrr/"setup_transformer_engine_tp_groupszModel z was successfully restored from r)(rload_config_and_state_dictrtuplerArfrrrrrrmodel_extracted_dirrrr3r:r_filtered_tar_info_unpack_nemo_filerrr(rrrLrrvaluesrrrJrKr$on_load_checkpointr7r$load_instance_with_state_dictrrW)r calling_clsr,r-r*rr.rr/r0 loaded_paramsrinstancerrrr3r6tmp_model_weights_ckpttmp_model_weights_dirrrvrrr~r restore_from)s%            4z$NLPSaveRestoreConnector.restore_fromrr)NNTFNTN)rWrXrYrZrr\rr$r'rrr!rdrr[r rFr_r~r~rrrs>r  rc sZeZdZdZ d deddedeejj j ddffdd Z e de d fd d ZZS)PipelineMixedPrecisionPlugintOverrides PTL autocasting to not wrap training/val/test_step. We do this because we have the megatron-core fwd/bwd functions in training_step. This means .backward is being called in training_step so we do not want the whole step wrapped in autocast. We instead wrap the fwd_output_and_loss_func that is passed to the megatron-core fwd/bwd functions. Nrwrrs16rrrscalerrScsn|dvrd}n|dvrd}ntd|dtj|||dd}|dvr)tj}n|dvr0tj}t|dS)NrrJrrrsrrsRprecision expected to be one of: ['16-mixed', '16', 16, 'bf16-mixed', 'bf16'] but  found)rK) RuntimeErrorrrrdrrset_autocast_gpu_dtype)rrwrrKplugin_precisionrrr~rrs z%PipelineMixedPrecisionPlugin.__init__recc dVdSz(Have the PTL context manager do nothing.Nr~rTr~r~rforward_context z,PipelineMixedPrecisionPlugin.forward_contextr)rWrXrYrZr r\rrdramp GradScalerrrr rUr_r~r~rrrGs  rGcsPeZdZdZ d deddedddffdd Zeded fd d Z Z S)FSDPMixedPrecisionPluginrHNrwrIrKrXrScs@|dvrd}n|dvrd}ntd|dtj||ddS)NrLrrMrsrNrO)rwrK)rPrr)rrwrKrRrr~rrsz!FSDPMixedPrecisionPlugin.__init__reccrSrTr~rTr~r~rrUrVz(FSDPMixedPrecisionPlugin.forward_contextr) rWrXrYrZr rrrr rUr_r~r~rrrYs rYcsheZdZdZ      dfdd Zfd d Zfd d ZddZdddZddZ ddZ Z S)rXz 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. @@?Trcs&tj|||||dd|_||_dS)N) init_scale growth_factorbackoff_factorgrowth_intervalenabled)rroptimizer_update_skipped hysteresis)rr^r_r`rarbrdrr~rrs  zGradScaler.__init__cs8t|trtj|jgtj|d|_dS|j|_dS)Nr)r_lazy_init_scale_growth_tracker HAVE_AMP_Crdrrdint32_hysteresis_tracker)rdevrr~rres  z*GradScaler._lazy_init_scale_growth_trackercs*t|ddr |j|Stj|g|RS)N_custom_amp_unscale_gradsF)r5 unscale_gradsr_unscale_grads_)rrargsrr~rrl s  zGradScaler._unscale_grads_cOsd}t|d}t|jddd}tjj|tjjjt d| |_ ||_ |j dkr<|j|i|}d|_|Sd|_|S)Nfound_inf_per_devicerT)dimkeepdimrF)r9r=rdstacksumrerr"MAXrAget_model_parallel_groupitem_found_infs_cpu_found_infs_cudasteprc)rrrrmrretval found_infs found_infr~r~r_maybe_opt_steps  zGradScaler._maybe_opt_stepNc s|jsdS|d\}|durCt|tr|j|nd}t|tjjs)J|| dks3J||j duss z%GradScaler.update..rz,No inf checks were recorded prior to update.rg)!_enabled_check_scale_growth_trackerrrrfill_rdr FloatTensornumel requires_gradcopy__per_optimizer_statesr=rrerr"rsrArtr7rfamp_Cupdate_scale_hysteresisrh_growth_factor_backoff_factor_growth_intervalrd_amp_update_scale_rr1)r new_scale_growth_trackerreasonrzfound_inf_combinedir{r~r~rr%sv       zGradScaler.updatecCs,|jr||j|j|j||jdSiS)zL Add hysteresis_tracker to the native functions' state_dict )scaler_r`rarrh)r get_scalerrr_get_growth_trackerrhrTr~r~rrs  zGradScaler.state_dictcCs|jsdSt|dkrtd|d|_|jdur!|j|d|d|_|d|_|d|_|d|_ |j durB|j |dd |vrM|d |_ dSt r\t jd gt jd 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.rr_r`rar_hysterisis_trackerrrr)rrrP _init_scalerrrrr_init_growth_trackerrrhrfrdrrgrr~r~rrs(         zGradScaler.load_state_dict)rZr[r\r]Trr) rWrXrYrZrrerlr|rrrr_r~r~rrrXs   ^rXc seZdZdZ ddeeefdedeej j j ddffdd Z d ej jd ed ejjfd egefd eddf ddZededfddZZS)MegatronHalfPrecisionPlugina Plugin for Half (FP16 and BF16) precision training. This plugin assumes the use of the optimizer with master parameters (fp32). This plugin uses half-precision at all operators in the model so need of input precision at each layer operator. Args: precision: Whether to use ``torch.float16`` (``16``) or ``torch.bfloat16`` (``'bf16'``). device: The device for ``torch.autocast``. scaler: An optional :class:`torch.cuda.amp.GradScaler` to use. NrwrrKrScs@t|||d}|dkrtj}n|dkrtj}t|dS)Nrrs)rrrdrrrQ)rrwrrKrrr~rrsz$MegatronHalfPrecisionPlugin.__init__rrzpl.LightningModuleclosurercKst|ts Jd|jdur&|jsJd|}||||jdi|S|jr-Jd|}||j|||||du}t|tj rO|j rO|s`|jj|fi||j dSdS)z5Run optimizer step and scale gradients, if necessary.zNMegatronHalfPrecisionPlugin supports only the optimizer with master parametersNz BF16 uses FP32 grad accumulationz FP16 uses FP16 grad accumulationr~) rr7rKfp32_grad_accumulation_after_closurerxcopy_model_grads_to_main_gradsunscale_plLightningModuleautomatic_optimizationr)rrrrrrclosure_resultskipped_backwardr~r~roptimizer_steps(    z*MegatronHalfPrecisionPlugin.optimizer_stepreccszdVWdSw)zHNo explicit precision casting. Inputs are supposed to be manually castedNr~rTr~r~rrUsz+MegatronHalfPrecisionPlugin.forward_contextr)rWrXrYrZrr\r]rrdrrWrXrrrnnModulerrrrr rUr_r~r~rrrs2      $rcsBeZdZdZd dededdffdd Zd eddfd d ZZ S)GlobalBatchDataFetcherz8Overrides PTL DataFetcher. Used to fetch global batches.rFprefetch_batchesstore_on_devicerSNcs0tstdtstdtj||ddS)NzKApex was not found. Using model parallel or megatron models will error out.zUMegatron-core was not found. Using model parallel or megatron models will error out..)rrr)rrrrr~rrs   zGlobalBatchDataFetcher.__init__iteratorcsn|}fddttD}|jd7_|js/|jr/|j}t|ts'J|jt |k|_ | ||dS)Ncsg|]}tqSr~)r!)rrrr~rrsz.r) on_fetch_startr7rKfetchedr_has_len dataloaderrrrdone on_fetch_end)rr start_outputbatchrr~rr_fetch_next_batchs z(GlobalBatchDataFetcher._fetch_next_batch)rF) rWrXrYrZr]r[rr rr_r~r~rrrs rcs8eZdZdZddZfddZddZdd ZZS) CustomProgressBarz Add CustomProgressBar to remove 's/it' and display progress per step instead of per microbatch for megatron models cCs|jjjjjjjjS)z7 Get the value of step within an epoch )fit_loop epoch_looproptim_progressrrxcurrent completed)rrr~r~rget_current_epoch_stepsz(CustomProgressBar.get_current_epoch_stepcst|_d|j_|jS)z8 Override bar_format to not have 's/it' zV{desc}: {percentage:3.0f}%|{bar}| {n_fmt}/{total_fmt} [{elapsed}<{remaining}{postfix}])rinit_train_tqdmbar bar_formatrTrr~rrs z!CustomProgressBar.init_train_tqdmcGs2|j}|j|d|j_|jd|jdS)zPOverride parent class on_train_epoch_start to initialize the progress bar state.rzEpoch N) max_stepstrain_progress_barresetinitialset_description current_epoch)rrrnum_training_batchesr~r~ron_train_epoch_start&s z&CustomProgressBar.on_train_epoch_startcOsB||}|||jjrt|j||j|||dSdS)z| Override parent class on_train_batch_end to update progress bar per global batch instead of per microbatch N)r_should_updatertotalr set_postfix get_metrics)rr pl_moduler__nr~r~ron_train_batch_end0s  z$CustomProgressBar.on_train_batch_end) rWrXrYrZrrrrr_r~r~rrrs   r)NrO)F)rrrrrr9 collectionsrr contextlibrpathlibrtypingrrrr r r r r rrrlightning.pytorchpytorchrrdlightning.fabric.pluginsr#lightning.fabric.utilities.cloud_ior$lightning.fabric.utilities.optimizerr$lightning.pytorch.callbacks.progressr2lightning.pytorch.callbacks.progress.tqdm_progressr lightning.pytorch.core.optimizerr lightning.pytorch.loops.fetchersrlightning.pytorch.pluginsr.lightning.pytorch.plugins.io.checkpoint_pluginr$lightning.pytorch.plugins.io.wrapperr#lightning.pytorch.plugins.precisionr(lightning.pytorch.plugins.precision.fsdprlightning.pytorch.strategiesrr lightning.pytorch.trainer.statesr!lightning.pytorch.trainer.trainerr omegaconfr!torch._C._distributed_c10dr";torch.distributed.algorithms.ddp_comm_hooks.debugging_hooksr#torch.distributed.fsdpr$r%r&rkr'r(r)r*r+torch.distributed.fsdp.apir,r-torch.distributed.fsdp.wrapr.torch.nn.parallelr/nemo.utils.get_rankr0torch.cuda.amp.grad_scalerr1rtorch.amp.grad_scaler3nemo.collections.nlp.modules.common.megatron.moduler28nemo.collections.nlp.modules.common.megatron.transformerr3r4nemo.collections.nlp.partsr5+nemo.core.connectors.save_restore_connectorr6nemo.core.optimr7nemo.core.optim.optimizersr8 nemo.utilsr9r:nemo.utils.model_utilsr;r<r= nemo.core.optim.distributed_adamr>nemo.core.optim.mcore_optimr?rModuleNotFoundErrorrrf megatron.corer@rA%megatron.core.dist_checkpointing.corerB+megatron.core.dist_checkpointing.dict_utilsrC(megatron.core.dist_checkpointing.mappingrD*megatron.core.dist_checkpointing.optimizerrErFrG$megatron.core.tensor_parallel.layersrH megatron.core.transformer.moduler+megatron.core.transformer.transformer_layerrIr!nemo.utils.callbacks.dist_ckpt_iorJr)megatron.core.num_microbatches_calculatorrKr (apex.transformer.pipeline_parallel.utilsmodelopt.torch.opt.pluginsrLrMrrrNURLr4r[r\r]rrr`rrrnrprqrrGrYrrWrXrrrr~r~r~rs  4                                      7 () IG