o Zh5@sddlZddlZddlZddlmZddlmZddlmZm Z m Z ddl m Z ddl Z ddlmZddlmZmZddlmZddlmZdd lmZdd lmZdd lmZdd lmZdd l m!Z!m"Z"ddl#m$Z$m%Z%ddl&m'Z'ddl(m)Z)ddl*m+Z+ddl,m-Z-ddl.m/Z/m0Z0m1Z1gdZ2GdddeZ3e de4d   d2de"de+de ej5de6de7d e e%d!e!fd"d#Z8ed$d%e/dddddd&de"d'e e9ej:dfde e+d e e%de ej5de7d!e!fd(d)Z;ed$d%dddde3j   d2de"de+de ej5de6de7d e e%d!e!fd0d1Z?dS)3N)Future)Enum)castOptionalUnion) deprecated)_copy_state_dict_create_cpu_state_dict)_AsyncCheckpointExecutor)$_ProcessBasedAsyncCheckpointExecutor)#_ThreadBasedAsyncCheckpointExecutor)_storage_setup)DefaultSavePlanner)_dcp_method_logger)MetadataSTATE_DICT_TYPE)SavePlan SavePlanner) AsyncStager)Stateful) StorageWriter)_get_default_group) _api_bc_check _DistWrapper_profile)save_state_dictsave async_saveAsyncCheckpointerTypec@seZdZdZdZdZdS)rz!Enum for async checkpointer type.threadprocessN)__name__ __module__ __qualname____doc__THREADPROCESSr(r(\/var/www/auris/lib/python3.10/site-packages/torch/distributed/checkpoint/state_dict_saver.pyr'srza`save_state_dict` is deprecated and will be removed in future versions.Please use `save` instead.)categoryF state_dictstorage_writer process_groupcoordinator_rankno_distplannerreturncCsD|tt||||||WdS1swYdS)z3This method is deprecated. Please switch to 'save'.N)resetr_save_state_dict)r+r,r-r.r/r0r(r(r)r.s$rT)Zlog_exceptions) checkpoint_idr,r0r-r/r4cCstjd|pt pt }|rtdtt t t ||dd}t t |||||dWdS1s>> # 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.savezptorch.distributed is disabled, unavailable or uninitialized, assuming the intent is to save in a single process.Freader)r+r,r-r/r0N)torch_C_log_api_usage_oncedist is_availableis_initializedwarningswarnrrrr r3_stateful_to_state_dict)r+r4r,r0r-r/r(r(r)rJs$ Y$r)r4r,r0r-async_checkpointer_typer@c Cstjdtrtr|pt}td|jvsJdt t t ||dd}t |}t |tr8||}n t|}t||dd|tjkrKtnt}|j|||||d} t |trd|jrd|| S)aAsynchronous version of ``save``. This code first de-stages the state_dict on to the staging storage (defaults to CPU memory), and then calls the `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 'stage' and 'save'. 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'Fr5)Z type_check)r4r,r0r-)r7r8r9r:r;r<rZdeviceZ _device_typesrrr r? isinstancerZstager rrr'r r Z execute_saveZ should_synchronize_after_executeZsynchronize_staging) r+r4r,r0r-r@ZpgZstaged_state_dictexecutorfr(r(r)rs> 9     rcCs4i}|D]\}}t|tr|n|||<q|S)z]Creates a shallow copy of `state_dict` where `state_dict` is called for each Stateful object.)itemsrBrr+)r+Zstateful_state_dictkeyelemr(r(r)r?s r?c stjdt|| |durtdusJdi}tdd}dur1||d<j|d<tdi|fdd}tdi|fdd} d|| tdi|fd d } tdi|fd d }  d | | S)Nz,torch.distributed.checkpoint.save_state_dictr4r-cstdusJ}dtjjvr tdjn j|jdj } |}|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.)r+rHis_coordinator) rHinspect signatureZset_up_planner parametersr=r>rIZset_up_storage_writerZcreate_local_planZprepare_local_plan)rHZ local_plan)distWr0r+r,r(r) local_step>s   z$_save_state_dict..local_stepcs(dusJ|\}|}|SN)Zcreate_global_planZprepare_global_plan)Zall_local_plans)global_metadatar0r,r(r) global_stepUs  z%_save_state_dict..global_stepZplancs2dusJ}|}||SrO)Z finish_plan write_datawaitvalue)Zfinal_local_planZ all_writes) central_planr0r,r(r)rR`s   z$_save_state_dict..write_datacsdusJj|dS)N)metadataresults)finish)Z all_results)rPr,r(r)finish_checkpointis z+_save_state_dict..finish_checkpointwriter() r7r8r9rrgetattrgrouprZreduce_scatterZ all_reduce) r+r,r-r.r/r0Z ckpt_kwargsZckpt_idrNrQrRrYr()rUrMrPr0r+r,r)r3(s(       r3)NrFN)@rJosr=concurrent.futuresrenumrtypingrrrZtyping_extensionsrr7Ztorch.distributed distributedr:Z#torch.distributed._state_dict_utilsrr Z,torch.distributed.checkpoint._async_executorr Z4torch.distributed.checkpoint._async_process_executorr Z3torch.distributed.checkpoint._async_thread_executorr Z+torch.distributed.checkpoint._storage_utilsr Z,torch.distributed.checkpoint.default_plannerrZ#torch.distributed.checkpoint.loggerrZ%torch.distributed.checkpoint.metadatarrZ$torch.distributed.checkpoint.plannerrrZ$torch.distributed.checkpoint.stagingrZ%torch.distributed.checkpoint.statefulrZ$torch.distributed.checkpoint.storagerZ"torch.distributed.distributed_c10drutilsrrr__all__r FutureWarningZ ProcessGroupintboolrstrPathLikerr&rr?r3r(r(r(r)s               m d