ȅi+HSSKrSSKrSSKrSSKJr SSKJr SSKJr SSK J r J r SSK J r SSKrSSKJr SSKJr SSKJr SS KJr SS KJr SS KJr SS KJr SS KJr SSK J!r!J"r" SSK#J$r$J%r%J&r& SSK'J(r( SSK)J*r*J+r+ SSK,J-r- SSK.J/r/J0r0J1r1 \ (aSSK2J3r3 /SQr4"SS\5r5\ "S\6S9S2S\S\*S\RnS-S\8S \9S!\"S-S"\4S#jj5r:\"S$S%9\/SSSSSS$S&.S\S'\;\Rx-S-S\*S-S!\"S-S\RnS-S \9S(\9S"\4S)jj55r=\"S*S+55r>\"S$S%9SSSS\5R~SSS$S,.S\S'\;\Rx-S-S\*S-S!\"S-S\RnS-S-\5S.\$S-S \9S(\9S"\\>-4S/jj5r@\"S$S%9S\S"\4S0j5rAS3S\S\*S\RnS-S\8S \9S!\"S-S(\9S"\4S1jjrBg)4N)Future) dataclass)Enum)cast TYPE_CHECKING) deprecated)STATE_DICT_TYPE)$_ProcessBasedAsyncCheckpointExecutor)#_ThreadBasedAsyncCheckpointExecutor)_storage_setup)DefaultSavePlanner)_dcp_method_logger)Metadata)SavePlan SavePlanner) AsyncStager DefaultStagerStagingOptions)Stateful) StorageWriter WriteResult)_get_default_group) _api_bc_check _DistWrapper_profile)_AsyncCheckpointExecutor)save_state_dictsave async_saveAsyncCheckpointerTypeAsyncSaveResponsec \rSrSrSrSrSrSrg)r!3z!Enum for async checkpointer type.threadprocessN)__name__ __module__ __qualname____firstlineno____doc__THREADPROCESS__static_attributes__r'g/home/james-whalen/.local/lib/python3.13/site-packages/torch/distributed/checkpoint/state_dict_saver.pyr!r!3s+ FGr0r!za`save_state_dict` is deprecated and will be removed in future versions.Please use `save` instead.)categoryF state_dictstorage_writer process_groupcoordinator_rankno_distplannerreturnc UR5 [5 [UUUUUU5sSSS5 $!,(df  g=f)z3This method is deprecated. Please switch to 'save'.N)resetr_save_state_dict)r3r4r5r6r7r8s r1rr:s;          s 5 ATlog_exceptions checkpoint_idr4r8r5r7use_collectivesr@rAc [RRS5 U=(d; [R"5(+=(d [R "5(+nU(a[ R"SSS9 [5 [[[X!SS95n[[U5UUUUUS9sSSS5 $!,(df  g=f) 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. stacklevelF)reader)r3r4r5r7r8rAN)torch_C_log_api_usage_oncedist is_availableis_initializedwarningswarnrrrr r<_stateful_to_state_dict)r3r@r4r8r5r7rAs r1rrVs~ HH  !DEQd//11Q4;N;N;P7PG ~  >.PUV  .z:)'+  s /C Cc:\rSrSr%Sr\S\S'\S\S'Srg)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_completionr')r(r)r*r+r,r__annotations__r/r'r0r1r"r"st $d|#r0r")r@r4r8r5async_checkpointer_type async_stagerr7rArUrVc ^^[RRS5 [R"5(a\[R "5(aBU=(d [ 5n [R"S5U R;a [S5eTc2Ub[U[5(aUmO[[SSSS55m[T5m[SS9S[ ["["-4UU4S jj5n U "5n U[$R&:Xa [)5O [+5n U R-U UUUUUUS 9n [U [ 5(akU n[!5nU4S [ ["S [ S4S jjnUR/5(dUR1U5 OUR3S5 [5XS9$[SS9U4Sj5nU"5 U $)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'NFTr=r9c&>TRT5$N)stage)rVr3sr1stage_state_dict$async_save..stage_state_dict>s!!*--r0r?original_staging_futurereturn_staging_futurecUR5 URS5 g![anURU5 SnAgSnAff=frZ)result set_result Exception set_exception)r^r_es r1callbackasync_save..callbackXsB 7'..0%006 7%33A66 7s!$ A AA )rRrScJ>TR(aTR5 ggrZ) should_synchronize_after_executesynchronize_staging)rVsr1maybe_synchronize_staging-async_save..maybe_synchronize_stagingms<<002=r0)rGrHrIrJrKrLrdevice _device_typesAssertionError isinstancerrrrOrrr r!r.r r execute_savedoneadd_done_callbackrbr")r3r@r4r8r5rUrVr7rApgr\staging_future_or_state_dictupload_executor upload_futurestaging_futurer_rfrks` ` r1r r sT HH  !JK t2244  202 << b&6&6 6 x   %*^[*Q*Q)L( L)4Jt,.f_5G.-.$4#5  #&;&C&C C -. 0 2 ,88$#%#'9M.775.4h3H 7%+O%< 7#)$< 7""$$  , ,X 6 ! , ,T 2!4  4 0 3 1 3 "#r0c0nUR5H)up#SnSU3Ul[SS9"U5"U5X'M+ U$)z]Creates a shallow copy of `state_dict` where `state_dict` is called for each Stateful object.cP[U[5(aUR5$U$rZ)rprr3)elems r1_elem_to_state_dict4_stateful_to_state_dict.._elem_to_state_dict|s (24(B(B4??$ L Lr0z_stateful_to_state_dict.Tr=)itemsr(r)r3stateful_state_dictkeyr{r|s r1rOrOvs^%%'  M*B#'G$#5T#J $ $ ( r0c^^^^^^^[RRS5 [X$(+U5mTc [ 5mTc [ S5eSm0n[ TSS5=nbXS'TRUS'[S 0UD6UUUUU4Sj5n [S 0UD6UUU4Sj5n SmT(aTRSX5mOU "5n U "U /5n U Sm[S 0UD6UUU4S j5n [S 0UD6UU4S j5nT(aTRS X5nU$U "5nU"U/5nTR5 U$) Nz,torch.distributed.checkpoint.save_state_dictplanner is Noner@r5cZ>Tc [S5eTR5nS[R"TR5R ;a2[ R"SSS9 TR TTR5 OTR TUTRS9 S[R"TR5R ;a&TRTRTRTS9 OTRTR5 TR5nTRU5nU$) Nr 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.rCrD)r3ris_coordinatorkwargs)rankrA) rorinspect signatureset_up_planner parametersrMrNrset_up_storage_writerrcreate_local_planprepare_local_plan)r local_plandistWr8r3r4rAs r1 local_step$_save_state_dict..local_steps ? !23 3%224 !2!273I3I!J!U!U U MM.    " ":u/C/C D  " "%)$33 #    !E!EFQQ R  0 0$$ZZ / 1   0 01E1E F..0 #66zB r0cn>Tc [S5eTRU5unmTRU5nU$)Nr)rocreate_global_planprepare_global_plan)all_local_plansglobal_metadatar8r4s r1 global_step%_save_state_dict..global_stepsA ? !23 3+2+E+Eo+V((<<_Mr0planrc>Tc [S5eTc [S5eTRT5nTRUT5nUR5 UR 5$)Nrzcentral_plan is None)ro finish_plan write_datawaitvalue)final_local_plan all_writes central_planr8r4s r1r$_save_state_dict..write_datasc ? !23 3   !78 8"..|<#../?I !!r0cD>Tc [S5eTRTUS9 T$)Nzglobal_metadata is None)metadataresults)rofinish) all_resultsrr4s r1finish_checkpoint+_save_state_dict..finish_checkpoints.  " !:; ; Lr0writer') rGrHrIrr rogetattrgrouprreduce_scatter all_reducebarrier)r3r4r5r6r7r8rA ckpt_kwargsckpt_idrrr global_planrrr write_resultsrrrs`` `` @@@r1r<r<sd HH  !OP  5E FE$&.//OK>?DAAN'.O$',{{ O$&+&!!'!F&+&'%)L++FJL )| &1:,&? "1~ &+& "' "&+&' ##GZK O ,6< $m_5  Or0)NrFN)NrFNT)CrosrMconcurrent.futuresr dataclassesrenumrtypingrrtyping_extensionsrrGtorch.distributed distributedrJ#torch.distributed._state_dict_utilsr 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,torch.distributed.checkpoint._async_executorr__all__r! FutureWarning ProcessGroupintboolrstrPathLikerr"r-r rOr<r'r0r1rs= %!&( ?GKB:F ;KA88U D ! /3"&   ! $$t+     4      .4(/3+/"&.2 r r $t+r "D( r 4  r $$t+ r r r r )r j  $ $  $4(/3+/"&.25J5Q5Q'+ WW$t+W"D( W 4  W $$t+ W3W$WWW W)Wt4(  O ) &/3"& hh!h$$t+h h  h 4  hhhr0