U yh.1 @sddlZddlZddlZddlmZmZddlmZmZm Z ddl m Z ddl Z ddl mZddlmZddlmZddlmZddlmZdd lmZmZdd lmZmZdd lmZdd l m!Z!dd l"m#Z#ddl$m%Z%ddl&m'Z'm(Z(m)Z)dddgZ*e de+dd$ee#eej,e-e.eeedddZ/edde'dddddee e0ej1dfee#eeeej,edddZ2edddddddee e0ej1dfee#eeeej,edddZ3eedd d!Z4d%ee#eej,e-e.eeedd"d#Z5dS)&N)FutureThreadPoolExecutor)castOptionalUnion) deprecated)_offload_state_dict_to_cpu)_storage_setup)DefaultSavePlanner)_dcp_method_logger)MetadataSTATE_DICT_TYPE)SavePlan SavePlanner) AsyncStager)Stateful) StorageWriter)_get_default_group) _api_bc_check _DistWrapper_profilesave_state_dictsave async_saveza`save_state_dict` is deprecated and will be removed in future versions.Please use `save` instead.)categoryF) state_dictstorage_writer process_groupcoordinator_rankno_distplannerreturnc Cs8|t t||||||W5QRSQRXdS)z3This method is deprecated. Please switch to 'save'.N)resetr_save_state_dict)rrrrr r!r%_/var/www/html/venv/lib/python3.8/site-packages/torch/distributed/checkpoint/state_dict_saver.pyrsT)Zlog_exceptions checkpoint_idrr!r)rr(rr!rr"c Csttjdtot }|r,tdt8t t t ||dd}t t |||||dW5QRSQRXdS)a Save a distributed model in SPMD style. This function is different from ``torch.save()`` as it handles ``ShardedTensor`` , and ``DTensor`` by having each rank only save their local shards. For each ``Stateful`` object (having both a ``state_dict`` and a ``load_state_dict``), save will call ``state_dict`` before serialization. .. warning:: There is no guarantees of Backwards Compatibility across PyTorch versions for saved state_dicts. .. warning:: If using the `process_group` argument, make sure that only its ranks call `save_state_dict` and that all data in state_dict belong to it. .. note:: When saving checkpoint for FSDP's `ShardingStrategy.HYBRID_SHARD`, only one of the shard_group should be calling `save_state_dict` and the corresponding process group needs to be passed in. .. note:: If no process group is available, this function assumes the intention is to save the state_dict in the local process. .. note: Rank 0 is assumed to be the coordinator rank. Args: state_dict (Dict[str, Any]): The state_dict to save. checkpoint_id (Union[str, os.PathLike, None]): The ID of this checkpoint instance. The meaning of the checkpoint_id depends on the storage. It can be a path to a folder or to a file. It can also be a key if the storage is a key-value store. (Default: ``None``) storage_writer (Optional[StorageWriter]): Instance of StorageWriter used to perform writes. If this is not specified, DCP will automatically infer the writer based on the checkpoint_id. If checkpoint_id is also None, an exception will be raised. (Default: ``None``) planner (Optional[SavePlanner]): Instance of SavePlanner. If this is not specificed, the default planner will be used. (Default: ``None``) process_group (Optional[ProcessGroup]): ProcessGroup to be used for cross-rank synchronization. (Default: ``None``) Returns: Metadata: Metadata object for the saved checkpoint. Example: >>> # xdoctest: +SKIP >>> my_model = MyModule() >>> state_dict = {"model": my_model} >>> fs_storage_writer = torch.distributed.checkpoint.FileSystemWriter("/checkpoint/1") >>> torch.distributed.checkpoint.save( >>> state_dict=state_dict, >>> storage_writer=fs_storage_writer, >>> ) .. note:: save_state_dict uses collectives to coordinate writes across ranks. For NCCL-based process groups, internal tensor representations of objects must be moved to the GPU device before communication takes place. In this case, the device used is given by ``torch.cuda.current_device()`` and it is the user's responsibility to ensure that this is set so that each rank has an individual GPU, via ``torch.cuda.set_device()``. z!torch.distributed.checkpoint.savezftorch.distributed is unavailable or uninitialized, assuming the intent is to save in a single process.Freader)rrrr r!N)torch_C_log_api_usage_oncedist is_availableis_initializedwarningswarnrrrr r$_stateful_to_state_dict)rr(rr!rr r%r%r&r9s$R  cstjdtr>tr>|p$t}td|jks>t dt t t ||dd}t |}t|trp||}n t|dd}tddjt|||||d }|fd d t|tr|jr||S) a$Asynchronous version of ``save``. This code first de-stages the state_dict on CPU, and then calls `save` in a separate thread. .. warning:: This feature is experimental and subject to change. Args: state_dict (Dict[str, Any]): The state_dict to save. checkpoint_id (Union[str, os.PathLike, None]): The ID of this checkpoint instance. The meaning of the checkpoint_id depends on the storage. It can be a path to a folder or to a file. It can also be a key if the storage is a key-value store. (Default: ``None``) storage_writer (Optional[StorageWriter]): Instance of StorageWriter used to perform writes. If this is not specified, DCP will automatically infer the writer based on the checkpoint_id. If checkpoint_id is also None, an exception will be raised. (Default: ``None``) planner (Optional[SavePlanner]): Instance of SavePlanner. If this is not specificed, the default planner will be used. (Default: ``None``) process_group (Optional[ProcessGroup]): ProcessGroup to be used for cross-rank synchronization. (Default: ``None``) Returns: Future: A future holding the resultant Metadata object from `save`. Example: >>> # xdoctest: +SKIP >>> my_model = MyModule() >>> state_dict = {"model": my_model} >>> fs_storage_writer = torch.distributed.checkpoint.FileSystemWriter("/checkpoint/1") >>> checkpoint_future = torch.distributed.checkpoint.async_save( >>> state_dict=state_dict, >>> storage_writer=fs_storage_writer, >>> ) >>> >>> # ... do some work ... >>> >>> checkpoint_future.result() z'torch.distributed.checkpoint.async_savecpuzfA CPU backend must be enabled for async save; try initializing process group with 'cpu:gloo,cuda:nccl'Fr))Z type_checkr) max_workersr'cs jddS)NF)wait)shutdown)fexecutorr%r&zasync_save..)r+r,r-r.r/r0rZdeviceZ _device_typesAssertionErrorrrr r3 isinstancerZstagerrZsubmitrZadd_done_callbackZ should_synchronize_after_executeZsynchronize_staging)rr(rr!rZpgZstaged_state_dictr8r%r9r&rs>6       )rr"cCs4i}|D]"\}}t|tr&|n|||<q |S)z]Creates a shallow copy of `state_dict` where `state_dict` is called for each Stateful object.)itemsr>rr)rZstateful_state_dictkeyelemr%r%r&r3s r3c stjdt|| |dkr(tdk s4tdi}tdd}dk rX||d<tf|fdd}tf|fdd} d|| tf|fdd } tf|fd d }  d | | S) Nz,torch.distributed.checkpoint.save_state_dictr(cstdk s t}dtjjkr@tdjnj|jd j } |}|S)N storage_metazThe function definition for SavePlanner.set_up_planner has been updated to include the storage_meta argument. Please update your implementation to include this parameter.)rrBis_coordinator) r=rBinspect signatureZset_up_planner parametersr1r2rCZset_up_storage_writerZcreate_local_planZprepare_local_plan)rBZ local_plan)distWr!rrr%r& local_steps    z$_save_state_dict..local_stepcs(dk s t|\}|}|SN)r=Zcreate_global_planZprepare_global_plan)Zall_local_plans)global_metadatar!rr%r& global_step3s  z%_save_state_dict..global_stepZplancs2dk s t}|}||SrI)r=Z finish_plan write_datar6value)Zfinal_local_planZ all_writes) central_planr!rr%r&rL>s    z$_save_state_dict..write_datacsdk s tj|dS)N)metadataresults)r=finish)Z all_results)rJrr%r&finish_checkpointGs z+_save_state_dict..finish_checkpointwrite) r+r,r-rr r=getattrr Zreduce_scatterZ all_reduce) rrrrr r!Z ckpt_kwargsZckpt_idrHrKrLrRr%)rNrGrJr!rrr&r$s&  r$)NrFN)NrFN)6rDosr1concurrent.futuresrrtypingrrrtyping_extensionsrr+Ztorch.distributed distributedr.Z#torch.distributed._state_dict_utilsrZ+torch.distributed.checkpoint._storage_utilsr Z,torch.distributed.checkpoint.default_plannerr Z#torch.distributed.checkpoint.loggerr Z%torch.distributed.checkpoint.metadatar r Z$torch.distributed.checkpoint.plannerrrZ$torch.distributed.checkpoint.stagingrZ%torch.distributed.checkpoint.statefulrZ$torch.distributed.checkpoint.storagerZ"torch.distributed.distributed_c10drutilsrrr__all__ FutureWarningZ ProcessGroupintboolrstrPathLikerrr3r$r%r%r%r&s           f[