o }oi;K@sddlZddlmZmZddlmZddlmZmZm Z m Z m Z m Z m Z ddlmZddlZddlmZddlmZddlmZddlmZmZdd lmZdd lmZmZdd l m!Z!dd l"m#Z#dd lm$Z$ddl%m&Z&m'Z'ddl(m)Z)m*Z*ddl+m,Z,ddl-m.Z.ddl/m0Z0zddl1m2Z2Wn e3yeZ2Ynwe dej4dZ5e de$j6dZ7eGddde.e e5Z8de e9efdefddZ:Gddde2e.Z;de de fd d!Zd$e e9efde e?fd%d&Z@dS)'N) dataclassfield)Path)AnyCallableDictGenericOptionalTypeVarUnion) CheckpointIO)get_filesystem)_PATH)!get_default_load_sharded_strategy!get_default_save_sharded_strategy)SaveShardedStrategy) FullyParallelLoadStrategyWrapper FullyParallelSaveStrategyWrapper)TorchDistSaveShardedStrategy)get_data_parallel_group)nn)Selfoverride) WEIGHTS_PATH ckpt_to_dir) IOProtocol)IOMixin)logging)AsyncCompatibleCheckpointIOLightningModuleT)boundModuleTc@sveZdZUdZeed<ejed<ee dZ e e e fed<edejdefddZedejde e e ffd d Zd S) TrainerContexta A context wrapper for a PyTorch Lightning Trainer and its associated model. This class ensures that both the trainer and its LightningModule extend `IOMixin` and provides additional context information. Attributes: model (LightningModuleT): The Lightning model associated with the trainer. trainer (pl.Trainer): The PyTorch Lightning trainer instance. extra (Dict[str, Any]): Additional context data, such as the `datamodule`, if available. modeltrainer)default_factoryextrareturncCsDt|ds tdtdt|jdstd|||j||dS)a Creates a `TrainerContext` instance from a given `pl.Trainer`. Ensures that the trainer and its associated LightningModule support the `IOMixin` interface. Args: trainer (pl.Trainer): A PyTorch Lightning Trainer instance. Returns: TrainerContext: A new instance containing the trainer, model, and extra context. Raises: ValueError: If the trainer or its LightningModule does not extend `IOMixin`. __io__zTrainer must be an instance of z#. Please use the Trainer from nemo.z$LightningModule must extend IOMixin.)r$r#r&)hasattr ValueErrorrlightning_moduleconstruct_extra)clsr$r.H/home/ubuntu/.local/lib/python3.10/site-packages/nemo/lightning/io/pl.py from_trainerHs  zTrainerContext.from_trainercCs*i}t|drt|jdr|jj|d<|S)as Constructs an `extra` dictionary containing additional relevant context. If the trainer has a `datamodule` that supports `IOMixin`, it will be added to `extra`. Args: trainer (pl.Trainer): A PyTorch Lightning Trainer instance. Returns: Dict[str, Any]: A dictionary containing extra context information. datamoduler()r)r1r()r-r$r&r.r.r/r,_s  zTrainerContext.construct_extraN)__name__ __module__ __qualname____doc__r__annotations__plTrainerrdictr&rstrr classmethodrr0r,r.r.r.r/r"6s  "r"filepathr'cCst|d}|}t|trJ|jdtkr!|t}|s|r!|}t|dr?|jjdtkr?|jt}|s9|r?|jt|_|rQ|jdtksJJ|j|ksQJ|S)z|Given an input checkpoint filepath, clean it using `ckpt_to_dir` and then return the weights subdirectory, if it exists.)r<base_model_path) r isinstancer:partsris_dirr)r>parent)r< is_savingbase_dirmaybe_base_dirmaybe_base_model_pathr.r.r/ckpt_to_weights_subdirss     rGc@seZdZdZ        d%dededed eed ed ed ed efddZe d&de ee fde dee ddfddZ e   d'de deededeBde ee ffddZe de ddfddZddZed(d d!Zde d"e ee ffd#d$ZdS))MegatronCheckpointIOzCheckpointIO that utilizes :func:`torch.save` and :func:`torch.load` to save and load checkpoints respectively, common for most use cases. .. warning:: This is an :ref:`experimental ` feature. torch_distTFNsave_ckpt_formatload_directly_on_device async_savetorch_dist_multiprocassume_constant_structure parallel_saveparallel_save_within_dp parallel_loadc Cs@||_||_||_||_||_||_||_||_d|_d|_ dS)NF) rJrKrLrMrNrOrPrQ_save_sharded_strategyvalidated_consistency) selfrJrKrLrMrNrOrPrQr.r.r/__init__s  zMegatronCheckpointIO.__init__ checkpointpathstorage_optionsr'cs4ddlm}|durt|dkrt|jjd|dtdd}t|}|j |dd|j o3|j }d|_ t j }t|t} |j|||j||jd } t} d d |durcd ndd | ddd| | ddf} ddd| D} t| fdd}|jr| dusJ| || S)aSave model/training states as a checkpoint file through state-dump and file-write. Args: checkpoint: dict containing model and trainer state path: write-target path storage_options: not used in ``TorchCheckpointIO.save_checkpoint`` Raises ------ TypeError: If ``storage_options`` arg is passed in rdist_checkpointingNz7 does not support storage_options, but storage_options=z- was provided. Ignoring given storage_optionsTrC)exist_ok)sharded_state_dictcheckpoint_dirsharded_strategyvalidate_access_integrityasync_sharded_savezGlobal Checkpoint SavezRank: z Iteration: z Start time: .3fszSave duration: z : css|] }|dur|VqdSNr.).0partr.r.r/ sz7MegatronCheckpointIO.save_checkpoint..cs tdtdddS)Nz-Successfully saved checkpoint from iteration 7dz to )rinfointr. iterationrWr.r/iter_finalize_fns z>MegatronCheckpointIO.save_checkpoint..iter_finalize_fn) megatron.corerZlenrwarning __class__r2rGr makedirsrSrNtorch distributedget_rank_get_iteration_from_checkpointtimesavesave_sharded_strategyrLjoinriadd_finalize_fn)rTrVrWrXrZr^fsvalidate_sharding_integrityrank start_timeasync_save_requestend_time log_parts log_messagermr.rkr/save_checkpointsH        z$MegatronCheckpointIO.save_checkpoint map_locationstrictStrictHandlingc Csddlm}ddlm}|durtdt|}||s$td|||s1td|dt |d d }t |d rF|j sF|j j |_ |j d krZ|jrZdd lm}|dd} nd} |jro| durgt|} t| tdd} | dur|td| dt|tr|s|||}|r|jn|j}|dur|j}t} |j|t|| |d} t| } t} | | } tdt j!"d| dd| dd| S)aLoads checkpoint using :func:`torch.load`, with additional handling for ``fsspec`` remote loading of files. Args: path: Path to checkpoint map_location: a function, :class:`torch.device`, string or a dict specifying how to remap storage locations. Returns: The loaded checkpoint. Raises ------ FileNotFoundError: If ``path`` is not found by the ``fsspec`` filesystem rrY)rNzT`map_location` argument is not supported for `MegatronCheckpointIO.load_checkpoint`.zCheckpoint file not found: z6Distributed checkpoints should be a directory. Found: .Fr[r>zarr)TensorStoreLoadShardedStrategyT)rKwith_context_parallelUsing z dist-ckpt load strategy.)r]r^r_rz Global Checkpoint Load : Rank : z : Start time : rbz#s : Time spent in load_checkpoint: rc)#rnrZ+megatron.core.dist_checkpointing.validationrr*r existsFileNotFoundErrorisdirrGr)r>rBrJrK7megatron.core.dist_checkpointing.strategies.tensorstorerrQrrrrrir?booladjust_non_strict_loadASSUME_OK_UNEXPECTEDLOG_ALLrwloadr:_fix_tensors_devicersrtru)rTrWr]rrrZrr|rr_rrVrdurationr.r.r/load_checkpointsd           z$MegatronCheckpointIO.load_checkpointcCs8t|}||r|j|ddtd|dSdS)ziRemove checkpoint file from the filesystem. Args: path: Path to checkpoint T) recursivezRemoved checkpoint: N)r rrmrdebug)rTrWr|r.r.r/remove_checkpoint9s  z&MegatronCheckpointIO.remove_checkpointcCs|jdkr td|jr|jdkrtd|jdurint|jd}|jdkr5|r5t|jdfi|}nt|jd}t |drD|j |_ |j rX|j rOtd d nd}t|||j }td |d |S) aDetermine the saving strategy based on constructor args. Relies on the default MCore strategy unless extra PyT Distributed format arguments are passed in config or in case of a fully parallel save in which case a parallelization wrapper is applied. rz`zarr` distributed checkpoint backend is deprecated. Distributed optimizer checkpoint saving might be extremely slow. Please switch to PyTorch Distributed format (model.dist_ckpt_format=torch_dist).rIz9Async dist-ckpt save supported only for torch_dist formatN) thread_countuse_cached_ckpt_structureTrrz dist-ckpt save strategy.)rJrrprLr*rMr9rrr)rNrrOrPrrri)rTtorch_dist_kwargs save_strategyparallelization_groupr.r.r/"_determine_dist_ckpt_save_strategyFs(   z7MegatronCheckpointIO._determine_dist_ckpt_save_strategyrcCs|jdur ||_|jS)zF initializes (if needed) the sharding strategy and returns itsN)rRr)rTr.r.r/ryls  z*MegatronCheckpointIO.save_sharded_strategyr]cspddlm}ddlm}ddlm||ggdtffdd }|||\}}t d|S) a. Adjusts the loading of a non-strict sharded checkpoint by filtering out missing keys. This function loads the checkpoint's metadata and removes any `ShardedBase` keys from `sharded_state_dict` that do not exist in the checkpoint. It also logs unexpected keys that were not found in the checkpoint. Args: path (_PATH): The path to the checkpoint. sharded_state_dict (Dict[str, Any]): The state dictionary containing sharded parameters. Returns: Dict[str, Any]: The adjusted state dictionary with missing keys removed. Notes: - Keys that exist in `sharded_state_dict` but are not found in the checkpoint metadata are considered "unexpected" and are logged. - Missing keys are not computed yet. To fully determine missing keys: 1. Perform an `all_gather_object` operation on `loaded_keys`. 2. Compute `missing_keys` as the difference between `ckpt_sharded_metadata.keys()` and `loaded_keys`. rrY)extract_matching_values) ShardedBasexcs8t|r|jvr|jdS|jdSdS)a Helper function to determine if a `ShardedBase` key should be removed. Args: x (Any): The object to check. Returns: bool: True if the key should be removed, False otherwise. FT)r?keyappend)rrckpt_sharded_metadata loaded_keysunexpected_keysr.r/"should_remove_missing_sharded_bases   zWMegatronCheckpointIO.adjust_non_strict_load..should_remove_missing_sharded_basezEThe following keys are not in the checkpoint and will not be loaded: ) rnrZ+megatron.core.dist_checkpointing.dict_utilsr(megatron.core.dist_checkpointing.mappingrload_tensors_metadatarrri)rTrWr]rZrr_r.rr/rts    z+MegatronCheckpointIO.adjust_non_strict_load)rITFNFTFFrd)NNN)r'r)r2r3r4r5r:rr rjrUrrrrrrrrrpropertyryrr.r.r.r/rHs`  *<   T & rHckptcsXtjsJtjtjftjdtjdddlm}fdd}|||S)z4Ensure checkpoint tensors are on the correct device.cuda)indexr)dict_list_map_outplacecs*t|tjr|jr|jkr|}|Srd)r?rsTensoris_cudadeviceto)tcur_devr.r/ _fix_devices z(_fix_tensors_device.._fix_device)rsris_initialized is_availablercurrent_devicerr)rrrr.rr/rs "   rcCs0ddlm}t|}t|}||o||S)aCheck if the given path corresponds to a distributed checkpoint directory. This function determines if the specified path is a directory that contains a distributed checkpoint by checking the directory's metadata. Args: path (Union[str, Path]): The path to check for being a distributed checkpoint. Returns ------- bool: True if the path is a distributed checkpoint directory, False otherwise. rrY)rnrZrr rcheck_is_distributed_checkpoint)rWrZr^r|r.r.r/is_distributed_ckpts rrVcCs,|dididididdS)Nloopsfit_loopzepoch_loop.batch_progresstotal completed)get)rVr.r.r/rvs rv)Arw dataclassesrrpathlibrtypingrrrrr r r lightning.pytorchpytorchr7rslightning.fabric.pluginsr #lightning.fabric.utilities.cloud_ior lightning.fabric.utilities.typesr.megatron.core.dist_checkpointing.serializationrr0megatron.core.dist_checkpointing.strategies.baser:megatron.core.dist_checkpointing.strategies.fully_parallelrr1megatron.core.dist_checkpointing.strategies.torchrmegatron.core.parallel_staterrtyping_extensionsrrnemo.lightning.ckpt_utilsrrnemo.lightning.io.capturernemo.lightning.io.mixinr nemo.utilsr!nemo.utils.callbacks.dist_ckpt_ior ImportErrorLightningModulerModuler!r"r:rGrHrrrrjrvr.r.r.r/sH $            <)"