o h(@sddlZddlZddlZddlZddlZddlmZddlmZddl m Z ddl m Z ddl Z ddlmZddlmZdd lmZdd lmZdd lmZd d lmZd dlmZd dlmZmZm Z d dl!m"Z"d dl#m$Z$d dl%m&Z&ee'Z(e ddrddl)m*m+Z,ddZ-d4de.fddZ/ddZ0de1fddZ2d5d e.d!e.fd"d#Z3ed$d%Z4ed&d'Z5d(d)Z6d*d+Z7d6d,e8d-e.fd.d/Z9d0d1Z:d2d3Z;dS)7N)contextmanager)partial) MethodType) OrderedDict)Version) save_file)write_basic_config) get_logger) PartialState)FSDP_PYTORCH_VERSION)DistributedType)is_deepspeed_availableis_torch_distributed_availableis_tpu_available)id_tensor_storage) convert_model)is_torch_versionF) check_devicecCs(tdds ttds dSt|tjjjS)zD Check whether the module was compiled with torch.compile() =)FullyShardedDataParallelforward_original_forwardN __wrapped__ _converted_to_transformer_engineF)to_transformer_engine)rnnparallelDistributedDataParallel DataParallelr _orig_modr deepspeedr"rr r2torch.distributed.fsdp.fully_sharded_data_parallelr#rrgetattr__dict__poprr&rr$r) modelr!options is_compiledcompiled_modelr"FSDPr$original_forwardrrrextract_model_from_parallel8s<            r9cCstdS)a  Introduces a blocking point in the script, making sure all processes have reached this point before continuing. Make sure all processes will reach this instruction otherwise one of your processes will hang forever. N)r wait_for_everyonerrrrr:ls r: state_dictcstt}D]\}}t|ts|t||q dd|D}t}| D] }fdd|D}| |dd|ddD]}|=qFq,t |dkr\t d|d d dDS) z Cleans the state dictionary from a model and removes tensor aliasing if present. Args: state_dict (`dict`): The state dictionary from a model cSs"i|] \}}t|dkr||qS)r )len).0ptrnamesrrr s"z4clean_state_dict_for_safetensors..csg|]}|vr|qSrr)r=namer;rr sz4clean_state_dict_for_safetensors..r NrzRemoved shared tensor zk while saving. This should be OK, but check by verifying that you don't receive any warning while reloadingcSs*i|]\}}|t|tjr|n|qSr)rrTensor contiguous)r=kvrrrr@s*) collections defaultdictlistitemsrstrrappendsetvaluesupdater<loggerwarning)r;ptrsrAtensor shared_ptrs warn_namesr? found_namesrrBr clean_state_dict_for_safetensorsys&     rXsave_on_each_nodesafe_serializationcCs|rttddid}t|trt|}ntj}tjt j kr&t ||dStj r3|s3|||dStj r@|rB|||dSdSdS)a Save the data to disk. Use in place of `torch.save()`. Args: obj: The data to save f: The file (or file-like object) to use to save the data save_on_each_node (`bool`, *optional*, defaults to `False`): Whether to only save on the global main process safe_serialization (`bool`, *optional*, defaults to `False`): Whether to save `obj` using `safetensors` or the traditional PyTorch way (that uses `pickle`). formatpt)metadataN)rsafe_save_filerrrXrsaver distributed_typerTPUxmis_main_processis_local_main_process)objfrYrZ save_funcrrrr_s   r_ccs tj}tt_dV|t_dS)a A context manager that will cache origin `os.environ` and replace it with a empty dictionary in this context. When this context exits, the cached `os.environ` will be back. Example: ```python >>> import os >>> from accelerate.utils import clear_environment >>> os.environ["FOO"] = "bar" >>> with clear_environment(): ... print(os.environ) ... os.environ["FOO"] = "new_bar" ... print(os.environ["FOO"]) {} new_bar >>> print(os.environ["FOO"]) bar ``` N)osenvirondict)_old_os_environrrrclear_environments  rlcksi}|D]\}}|}|tjvrtj|||<t|tj|<qdV|D]}|}||vr:||tj|<q(tj|dq(dS)a A context manager that will add each keyword argument passed to `os.environ` and remove them when exiting. Will convert the values in `kwargs` to strings and upper-case all the keys. Example: ```python >>> import os >>> from accelerate.utils import patch_environment >>> with patch_environment(FOO="bar"): ... print(os.environ["FOO"]) # prints "bar" >>> print(os.environ["FOO"]) # raises KeyError ``` N)rKupperrhrirLr2)kwargs existing_varskeyvaluerrrpatch_environments rrcCsHt|dst|dst|d|}t|dr|jSt|dr |jSt|S)z( Gets a pretty name from `obj`. __qualname____name__ __class__)rr0rsrtrL)rerrrget_pretty_names   rvcCs@|D]\}}t|tr||i}t||q|||<q|S)z Recursively merges two dictionaries. Args: source (`dict`): The dictionary to merge into `destination`. destination (`dict`): The dictionary to merge `source` into. )rKrrj setdefault merge_dicts)source destinationrprqnoderrrrx s     rxportreturncCsR|durd}ttjtj}|d|fdkWdS1s"wYdS)z Checks if a port is in use on `localhost`. Useful for checking if multiple `accelerate launch` commands have been run and need to see if the port is already in use. NisL                4 #  #