o }oil@sddlZddlZddlZddlmZmZmZmZddlZddl m Z ddl m Z ddl mZddlmZddlmZddlmZdd lmZdd lmZdd lmZmZdd lmZdd lmZm Z e!e"Z#GdddeeZ$dede%dee ge&fddfddZ'dS)N)AnyCallableDictOptional) CheckpointIO)_PATH)MCoreTensorAwareStateDict)TensorAwareStateDict)BaseCheckpointManager)LocalCheckpointManager)LazyCliqueReplicationStrategy)dict_list_map_inplace)HierarchicalCheckpointIOLocalCheckpointCallback)Trainer)AsyncCompatibleCheckpointIOAsyncFinalizableCheckpointIOcseZdZdZ    ddededeegefde d e d e e j jd e ffd d Zdee efdefddZ ddefddZZS)MCoreHierarchicalCheckpointIOaHierarchicalCheckpointIO implementation compatible with MCore distributed checkpointing. Args: wrapped_checkpoint_io (CheckpointIO): previously used checkpoint_io (for global checkpoints). local_ckpt_manager (BaseCheckpointManager): local checkpoint manager used to store the local checkpoints get_global_ckpt_iteration_fn (Callable[[_PATH], int]): a function that retrieves the iteration of a global checkpoint that will be compared with local checkpoint iteration in order to decide which to resume from. async_save (bool, optional): enables asynchronous save. Passed down to the local checkpoint manager unless overriden with `local_ckpt_options` in `_save_local_checkpoint`. If True, MCoreHierarchicalCheckpointIO must be wrapped with `AsyncFinalizableCheckpointIO` wrapper local_ckpt_algo (str, optional): local checkpoint save algorithm. See MCoreTensorAwareStateDict for details. By default, uses a fully parallel save and load algorithm ('fully_parallel`). parallelization_group (ProcessGroup, optional): save/load parallelization group allow_cache (bool, optional): if True, subsequent checkpoint saves will reuse the cached parallelization metadata. Ffully_parallelNwrapped_checkpoint_iolocal_ckpt_managerget_global_ckpt_iteration_fn async_savelocal_ckpt_algoparallelization_group allow_cachecs.t||||||_||_d|_||_dSN)super__init__rrcached_metadatar)selfrrrrrrr __class__U/home/ubuntu/.local/lib/python3.10/site-packages/nemo/lightning/pytorch/local_ckpt.pyr;s   z&MCoreHierarchicalCheckpointIO.__init__ checkpointreturncCs@tj||j|j|jd\}}dd}t||j|jrd|_|S)zSpecialized implementation using MCoreTensorAwareStateDict. Wraps the state dict in MCoreTensorAwareStateDict and makes sure that "common" state dict doesn't have any CUDA tensors. )algorrcSs4t|tjr|jjdkrtd|jddd}|S)NcpuzMoving CUDA tensor to CPUT) non_blocking) isinstancetorchTensordevicetypeloggerdebugto)xr#r#r$to_cpuXs zHMCoreHierarchicalCheckpointIO.to_tensor_aware_state_dict..to_cpuN)rfrom_state_dictrrrr commonr)r r%state_dict_for_save_r3r#r#r$to_tensor_aware_state_dictKs  z8MCoreHierarchicalCheckpointIO.to_tensor_aware_state_dicttensor_aware_checkpointcCsBt|tsJdt||durtd|j||j|jdS)z8Unwraps MCoreTensorAwareStateDict to a plain state dict.z)Unexpected tensor aware state dict type: NzEMCoreTensorAwareStateDict does not yet support the 'strict' argument.)r'r)r*rr.r/warning to_state_dictrr)r r9sharded_state_dictstrictr#r#r$from_tensor_aware_state_dictcs  z:MCoreHierarchicalCheckpointIO.from_tensor_aware_state_dict)FrNF)NN)__name__ __module__ __qualname____doc__rr rrintboolstrrr+ distributed ProcessGrouprrrr r8r> __classcell__r#r#r!r$r(s4  rtrainerlocal_checkpoint_base_dirrr&c Ks|j}tdd|D}|sdS|jj}t|jdd}|r+t|ts(Jt||j}|jdkr4t }nd}t t j |dt|d} t|| |fd|i|} |rVt| } | |j_dS) aUpdate the Trainer with the corresponding MCoreHierarchicalCheckpointIO if local checkpointing is used. Args: trainer (nl.Trainer): Trainer object to drive training loop. local_checkpoint_base_dir (str): Root directory under which to save local checkpoints. get_global_ckpt_iteration_fn (Callable): a function that retrieves the iteration of a global checkpoint that will be compared with local checkpoint iteration in order to decide which to resume from. **kwargs (dict): Additional kwargs passed to initialize MCoreHierarchicalCheckpointIO. Note: Async saving of local checkpoints is inferred based on what was configured on the strategy, if available. css|]}t|tVqdSr)r*r).0cbr#r#r$ sz5update_trainer_local_checkpoint_io..NrF local_ckpt) repl_strategy) callbacksanystrategy checkpoint_iogetattrr*rr. num_nodesr r ospathjoinsocket gethostnamer) rIrJrkwargsrQuse_local_ckptrTrrPrhierarchical_checkpointing_ior#r#r$"update_trainer_local_checkpoint_iots8  r_)(loggingrWrZtypingrrrrr+lightning_fabric.pluginsr lightning_fabric.utilities.typesr8megatron.core.dist_checkpointing.tensor_aware_state_dictr9nvidia_resiliency_ext.checkpointing.local.base_state_dictr Dnvidia_resiliency_ext.checkpointing.local.ckpt_managers.base_managerr Envidia_resiliency_ext.checkpointing.local.ckpt_managers.local_managerr @nvidia_resiliency_ext.checkpointing.local.replication.strategiesr 0nvidia_resiliency_ext.fault_tolerance.dict_utilsr >nvidia_resiliency_ext.ptl_resiliency.local_checkpoint_callbackrrnemo.lightning.pytorch.trainerr!nemo.utils.callbacks.dist_ckpt_iorr getLoggerr?r/rrErCr_r#r#r#r$s6          L