o }oih@sddlZddlZddlZddlZddlZddlmZddlm Z ddl m Z m Z m Z mZmZddlmZddlZddlmZddlmZddlmZddlmZddlmZdd l m!Z!dd l"m#Z#dd l$m%Z%m&Z&m'Z'dd l(m)Z)dd l*m+Z+ddl,m-Z-m.Z.m/Z/m0Z0ddl1mZddl2m3Z3zddl4m5Z5ddl6m7Z7m8Z8Wne9yddl:m5Z5ddl;m7Z7m8Z8Ynwe3dddd\ZZ?e@eAZBGdddee+jCZDdS)N)contextmanager)Path)AnyDictLiteralOptionalUnion) CheckpointIO)rank_zero_info) reset_seed)ModelParallelStrategy) TrainerFn) STEP_OUTPUT)ColwiseParallelRowwiseParallelSequenceParallel)override)io)_destroy_dist_connection ckpt_to_dircreate_checkpoint_iofsdp2_strategy_parallelize)logging)safe_import_from)distribute_tensor) ReplicateShardztorch.distributed.fsdpMixedPrecisionPolicyz"torch.distributed._composable.fsdp)fallback_moduleCPUOffloadPolicycsheZdZdZddddddddeddf deedefdeedefd eed e d d d e dee e ee e effffdd Zeede fddZdGddZdGddZedGddZedejddffdd ZddZedGdd Zd!e fd"d#ZdHd$d%Zed&d'ZedHdefd(d)ZedHde fd*d+Z!edHdefd,d-Z"edHdefd.d/Z#e$edHd0ee fd1d2Z%eede&fd3d4Z'e'j(d5e&ddfd6d4Z'edefd7d8Z)ed9ee e*fddfd:d;Z+ede e e ffde e e fd9ee e*fd?ee ddfd@dAZ-edBe e*Bde e e ffdCdDZ.ee/j0 dIdEdFZ1Z2S)J FSDP2StrategyziFSDP2Strategy implementing FSDP via FSDP 2. Notes: - TP + FSDP2 is currently not supported. autoFNTdata_parallel_sizetensor_parallel_sizecontext_parallel_sizesequence_paralleloffload_policyruse_hf_tp_plancustom_tp_planc  sdtjd||d| ||_||_||_d|_||_|jdur4tdus(Jdtt j t j t j dd|_d|_ | |_ ||_ ||_| |_d|_| durU| |_tdddS|jr`tdd dSttd tttttttttd d } tttd dttttd d tttd d ttd tdd}|jr| || |_tddS)arInitializes the FSDP2Strategy with specified parallelization settings. Args: data_parallel_size (Union[Literal["auto"], int]): Size of data parallel. Defaults to "auto". tensor_parallel_size (Union[Literal["auto"], int]): Size of tensor parallel. Defaults to "auto". context_parallel_size (optional): Number of context-parallel groups. Defaults to 1. sequence_parallel (bool): Whether to enable sequence parallelism when use_hf_tp_plan is False and custom_tp_plan is not provided. Defaults to False. Only effective when tensor_parallel_size > 1. data_sampler (optional): Custom data sampler to process dataloaders. mp_policy (optional): Mixed precision policy for parameter and operation casting. Defaults to: ```python MixedPrecisionPolicy( param_dtype=torch.bfloat16, reduce_dtype=torch.float32, output_dtype=None, cast_forward_inputs=True, ) ``` parallelize_fn (callable, optional): Function for parallelizing the model. Defaults to None. use_hf_tp_plan (bool, optional): Whether to use the huggingface TP plan. This will be used if custom_tp_plan is not provided. Also, sequence_parallel option will be ignored if use_hf_tp_plan is set to True. Defaults to True. custom_tp_plan (Optional[Dict[str, Any]], optional): Custom tensor parallel plan for the model. tensor_parallel_size need to be > 1 to use this option. If provided, it overrides the default tensor parallel plan. sequence_parallel option will be ignored if custom_tp_plan is provided. **kwargs: Additional arguments for base class initialization. )r#r$Nz%Expected to have MixedPrecisionPolicyT) param_dtype reduce_dtype output_dtypecast_forward_inputszaYou are using a custom TP plan. Make sure it is compatible with the model. Parallelization would zYnot raise errors if the custom TP plan is not compatible. SP option will also be ignored.zZYou are using a huggingface TP plan. Make sure your model is a huggingface model. Certain zHparallelizations might not be supported. SP option will also be ignored.) input_layouts)output_layouts) model.embed_tokenszmodel.layers.*.self_attn.q_projzmodel.layers.*.self_attn.k_projzmodel.layers.*.self_attn.v_projmodel.layers.*.self_attn.o_projzmodel.layers.*.mlp.up_projzmodel.layers.*.mlp.gate_projmodel.layers.*.mlp.down_projlm_headr")r.r/)r0z model.normzmodel.layers.*.input_layernormr1z'model.layers.*.post_attention_layernormr2r3zaUsing default TP plan for parallelization. It is compatible with huggingface llama3-style models.)super__init___checkpoint_ior% data_sampler checkpoint mp_policyHAS_MIXED_PRECISION_POLICYrtorchbfloat16storeparallelize_fnr'r&r( tp_shard_planrinforrrrrupdate)selfr#r$r%r&r'r8 checkpoint_ior:r?r(r)kwargsbase_model_tp_planbase_model_sp_plan __class__r4d/home/ubuntu/.local/lib/python3.10/site-packages/nemo/lightning/pytorch/strategies/fsdp2_strategy.pyr6Fsn,       zFSDP2Strategy.__init__returncCsdS)z"Optim state restoration is enabledTr4rCr4r4rJlightning_restore_optimizersz)FSDP2Strategy.lightning_restore_optimizercC ||_dS)aStores a reference to the optimizer state-dict for later restoration. Instead of immediately restoring the optimizer's state, this method saves the checkpoint reference and defers the restoration until the first training step. This is necessary because, in NeMo 2.0, PeFT adapters are added dynamically just before the first training step. Attempting to restore the optimizer state-dict before the adapters are initialized would result in an error. Args: checkpoint (dict): A dictionary containing the trainer's checkpoint, including the optimizer state-dict. N)r9)rCr9r4r4rJload_optimizer_state_dicts z'FSDP2Strategy.load_optimizer_state_dictcCsRddlm}|jdur't|j|jdD] \}}||j||idqd|_dSdS)aRestores the optimizer state-dict from the stored checkpoint. This method applies the optimizer state stored in the checkpoint to the corresponding optimizers. It ensures that the optimizer states are correctly restored after the PeFT adapters have been added in the first training step. If no checkpoint is stored, the method exits without performing any restoration. Note: This operation runs only once, as the checkpoint reference is cleared after execution. r)set_optimizer_state_dictNoptimizer_states)optim_state_dictoptions)'torch.distributed.checkpoint.state_dictrPr9zip optimizerslightning_module)rCrP optimizer opt_stater4r4rJ_load_optimizer_state_dicts  z(FSDP2Strategy._load_optimizer_state_dictcCsddlm}|j|j||jdkr|j|_|jdkr#|j |_g}g}t |j|j |jggdD]\}}| t || |q4||jjt||d|_|jddkrc|jdjd d |j|j_d S) z-setup distributed environment and device meshr)init_device_meshr!) data_parallelcontext_paralleltensor_parallel) device_type mesh_shapemesh_dim_namesr]r"r\r]dp_cp) mesh_dim_nameN)torch.distributed.device_meshr[ accelerator setup_device root_device_setup_distributed_data_parallel_size num_nodes_tensor_parallel_size num_processesrUr%appendinttypetuple _device_meshsize_flattenrW)rCr[r`radimnamer4r4rJsetup_environments.      zFSDP2Strategy.setup_environmenttrainercsr||_|j|t|ddr||jdkr |j|j|_t|ddr5|j j t j kr7t |dSdSdS)zConfigures the strategy within the PyTorch Lightning trainer. Args: trainer (pl.Trainer): The PyTorch Lightning trainer instance. _init_model_parallelTr"_setup_optimizersN)rxrfsetupgetattr parallelizerj_lightning_moduletorhstatefnr FITTINGr5setup_optimizers)rCrxrHr4rJr{ s   zFSDP2Strategy.setupcCsF|jdur|j|jj|j|j|j|j|jdd|_dSt ddS)zApplies fully_shard on modelN) device_meshr:r(r@r'z"Called parallelize more than once.) r?rWmodelrrr:r(r@r'rwarningrLr4r4rJr}s  zFSDP2Strategy.parallelizec Cst|||_|jdurd|_|jdusJtjs$t dtj r0t ddS|j }|j}|jjtjd<t|jjtjd<t d|d|d d |tjj|j|||jd |jd krptttd d|jd|dd ddS)z-Initializes process group for communications.Nzcuda:nccl,cpu:gloozOtorch.distributed is not available. Cannot initialize distributed process groupz7torch.distributed is already initialized. Exiting early MASTER_ADDR MASTER_PORTz'Initializing distributed: GLOBAL_RANK: z , MEMBER: r"/)rank world_sizer>ncclzd----------------------------------------------------------------------------------------------------z distributed_backend=z5 All distributed processes registered. Starting with z processes  )r set_world_ranks_get_process_group_backend_process_group_backendr'cluster_environmentr< distributed is_available RuntimeErroris_initialized_loggerdebug global_rankr main_addressosenvironstr main_portrAinit_process_groupr>atexitregisterrr )rCrrr4r4rJri/s<           z FSDP2Strategy._setup_distributed step_typecCs4|ddfD]}t|j|rt|j|SqdS)aRetrieves the loss reduction method for a given step type. Args: step_type (str): The type of step (e.g., "training", "validation"). Returns: Callable: The loss reduction function, if defined; otherwise, None. _loss_reductionloss_reductionN)hasattrrWr|)rCrfn_namer4r4rJ_get_loss_reductionVs  z!FSDP2Strategy._get_loss_reductioncCsf|d}|j|jkr||j|j|||}n t|j|||}||}|r-|||S|d|ifS)aExecutes a training, validation, or test step and applies loss reduction if available. Args: step_type (str): The step type ("training", "validation", "test", "predict"). batch: The input batch. batch_idx (optional): Index of the batch. Returns: Tuple: The computed loss and a dictionary with reduced loss metrics. _stepavg)rrW_forward_redirectionr|rforward)rCrbatch batch_idx method_namelossrr4r4rJ _step_proxyds    zFSDP2Strategy._step_proxycCs|S)zreturns the sharded optim state) state_dict)rCrXr4r4rJoptimizer_statezszFSDP2Strategy.optimizer_statecCs|jdur ||jdusJ|jdusJ|jdkr&|jj||dd}n|j||}|jjd|jjdddd|jd|jj|S)zDefines the training step, logging relevant metrics. Args: batch: The input batch. batch_idx (optional): The index of the batch. Returns: STEP_OUTPUT: The loss for backpropagation. Nr"T)r] global_step)prog_barrank_zero_only batch_sizestep) r9rZrWrr% training_steplogrxr)rCrrrr4r4rJrs&  zFSDP2Strategy.training_stepcCs|jdusJ|jdusJ|j#|d||\}}|dr-|jjd|dddd|WdS1s9wYdS)zDefines the validation step, logging validation loss. Args: batch: The input batch. batch_idx (optional): The index of the batch. Returns: Any: The validation loss. N validationrval_lossTr"rr)rWrprecision_pluginval_step_contextrrrCrrrreducedr4r4rJvalidation_steps  $zFSDP2Strategy.validation_stepcCsx|jdusJ|jdusJ|j|d||\}}|jjd|dddd|WdS1s5wYdS)zDefines the test step, logging test loss. Args: batch: The input batch. batch_idx (optional): The index of the batch. Returns: Any: The test loss. Ntest test_lossrTr"r)rWrrtest_step_contextrrrr4r4rJ test_steps  $zFSDP2Strategy.test_stepcCs`|jdusJ|jdusJ|j|d||\}}|WdS1s)wYdS)zRuns one predict step. Args: batch (dict): the batch to use for pred. batch_idx (int, optional): the batch index. Defaults to None. Returns: STEP_OUTPUT: the reduced loss. Npredict)rWrrpredict_step_contextrrr4r4rJ predict_steps   $zFSDP2Strategy.predict_step empty_initccs dVdS)z'Context manager used for initializationNr4)rCrr4r4rJtensor_init_contexts z!FSDP2Strategy.tensor_init_contextcCs|jdur t|_|jS)zVCheckpointIO getter Returns: CheckpointIO: _description_ N)r7rrLr4r4rJrDs zFSDP2Strategy.checkpoint_iorcCrN)zcCheckpointIO setter Args: io (CheckpointIO): the checkpointio to use. N)r7)rCrr4r4rJrDs cCs*t|jjjjjjjjj |jjjj j jj S)zsGets the current step within an epoch. Returns: int: The step index within the epoch. ) maxrxfit_loop epoch_loopautomatic_optimizationoptim_progressrXrcurrent completedmanual_optimizationoptim_step_progressrLr4r4rJcurrent_epoch_stepsz FSDP2Strategy.current_epoch_stepfilepathcCsJt|}|jr!tj|rt|dStj|r#t|dSdSdS)zRemoves a checkpoint from the filesystem. Args: filepath (Union[str, Path]): Path to the checkpoint to be removed. N) ris_global_zerorpathislinkunlinkexistsshutilrmtree)rCrckptr4r4rJremove_checkpoint s  zFSDP2Strategy.remove_checkpointc Csddlm}|jdus J|j}t|jdtt|jdidd}|r2dd}tt||}nt|}i}|D] }| |}||||<q,s z;FSDP2Strategy.lightning_module_state_dict..) 'nemo.lightning.pytorch.strategies.utilsrrWrr|r7listfilterkeyspopdistbarrier) rCrtmp_sdis_adapter_only name_has_lora module_namesrrvparamr4r4rJlightning_module_state_dicts$    z)FSDP2Strategy.lightning_module_state_dictr9storage_optionscCs |jr|jj|||ddSdS)z` Unshards FSDP2 checkpoint and passes it to checkpoint_io for saving to a file. )rN)rrDsave_checkpoint)rCr9rrr4r4rJr9szFSDP2Strategy.save_checkpointcheckpoint_pathcCs |j|S)z#Loads checkpoint with checkpoint_io)rDload_checkpoint)rCrr4r4rJrDs zFSDP2Strategy.load_checkpointcsjdkrjddkrfddjD}fddjD}fddjD}dd|d D}|D]l\}tfd d |Dr[t|jtd d t fd|<q;tfdd |Dryt|jtd d td d fd|<q;tfdd |Drt|jtd d tdd fd|<q;t|jdtd d fd|<q;n)jdkrňjddkrŇfdd|d D}n fdd|d D}j j ||ddS)zShards a full state dictr"r]c g|] }tj|tr|qSr4) isinstancer@r.0krLr4rJ S z7FSDP2Strategy.load_model_state_dict..crr4)rr@rrrLr4rJrTrcrr4)rr@rrrLr4rJrUrcSsi|]\}}||qSr4r4rrvr4r4rJ Wsz7FSDP2Strategy.load_model_state_dict..rc3|] }t|VqdSNrematchrrrr4rJ [z6FSDP2Strategy.load_model_state_dict..rru placementsc3rrrrrr4rJr]rc3rrrrrr4rJr_rr\c s2i|]\}}|t|jdttddfdqS)rbrrr )rrrrrrrLr4rJrjs c s.i|]\}}|t|jdtddfdqS)r\rrr )rrrrrrLr4rJrrs)strictN) rlrrrsr@itemsanyrrrrrWload_state_dict)rCrr  colwise_keys rowwise_keysseq_parallel_keys sharded_staterr4)rrCrJload_model_state_dictIs0"&&     z#FSDP2Strategy.load_model_state_dict)rKNr)F)3__name__ __module__ __qualname____doc__rrrrorboolrrrrrr6propertyrrMrOrZrwplTrainerr{r}rirrrrrrrrrrrr rDsetterrrrrrrr<no_gradr __classcell__r4r4rHrJr ?s  m   ' &  '   !    r )Err_loggingrrr contextlibrpathlibrtypingrrrrrlightning.pytorchpytorchrr<torch.distributedrrlightning.fabric.pluginsr $lightning.fabric.utilities.rank_zeror lightning.fabric.utilities.seedr +lightning.pytorch.strategies.model_parallelr PLModelParallelStrategy lightning.pytorch.trainer.statesr !lightning.pytorch.utilities.typesr!torch.distributed.tensor.parallelrrrtyping_extensionsrnemo.lightningrrrrrr nemo.utilsnemo.utils.import_utilsrtorch.distributed.tensor._apir(torch.distributed.tensor.placement_typesrr ImportErrortorch.distributed._tensor.api)torch.distributed._tensor.placement_typesrr;rHAS_CPU_OFFLOAD_POLICY getLoggerrrIOMixinr r4r4r4rJsL