o }oiU @sddlZddlmZmZddlmZddlmZddlmZm Z m Z m Z ddl m ZddlZddlmZddlmZddlmZdd l mZdd lmZdd lmZzLdd lmZdd lmZddlm Z ddl!m"Z"m#Z#ddl$m%Z%ddl&m'Z'm(Z(ddl)m*Z*ddl+m,Z,m-Z-ddl.m/Z/ddl0m1Z1ddl2m3Z3dZ4Wne5e6fyZ7z dZ4de7Z8WYdZ7[7ndZ7[7wwede9fddZ:GdddeeZ;Gdd d eZd%e e9efd&e e?fd'd(Z@dS))N)ABCabstractmethod)contextmanager)time)AnyDictOptionalUnion) CheckpointIO)get_filesystem)_PATH)Callback)_WrappingCheckpointIO)logging)dist_checkpointing)extract_matching_values) ShardedBase)!get_default_load_sharded_strategy!get_default_save_sharded_strategy) tensorstore)AsyncCallsQueue AsyncRequest)SaveShardedStrategy) FullyParallelLoadStrategyWrapper FullyParallelSaveStrategyWrapper)TorchDistSaveShardedStrategy)StrictHandling)get_data_parallel_groupTFzmegatron-core was not found. Please see the NeMo README for installation instructions: https://github.com/NVIDIA/NeMo#megatron-gpt. Exact error: namec csTt}zdVWt|dt|dddSt|dt|ddw)z8Simple context manager for timing functions/code blocks.Nz took .3fs)rrdebug)rstartr#U/home/ubuntu/.local/lib/python3.10/site-packages/nemo/utils/callbacks/dist_ckpt_io.py _debug_time>s Br%c @s>eZdZdZe d deeefdede eddfdd Z dS) AsyncCompatibleCheckpointIOa CheckpointIO that can be used together with async saving. Differs from the regular CheckpointIO only by the `save_checkpoint` return type. The `save_checkpoint` method itself is synchronous, but returns callbacks that can be performed asynchronously. N checkpointpathstorage_optionsreturnrcCst)zAInterface to implement save_checkpoint and return an AsyncRequest)NotImplementedError)selfr'r(r)r#r#r$save_checkpointPsz+AsyncCompatibleCheckpointIO.save_checkpointN) __name__ __module__ __qualname____doc__rrstrrr rr-r#r#r#r$r&Hs r&c sxeZdZdZdeddffdd Zddeeefde d e eddfd d Z e d dde fdd Zdfdd ZZS)AsyncFinalizableCheckpointIOaCheckpointIO wrapper for async checkpoint saving and synchronous finalization. Runs main part of the checkpoint save in a separate process (not thread as the PTL AsyncCheckpointIO does). Allows to perform a (synchronous) finalization function after all ranks finish checkpoint saving. NOTE: for correctness, this plugin must be used together with the AsyncFinalizerCallback callback which performs the finalization checks. Args: checkpoint_io (CheckpointIO): wrapped checkpoint_io object. Must be of type AsyncCompatibleCheckpointIO. Requires the underlying checkpoint_io.save_checkpoint to return save_fn, save_args, finalize_fn. checkpoint_ior*Ncs@tsttt|tstdt|t|t |_ dS)Nz)Incompatible wrapped checkpoint_io type: ) HAVE_MEGATRON_CORE ImportError IMPORT_ERROR isinstancer& ValueErrortypesuper__init__rasync_calls_queue)r,r5 __class__r#r$r=hs    z%AsyncFinalizableCheckpointIO.__init__r'r(r)cCsl|pidd}t|jtsJt|j|j|||}|dur&|||j|}t d|dS)atExecutes async request returned from the underlying checkpoint_io asynchronously. Requires the underlying checkpoint_io.save_checkpoint to return an AsyncRequest. It is then applied with `self.async_calls_queue` asynchronously. Args: checkpoint (Dict[str, Any]): checkpoint to save. Passed to underlying checkpoint_io without modifications. path (_PATH): path to save the checkpoint. Passed to underlying checkpoint_io without modifications. storage_options (Any, optional): storage control modifiers. This class consumed the `finalize_fn` parameter (if any), which is expected to be a callback and is appended to async finalization functions. Applies underlying checkpoint_io finalize callback first, then the external one (postfix order). finalize_fnNzScheduled an async call #) popr9r5r&r;r-add_finalize_fnr>schedule_async_requestrr!)r,r'r(r)external_finalize_fn async_requestcall_idxr#r#r$r-qs  z,AsyncFinalizableCheckpointIO.save_checkpointz;AsyncFinalizableCheckpointIO.maybe_finalize_save_checkpointFblockingcCsl|jdkr dSt}|j|}|r!tddd|Dt}td||ddt|dkS) a&Performs checkpoint finalization (if possible). Args: blocking (bool, optional): if True, waits until all async saves are completed. Otherwise, finalizes only those async calls which are already done on all ranks. Defaults to False. rFzFinalized async calls: cSsg|]}d|qS)#r#).0idxr#r#r$ szOAsyncFinalizableCheckpointIO.maybe_finalize_save_checkpoint..zAsync finalization time took rz s)r>get_num_unfinalized_callsrmaybe_finalize_async_callsrr!infolen)r,rH start_timecall_idx_finalizedend_timer#r#r$maybe_finalize_save_checkpoints   cs*t|jdkrtddSdS)z0Warns if there are any pending checkpoint saves.rzrMrwarningr,r?r#r$rUs z%AsyncFinalizableCheckpointIO.teardownr.F)r*N)r/r0r1r2r&r=rr3rr rr-r%boolrTrU __classcell__r#r#r?r$r4Xs( r4c@s<eZdZdZdddZddd Zdd d Zdefd d ZdS)AsyncFinalizerCallbackzCallback which finalizes async saves initiated by the AsyncFinalizableCheckpointIO. Tries to perform non-blocking finalization on train_batch_end and train_epoch_end. On train_end performs a blocking finalization of all pending checkpoints. trainer pl.Trainerr*NcO||jdddS>Override hook to finalize pending checkpoint(s) if they exist.FrHN_get_checkpoint_iorTr,r\argskwargsr#r#r$on_train_batch_endz)AsyncFinalizerCallback.on_train_batch_endcOr^r_rbrdr#r#r$on_train_epoch_endrhz)AsyncFinalizerCallback.on_train_epoch_endcOs8||}|jdkrtd||jdddS)r`rzAPending async checkpoint saves. Finalizing them synchronously nowTraN)rcr>rMrrOrT)r,r\rerfr5r#r#r$ on_train_ends  z#AsyncFinalizerCallback.on_train_endcCs&|jj}t|tstd|j|S)Nz@Async finalizer requires an async compatible CheckpointIO, got: )strategyr5r9r4r:r@)r,r\r5r#r#r$rcs   z)AsyncFinalizerCallback._get_checkpoint_io)r\r]r*N) r/r0r1r2rgrirjr4rcr#r#r#r$r[s    r[csNeZdZdZ        d*dedededd ed eed ed ed edeffdd Ze d+de d efddZ e d d,de eefdedeededfddZe d    d-dedeede eefdededfdeede eeff d dZdede eeffd!d"Ze d#deddfd$d#Zed.d&d'Zd(d)ZZS)/DistributedCheckpointIOaCheckpointIO for a distributed checkpoint format. Args: save_ckpt_format (str): Distributed checkpoint format to use for checkpoint saving. load_directly_on_device (bool, optional): if True, loads the weights directly on GPU. Has effect only for `zarr` based checkpoints (PyT Distributed always loads on device). Defaults to True. load_strictness (StrictHandling, optional): defines loading strictness. If not None, overwrites the `strict` flag passed to `load_checkpoint`. Defaults to None. async_save (bool): whether to save asynchronously. Should be set to True if this class will be wrapped with AsyncFinalizableCheckpointIO. torch_dist_multiproc (int, optional): number of extra processes per rank used during ckpt save with PyTorch distributed format. Defaults, to None which means using an MCore default (2). parallel_save (bool): parallelizes the save across ranks. Defaults to True parallel_load (bool): parallelizes the load across ranks (followed by params all gather). Defaults to False due to some extra memory usage requirement. TNFsave_ckpt_formatload_directly_on_deviceload_strictnessr async_savetorch_dist_multiprocassume_constant_structure parallel_saveparallel_save_within_dp parallel_loadc s\tts tt||_||_||_||_||_ ||_ ||_ ||_ | |_ d|_d|_dS)NF)r<r=r6r7r8rmrnrorprqrrrsrtru_save_sharded_strategyvalidated_consistency) r,rmrnrorprqrrrsrtrur?r#r$r=s  z DistributedCheckpointIO.__init__ model_cfgc CsP||dd|dd|dd||dd|dd |d d |d d d S) akInstantiates a DistributedCheckpointIO from a config dict. Args: model_cfg (dict): model config dict. Most of the configuration is extracted from this config. async_save (bool, optional): async_save flag is not part of the model config, it should be provided separately. Defaults to False. dist_ckpt_format torch_distdist_ckpt_load_on_deviceTdist_ckpt_load_strictnessNdist_ckpt_torch_dist_multiprocdist_ckpt_parallel_saveF!dist_ckpt_parallel_save_within_dpdist_ckpt_parallel_load)rmrnrorprqrsrtruget)clsrxrpr#r#r$ from_configs       z#DistributedCheckpointIO.from_configz'DistributedCheckpointIO.save_checkpointr'r(r)r*rc st}|jdd|jo|j }d|_tj}t|t}t j ||j ||j d}t} dd|dur>dndd|d d d | |d d f} d d d| D} t| fdd} |j rs|dusnJ|| |S)a>Saves a distributed checkpoint. Creates the checkpoint root directory if doesn't exist. Args: checkpoint (Dict[str, Any]): sharded state dict to save path (_PATH): checkpoint directory storage_options (Any, optional): Optional parameters when saving the checkpoint T)exist_ok)sharded_state_dictcheckpoint_dirsharded_strategyvalidate_access_integrityasync_sharded_savezGlobal Checkpoint SavezRank: Nz Iteration: z Start time: rr zSave duration: z : css|] }|dur|VqdSr.r#)rJpartr#r#r$ /sz:DistributedCheckpointIO.save_checkpoint..cs tdtdddS)Nz-Successfully saved checkpoint from iteration 7dz to )rrOintr# iterationr(r#r$iter_finalize_fn2s zADistributedCheckpointIO.save_checkpoint..iter_finalize_fn)r makedirsrwrrtorch distributedget_rank_get_iteration_from_checkpointrrsavesave_sharded_strategyrpjoinrrOrC) r,r'r(r)fsvalidate_sharding_integrityrankrQasync_save_requestrS log_parts log_messagerr#rr$r- s8      z'DistributedCheckpointIO.load_checkpoint map_locationrstrictrc Cs.|durtd|durtd|jdkr|jrtjdd}nd}|jr4|dur,t|}t|tdd}|durAt d|d t |t rV|sN| ||}|rStjntj}|jdur^|j}|duretj}t d |t}tj|||||d }t} | |} t d tjd |dd| dd|S)a\Loads a distributed checkpoint. Args: path (_PATH): checkpoint directory map_location (Any, optional): required to be None in this implementation sharded_state_dict (Dict[str, Any], optional): state dict which defines the loading procedure for the distributed checkpoint. Defaults to None to comply with the CheckpointIO interface, but it's a required argument. strict (bool, StrictHandling, optional): adjust load strictness. bool value is translated to StrictHandling instance. Gets overwritten by `self.load_strictness`. Defaults to None. If `self.load_strictness` is also None, strict becomes StrictHandling.ASSUME_OK_UNEXPECTED. Returns: Dist[str, Any]: loaded checkpoint. NzWDistributedCheckpointIO requires passing sharded_state_dict argument to load_checkpointz;DistributedCheckpointIO doesnt handle map_location argumentzarrT)rnwith_context_parallelUsing z dist-ckpt load strategy.zDist ckpt load strictness: )rrrrrz Global Checkpoint Load : Rank : z : Start time : rz#s : Time spent in load_checkpoint: r )r:rmrnrTensorStoreLoadShardedStrategyrurrrrrOr9rYadjust_non_strict_loadrASSUME_OK_UNEXPECTEDLOG_ALLror!rrloadrrr) r,r(rrrrrrQretrSdurationr#r#r$load_checkpoint;sX    csJt|ggdtffdd }t||\}}td|S)z=Remove unexpected keys from being loaded into the state dict.xcs8t|tr|jvr|jdS|jdSdS)NFT)r9rkeyappend)rckpt_sharded_metadata loaded_keysunexpected_keysr#r$"should_remove_missing_sharded_bases    zZDistributedCheckpointIO.adjust_non_strict_load..should_remove_missing_sharded_basezEThe following keys are not in the checkpoint and will not be loaded: )rload_tensors_metadatarrrrO)r,r(rr_r#rr$rs  z.DistributedCheckpointIO.adjust_non_strict_loadz)DistributedCheckpointIO.remove_checkpointcCstj|dddS)zRemove a distributed checkpoint. Due to potentially large number of files, the implementation remove the whole directory at once. T) ignore_errorsN)shutilrmtree)r,r(r#r#r$remove_checkpointsrcCs|jdur ||_|jS)zHConditionally initialize and get the sharded strategy to use for saving.N)rv"_determine_dist_ckpt_save_strategyrWr#r#r$rs  z-DistributedCheckpointIO.save_sharded_strategycCs|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).rzz9Async dist-ckpt save supported only for torch_dist formatN) thread_countuse_cached_ckpt_structureTrrz dist-ckpt save strategy.)rmrrVrpr:rqdictrrhasattrrrrrsrtrrrO)r,torch_dist_kwargs save_strategyparallelization_groupr#r#r$rs(   z:DistributedCheckpointIO._determine_dist_ckpt_save_strategy)TNFNFFFFrXr.)NNNT)r*r)r/r0r1r2r3rYrrr= classmethodrrr%rrr r-r rrrpropertyrrrZr#r#r?r$rls    .    N rlr'r*cCs,|dididididdS)Nloopsfit_loopzepoch_loop.batch_progresstotal completedr)r'r#r#r$rs r)Arabcrr contextlibrrtypingrrrr lightning.pytorchpytorchplrlightning.fabric.pluginsr #lightning.fabric.utilities.cloud_ior lightning.fabric.utilities.typesr r $lightning.pytorch.plugins.io.wrapperr nemo.utilsr megatron.corer+megatron.core.dist_checkpointing.dict_utilsr(megatron.core.dist_checkpointing.mappingr.megatron.core.dist_checkpointing.serializationrr+megatron.core.dist_checkpointing.strategiesr7megatron.core.dist_checkpointing.strategies.async_utilsrr0megatron.core.dist_checkpointing.strategies.baser:megatron.core.dist_checkpointing.strategies.fully_parallelrr1megatron.core.dist_checkpointing.strategies.torchr+megatron.core.dist_checkpointing.validationrmegatron.core.parallel_staterr6r7ModuleNotFoundErrorer8r3r%r&r4r[rlrrr#r#r#r$sV                   N"