`iFddlZddlZddlZddlmZddlmZddlmZddl m Z m Z m Z ddl mZddlZddlmZddlmZddlmZdd lmZdd lmZdd lmZdd lmZdd lm Z ddl!m"Z"ddl#m$Z$m%Z%ddl&m'Z'm(Z(m)Z)ddl*m+Z+ddl,m-Z-m.Z.ddl/m0Z0ddl1m2Z2m3Z3m4Z4gdZ5GddeZ6ede7 d2dede-de ej8de9d e:d!e e%d"e"fd#Z;e d$%e2dddddd$d&ded'e eeGd*d+Z?e d$%dddde6j@ddd$d,ded'e er?c tjd|p'tj ptj }|rt jdt5ttt||d}tt||||||cdddS#1swxYwYdS)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 specified, the default planner will be used. (Default: ``None``) process_group (Optional[ProcessGroup]): ProcessGroup to be used for cross-rank synchronization. (Default: ``None``) no_dist (bool): If ``True``, this function will assume the intent is to load a checkpoint on a single rank/process. (Default: ``False``) use_collectives (bool): If ``False``, this function will assume the intent is to save a checkpoint without using cross-rank synchronization. (Default: ``True``) This configuration is experimental and should be used with caution. It will change the format of the saved checkpoint and may not be backward compatible. 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.savezptorch.distributed is disabled, unavailable or uninitialized, assuming the intent is to save in a single process.Freader)r1r2r3r5r6r?N)torch_C_log_api_usage_oncedist is_availableis_initializedwarningswarnrrrrr:_stateful_to_state_dict)r1r>r2r6r3r5r?s r/r r Us$~ H  !DEEEQd/111Q4;N;P;P7PG  ~        >.-PUVVV   .z::)'+                      s.ACC Cc@eZdZUdZeded<eded<dS)r#a!This class contains futures for staging and upload completion. It is returned by async_save(). staging_completion is a future that indicates when local copy of state_dict is complete. upload_completion is a future that indicates when a checkpoint completed saving. Nstaging_completionupload_completion)r'r(r)r*r__annotations__r-r.r/r#r#s>t $$$d|#####r.r#)r>r2r6r3async_checkpointer_type async_stagerr5r?rPrQc tjdtjrHtjr5|p t } tjd| jvs Jd9|t|tr|nttddddttt||d}t!t#dd t$t&t(t(fffd } | } |t*jkrt/n t1} | | |||||| } t| t&r| }t'}|fd t&t(d t&dfd}|s||n|dt;|| St#dfd}|| S)a Asynchronous 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. MUST CALL CLOSE AFTER LAST CHECKPOINT IS SAVED 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 specified, the default planner will be used. (Default: ``None``) process_group (Optional[ProcessGroup]): ProcessGroup to be used for cross-rank synchronization. (Default: ``None``) async_checkpointer_type (AsyncCheckpointerType): whether to do checkpoint in separate thread or process (Default: ``AsyncCheckpointerType.THREAD``) async_stager (AsyncStager): provides staging implementation. If storage_writer implements AsyncStager and async_stager is provided, async_stager will be used for staging no_dist (bool): If ``True``, this function will assume the intent is to save a checkpoint on a single rank/process. (Default: ``False``) use_collectives: If False, Save the checkpoint without rank coordination. (Default: ``True``) This configuration is experimental and should be used with caution. It will change the format of the saved checkpoint and may not be backward compatible. 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'NFrATr;r7c.SN)stage)rQr1sr/stage_state_dictz$async_save..stage_state_dictAs!!*---r.r=original_staging_futurereturn_staging_futurec ||ddS#t$r }||Yd}~dSd}~wwxYwrU)result set_result Exception set_exception)rXrYes r/callbackzasync_save..callback[sx 7'..000%0066666 7 7 7%33A666666666 7s)- AAA)rMrNcBjrdSdSrU) should_synchronize_after_executesynchronize_staging)rQsr/maybe_synchronize_stagingz-async_save..maybe_synchronize_stagingps0< 30022222 3 3r.)rCrDrErFrGrHrdevice _device_types isinstancerrrrrrrKrrrr r"r,r r execute_savedoneadd_done_callbackr\r#)r1r>r2r6r3rPrQr5r?pgrWstaging_future_or_state_dictupload_executor upload_futurestaging_futurerYr`rds` ` r/r!r!sT H  !JKKK  t244  2022 L  2#3 3 3 3 t 4 3 3   %*^[*Q*Q %)LL( L~nmERRRN)44Jt,,,.eF?$;_$LM......-,.$4#3#5#5  #&;&C C C -... 0 2 2 ,88$#%#'9M.775.4hh3H 7 7%+O%< 7#)$< 7 7 7 7""$$ 3  , ,X 6 6 6 6 ! , ,T 2 2 2!4     4 0 0 0 3 3 3 3 1 0 3 "!###r.ci}|D]3\}}t|tr|n|||<4|S)z]Creates a shallow copy of `state_dict` where `state_dict` is called for each Stateful object.)itemsrgrr1)r1stateful_state_dictkeyelems r/rKrKys_%%''  T!+D(!;!; EDOO    C   r.ctjdt|| |t Jdi}t ddx}||d<j|d<td i|fd} td i|fd} drd| | n| } | | g} | dtd i|fd} td i|fd }r d | |}n*| }||g} |S) Nz,torch.distributed.checkpoint.save_state_dictr>r3cJ}dtjjjvr0t jdjn|jdtjjjvr#jj nj } |}|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.)r1rwis_coordinatorkwargs)rankr?) rwinspect signatureset_up_planner parametersrIrJrxset_up_storage_writerrzcreate_local_planprepare_local_plan)rw local_plandistWr6r1r2r?s r/ local_stepz$_save_state_dict..local_steps6"""%2244 !273I!J!J!U U U M.     " ":u/C D D D D  " "%)$3 #     !EFFQ R R  0 0$Z / 1      0 01E F F F..00 #66zBB r.cjJ|\}|}|SrU)create_global_planprepare_global_plan)all_local_plansglobal_metadatar6r2s r/ global_stepz%_save_state_dict..global_stepsB"""+2+E+Eo+V+V((<<_MMr.planrcJJ}|}||SrU) finish_plan write_datawaitvalue)final_local_plan all_writes central_planr6r2s r/rz$_save_state_dict..write_datasg"""'''"..|<<#../?II !!!r.c>J|S)N)metadataresults)finish) all_resultsrr2s r/finish_checkpointz+_save_state_dict..finish_checkpoints.*** LLLr.writer-) rCrDrErrgetattrgrouprreduce_scatter all_reducebarrier)r1r2r3r4r5r6r? ckpt_kwargsckpt_idrrr global_planrrr write_resultsrrrs`` `` @@@r/r:r:s# H  !OPPP G 5E F FE$&&    OK>?DAAAN'. O$',{ O$&&+&&'&B&&+&&'&(,L&++FJ LL )z|| &1k:,&?&? "1~ &&+&&""""""'&"&&+&&'& ##GZ9JKK+5:<< $$m_55  Or.)NrFN)NrFNT)Dr{osrIconcurrent.futuresr dataclassesrenumrtypingrrrtyping_extensionsr rCtorch.distributed distributedrF#torch.distributed._state_dict_utilsr ,torch.distributed.checkpoint._async_executorr 4torch.distributed.checkpoint._async_process_executorr 3torch.distributed.checkpoint._async_thread_executorr +torch.distributed.checkpoint._storage_utilsr,torch.distributed.checkpoint.default_plannerr#torch.distributed.checkpoint.loggerr%torch.distributed.checkpoint.metadatar$torch.distributed.checkpoint.plannerrr$torch.distributed.checkpoint.stagingrrr%torch.distributed.checkpoint.statefulr$torch.distributed.checkpoint.storagerr"torch.distributed.distributed_c10drutilsrrr__all__r" FutureWarning ProcessGroupintboolrstrPathLiker r#r+r!rKr:r-r.r/rs %%%%%%!!!!!!(((((((((((((((( ??????GFFFFFKKKKKKBBBBBB::::::FFFFFFFF ;:::::KKKKKKKKAAAAAA8888888888   D ! 26%)    ! D-.     k "       .4(((48.2%)15 q q q q bk4/0q ]+ q k " q D-. q q q q q q )(q h  $ $ $ $ $ $ $  $4(((48.2%)155J5Q*. \\\\bk4/0\]+ \ k " \ D-. \3\;'\\\ 6$ $%\\\)(\~4(((O)(26%) aaa!aD-.a a  a k " aaaaaaaar.