o 0i @sFddlZddlZddlZddlZddlZddlZddlmZddlm Z ddl m Z ddl m Z ddlmZmZddlZddlmZd d lmZmZmZmZd d lmZd d lmZmZd d lmZm Z m!Z!d dl"m#Z#ee$Z%ddZ&ddZ'dFddZ(dFddZ)ddZ*dGddZ+dGddZ,dHddZ-dGd d!Z.dId#e/d$e/d%e0fd&d'Z1 dJd#e/d(e/d%e0d)e0fd*d+Z2d,ej3j4d-ej5fd.d/Z6d,ej3j4d0e7fd1d2Z8d3ej9j:d4e7fd5d6Z;d,ej3j4fd7d8Zd>d?Z?d@e7d9e7fdAdBZ@dCee ej3j4e/fd9eAej3jBfdDdEZCdS)KN) defaultdict)Iterable) nullcontext)Path)CallableUnion) get_logger)FSDP_MODEL_NAMEOPTIMIZER_NAMESAFE_WEIGHTS_NAME WEIGHTS_NAME)get_module_class_from_name)get_non_persistent_buffers is_peft_model)get_module_children_bottom_upis_compiled_modulesave)is_torch_versioncCs"dtjvr dtjd<dtjd<dS)z[ Enables RAM efficient loading of Hugging Face models for FSDP in the environment. ACCELERATE_USE_FSDPTrueFSDP_CPU_RAM_EFFICIENT_LOADINGNosenvironrr]/sda-disk/www/egybert/egybert_env/lib/python3.10/site-packages/accelerate/utils/fsdp_utils.py!enable_fsdp_ram_efficient_loading's  rcCsdtjd<dS)z\ Disables RAM efficient loading of Hugging Face models for FSDP in the environment. FalserNrrrrr"disable_fsdp_ram_efficient_loading1sr FcCsN|rt|rddlm}|||jdS|dur#ddlm}|||dS|S)Nr)get_peft_model_state_dict adapter_name)get_model_state_dictoptions)rpeftr!active_adapter'torch.distributed.checkpoint.state_dictr$ state_dict)model adapter_only sd_optionsr!r$rrr_get_model_state_dict8s    r.cCsT|rt|rddlm}||||jdS|dur%ddlm}||||dS||S)Nr)set_peft_model_state_dictr")set_model_state_dictr%)rr'r/r(r)r0load_state_dict)r+r*r,r-r/r0rrr_set_model_state_dictGs    r2cCsTd}|jdkr(ddlm}ddlm}||j|jkt|jddt|jddd}|S) Nrr)StateDictOptions StateDictTypeoffload_to_cpuF rank0_only)full_state_dict cpu_offloadbroadcast_from_rank0) fsdp_versionr)r32torch.distributed.fsdp.fully_sharded_data_parallelr5state_dict_typeFULL_STATE_DICTgetattrstate_dict_config) fsdp_pluginr-r3r5rrr_prepare_sd_optionsVs      rBcCsdddlmm}ddlm}ddlm}ddlm} tj |dd|j | j kr5|j dk} | |j _| |j _|jdkrE| ||j |j |jnt} t|} | t||| d} |j | j kr|dkretd ntd |d }tj||}|jdkrtd |t| |td |n|j | jkr|dkrtd |jd n td |d |jd }tj||}td |t| |td |nR|j | jkrtj|td |}tj |ddtd |d| i} |j| |||dtd |WddSWddSWddSWddS1s+wYdS)NrDefaultSavePlannerFullyShardedDataParallelr4Texist_okr r,r-.bin_zSaving model to zModel saved to _rankr+r*storage_writerplanner) torch.distributed.checkpoint distributed checkpoint,torch.distributed.checkpoint.default_plannerrDr<rFr5rmakedirsr=r> num_processesr@r6r7r;optim_state_dict_configrrBr.r pathjoin process_indexloggerinfotorchrLOCAL_STATE_DICTSHARDED_STATE_DICTFileSystemWriter)rA acceleratorr+ output_dir model_indexr,dist_cprDFSDPr5is_multi_processctxr-r* weights_nameoutput_model_fileckpt_dirrrrsave_fsdp_modelgsj       "    $rjcCsddlmm}ddlm}ddlm}ddlm} ||j | j kr2|j dk} | |j _ | |j _|jdkrB| ||j |j |jnt} t|} | |j | j krt||uru|jdkru|jsu|jsl|jdkrltd WddS|dkr~tdntd|d} tj|| }td ||j p|j}|rtj |d d }ni}td |n|j | j!kr|dkrtd |jdn td|d |jd} tj|| }td |tj |d d }td |nD|j | j"kr3t|vrtj|td|n|}td |dt#||| di}|j ||$||d|d}td |t%|||| d}Wd|S1sGwY|S)Nr)DefaultLoadPlannerrEr4r zzSet the `sync_module_states` flag to `True` so that model states are synced across processes when initializing FSDP objectrJrKzLoading model from T weights_onlyzModel loaded from rLr+rI)r*storage_readerrO)&rPrQrRrSrkr<rFr5wait_for_everyoner=r>rUr@r6r7r;rVrrBtyperYis_fsdp2sync_module_states ValueErrorr rrWrXrZr[is_main_processr\loadr]r^r.FileSystemReaderr2)rAr`r+ input_dirrbr,rcrkrdr5rerfr-rginput_model_file load_modelr*ri load_resultrrrload_fsdp_modelsz       "    ..r{cCsddlmm}ddlm}ddlm}ddlm} tj |dd|j dkr2| ||j |j |j nt} t|} | |j dkrOdd lm} | ||| d } n|||} |j | jkr|jdkr|dkritd ntd |d }tj||}td |t| |td|n9tj|td |}tj |ddtd ||jd| i|||dtd|WddSWddS1swYdS)NrrCrEr4TrGr r)get_optimizer_state_dictr%rJrKzSaving Optimizer state to zOptimizer state saved in optimizerrM)rPrQrRrSrDr<rFr5rrTr;r=r@rVrrBr)r|optim_state_dictr>rYr rWrXrZr[r\rr_)rAr`r}r+raoptimizer_indexrcrDrdr5rfr-r| optim_stateoptim_state_nameoutput_optimizer_filerirrrsave_fsdp_optimizersL           "rcCsddlmm}ddlm}ddlm} ||jdkr)|||j|j |j nt } t |} | |j| j krtd} |jdksD|j jss|dkrMtdntd|d} tj|| }td|tj|dd } td |n8t|vrtj|td|n|}td |d |i} |j| |||d | d } td||jdkr|j||| d}||nddlm}|||| | dWddSWddS1swYdS)NrrEr4r rJrKzLoading Optimizer state from TrlzOptimizer state loaded from zLoading Optimizer from r}) checkpoint_idrnzOptimizer loaded from )r+optimr~)set_optimizer_state_dictr%)rPrQrRr<rFr5ror;r=r@rVrrBr>rYr7r rrWrXrZr[r\rur*rvoptim_state_dict_to_loadr1r)r)rAr`r}r+rwrr,rcrdr5rfr-roptimizer_nameinput_optimizer_fileri flattened_osdrrrrload_fsdp_optimizersV          "rTcheckpoint_dir save_pathsafe_serializationcCsddlmm}ddlmmm}i}t|}|jdd|j|||| dd|r5|t n|t }t | dkrI|t|d}t|||d|S)z Passthrough to `torch.distributed.checkpoint.format_utils.dcp_to_torch_save` Will save under `save_path` as either `model.safetensors` or `pytorch_model.bin`. rNTrG)rnrOno_distr )r)rPrQrR)torch.distributed.checkpoint.format_utils format_utilsrmkdir_load_state_dictrv_EmptyStateDictLoadPlannerr rlenkeyslistr)rrrrcdist_cp_format_utilsr*rrr)_distributed_checkpoint_to_merged_weightsLs  r output_pathremove_checkpoint_dirc Cs2t|}ddlm}tddstd|sh|d}|d}d|d }|rD|rD|d 7}|d |d |d 7}|d7}t||rV|d7}|d|d7}t||rd|d7}|d|d7}t||}|jrtd|t |||} td| |rtd|t || dS)a? Merge the weights from sharded FSDP model checkpoints into a single combined checkpoint. Should be used if `SHARDED_STATE_DICT` was used for the model. Weights will be saved to `{output_path}/model.safetensors` if `safe_serialization` else `pytorch_model.bin`. Note: this is a CPU-bound process. Args: checkpoint_dir (`str`): The directory containing the FSDP checkpoints (can be either the model or optimizer). output_path (`str`): The path to save the merged checkpoint. safe_serialization (`bool`, *optional*, defaults to `True`): Whether to save the merged weights with safetensors (recommended). remove_checkpoint_dir (`bool`, *optional*, defaults to `False`): Whether to remove the checkpoint directory after merging. r) PartialStatez>=z2.3.0z/`merge_fsdp_weights` requires PyTorch >= 2.3.0`pytorch_model_fsdp_0 optimizer_0zTried to load from z) but couldn't find a valid metadata file.zE However, potential model and optimizer checkpoint directories exist.zPlease pass in either z/pytorch_model_fsdp_0 or z /optimizer_0zinstead.z8 However, a potential model checkpoint directory exists.zPlease try passing in z/pytorch_model_fsdp_0 instead.z< However, a potential optimizer checkpoint directory exists.z/optimizer_0 instead.zMerging FSDP weights from z.Successfully merged FSDP weights and saved to z"Removing old checkpoint directory N) raccelerate.staterrrsexistsrtrZr[rshutilrmtreero) rrrrrmodel_path_existsoptimizer_path_existserrstaterrrrmerge_fsdp_weightshs<        rr+devicec st|dd}|s Si|D]%}|d}d|dd|d}}||}t||}dt|<qdtjjffdd }|S)N_tied_weights_keys.modulecstt}|jddD]\}}t|vr|t||q |}|D]\}}|D]}|}|dur>t|||<q,t|||q,q&|S)NF)recurse)rrnamed_parametersidappenditemsr?setattr)r params_to_tienparamid_key _param_names param_name _tied_params param_init_fnrrparam_init_fn_tied_params  z7ensure_weights_retied..param_init_fn_tied_param)r?splitrX get_submodulerr\nnModule) rr+r _tied_namesnamermodrrrrrensure_weights_retieds    rfull_sdcCsVddlm}ddlm}m}|}i}dd}dd} |jrgt|| D]=\\} } } | j } | | j } t| |rB| } |j| d|jjd|| | | j}||| | \}}| |||}||| <q(n;|D]6\} } | j } tj| | j | jd }|j|d|jjd||| | j}||| |\}}| |||}||| <qk|j|d d |S) a Loads the full state dict (could be only on rank 0) into the sharded model. This is done by broadcasting the parameters from rank 0 to all other ranks. This function modifies the model in-place. Args: accelerator (`Accelerator`): The accelerator instance model (`torch.nn.Module`): The model to load the state dict into, expected to be on meta device or a VRAM spike can occur full_sd (`dict`): The full state dict to load, can only be on rank 0 rN)DTensordistribute_tensorc Ssz||}Wnty"|dd\}}||}t||}Ynwttd}d}|o1|jtjk} |jj r;| s;|j}|duoB| |fS)Nrr float8_e4m3fn) get_parameter_or_bufferAttributeErrorrsplitrr?hasattrr\dtyperis_floating_point is_contiguous) r+r empty_param old_parambase_param_namelocal_param_name submoduleis_torch_e4m3fn_available casting_dtypeis_param_float8_e4m3fnrrr_infer_parameter_dtypes    z:fsdp2_load_full_state_dict.._infer_parameter_dtypecSs$|dur |j|d}|r|}|S)N)r)to contiguous)tensor to_contiguousrrrr_cast_and_contiguouss  z8fsdp2_load_full_state_dict.._cast_and_contiguous)srcgroup)rrT)assign)torch.distributedrQtorch.distributed.tensorrrr*rtziprvalues device_meshdetachr device_type isinstanceto_local broadcastrWORLD placementsr\emptysizerr1)r`r+rdistrrmeta_sharded_sd sharded_sdrrr full_param sharded_paramrsharded_tensorrr full_tensorrrrfsdp2_load_full_state_dictsH      rr}mappingcs\ddlm}i}d||<z|jD]}fdd|dD|d<qWdSty-tdw) a Switches the parameters of the optimizer to new ones (sharded parameters in usual case). This function modifies the optimizer in-place. Args: optimizer (`torch.optim.Optimizer`): Optimizer instance which contains the original model parameters mapping (`dict`): Mapping from the original parameter (specified by `data_ptr`) to the sharded parameter Raises: KeyError: If a parameter in the optimizer couldn't be switched to its sharded version. This should never happen and indicates a bug. If we kept the original params instead of raising, the training wouldn't be numerically correct and weights wouldn't get updated. r)r _local_tensorcsg|]}|jqSr)data_ptr.0prrr 5sz5fsdp2_switch_optimizer_parameters..paramszA parameter in the optimizer couldn't be switched to its sharded version. This breaks the training. Please raise an issue on GitHub.N)rr param_groupsKeyError)r}rraccessor_mapping param_grouprrr!fsdp2_switch_optimizer_parameterss   rc Csddlm}t|jj|}t|ddddD]3\}}t|ddkr-|dd\}}nd}|}|r8| |n|}||rJ||d d }| ||q|S) a8 Applies the activation checkpointing to the model. Args: accelerator (`Accelerator`): The accelerator instance model (`torch.nn.Module`): The model to apply the activation checkpointing to Returns: `torch.nn.Module`: The model with the activation checkpointing applied r)checkpoint_wrapperT) return_fqnsNrrr F)preserve_rng_state) ;torch.distributed.algorithms._checkpoint.checkpoint_wrapperrfsdp2_prepare_auto_wrap_policyrrArrrrrregister_module) r`r+rauto_wrap_policy_func layer_namelayer parent_name child_name parent_modulerrrfsdp2_apply_ac>s   rreturncsbddlm}m}m}t||pt|ot|j|}|r|S|jj}| || }t |dd}|j |j |jp9||durE|t|jjndt|j||jd} d} |D]\} } | jjdkrdd} nqV|jr| st|ddd tfd d |D} |td }t|d r| t!||}|durt"|ddD]}||rt||s||fi| qt||s||fi| |jrt#||||jr | s | $D](\}}||j}d|vr|%dd\}}|&|}n|}|}|j'||ddqt|d r | t |dd}|j(dkr/|dus |tj)kr/|tj)}|j*r/t+,d|S)a"Prepares the model for FSDP2 in-place. Also returns the model to avoid misuse of the original model. Args: accelerator (`Accelerator`): The accelerator instance model (`torch.nn.Module`): The model to prepare Returns: `torch.nn.Module`: Prepared model r) FSDPModuleMixedPrecisionPolicy fully_shardtorch_device_meshN)reshard_after_forwardoffload_policy mp_policymeshignored_paramsF Params4bitT)rfqnscsi|] \}}|vr||qSrrrkvnon_persistent_buffer_fqnsrr z'fsdp2_prepare_model..meta tie_weightsrrr ) persistentrnoz~FSDP upcast of low precision parameters to fp32 (since mixed_precision != 'no') may affect the precision of model checkpoints.)-torch.distributed.fsdprrrrr _orig_modrrAset_auto_wrap_policyr*r?rr9mixed_precision_policytupleparallelism_configfsdp_dim_namesget_parameters_from_modulesignored_modulesrr __class____name__cpu_ram_efficient_loadingrcopydeepcopy named_buffersrr\rr&r rrrrrregister_buffermixed_precisionfloat32rtwarningswarn)r`r+rrr is_type_fsdp fsdp2_plugin original_sdr fsdp2_kwargsmodel_has_params4bitrroriginal_non_persistent_buffersr rfqn buffer_tensor parent_fqnlocal_buffer_namer model_dtyperr!rfsdp2_prepare_model_sv                " rHc sddlm}m}j}t|tjr|j}||ur^t|dd}|dur$g}t |}j dur0j }t |D]}t ||}|durHt d|d|q5dtjjdtffd d } | S||urqdtjjdtffd d } | SdS) a!Prepares the auto wrap policy based on its type, done to mimic the behaviour of FSDP1 auto wrap policy. Args: fsdp2_plugin (`FullyShardedDataParallelPlugin`): Instance of `FullyShardedDataParallelPlugin` containing the configuration options auto_wrap_policy_type (`str`): Either `transformer` or `size` model (`torch.nn.Module`): The model to wrap Returns: `Callable[[torch.nn.Module], bool]`: The auto wrap policy function to be applied to the model r)size_based_auto_wrap_policytransformer_auto_wrap_policy_no_split_modulesNz+Could not find the transformer layer class z in the model.rrcsjdurdSt|tS)NF)transformer_cls_names_to_wraprr-)rr>transformer_cls_to_wraprrpolicys z.fsdp2_prepare_auto_wrap_policy..policycs tdd|D}|jkS)Ncss|]}|VqdS)N)numelrrrr szAfsdp2_prepare_auto_wrap_policy..policy..)sum parametersmin_num_params)rmodule_num_params)r>rrrOs )torch.distributed.fsdp.wraprIrJauto_wrap_policyr functoolspartialfuncr?rrLsetrrsaddr\rrbool) r>r+rIrJfnno_split_modulesrL layer_classtransformer_clsrOrrMrr s.      r cKsddlm}|di|S)a Returns a `GradScaler` for FSDP2, as the current implementation of `get_grad_scaler` doesn't accept other args. We need this as current `get_grad_scaler` accepts only `distributed_type` as arg, which doesn't differentiate between FSDP1 and FSDP2 r) GradScalerNr)torch.amp.grad_scalerrb)kwargsrbrrrget_fsdp2_grad_scalers re named_paramscCs:dd|D}dd|D}dd|D}|S)a6Removes parameter name modifiers in order to map them back to their original names. See huggingface/accelerate#3554 for more context. Args: named_params (`dict`): The named parameters dictionary to canonicalize. Returns: `dict`: The canonicalized named parameters dictionary cSi|] \}}|dd|qS)z._checkpoint_wrapped_modulereplacerrrrr#r$z,fsdp2_canonicalize_names..cSs,i|]\}}|dr|ddn||qS)z _orig_mod.rh) startswithrjrrrrr#s cSrg)z ._orig_modrhrirrrrr# r$)r)rfrrrfsdp2_canonicalize_namess  rlmodulescCs|durtSg}t|tr/t|}g}|D]\}}||r,||||q|}|D] }| t | q1t|S)zConverts modules to parameters where modules can be a string or list of torch.nn.Module Args: modules (`Union[Iterable[torch.nn.Module], str]`): List of modules Returns: `set[torch.nn.Parameter]`: List of parameters N) r[rstrrecompile named_modules fullmatchrrextendrrS)rmr+rrSregmapped_modulesrrrrrr0$s      r0)FN)rF)r)T)TF)Dr5rXrrorr; collectionsrcollections.abcr contextlibrpathlibrtypingrrr\loggingr constantsr r r r dataclassesrmodelingrrotherrrrversionsrr3rZrr r.r2rBrjr{rrrnr]rrrrrrdictrr OptimizerrrrHr rerlr[ Parameterr0rrrrsh            : H 03 7.R!$r5