qiHddlZddlZddlZddlmZddlmZddlmZddl m Z m Z m Z m Z ddlmZddlZddlmZddlmZddlmZdd lmZdd lmZdd lmZdd lmZdd l m!Z!ddl"m#Z#m$Z$ddl%m&Z&m'Z'm(Z(ddl)m*Z*ddl+m,Z,m-Z-ddl.m/Z/ddl0m1Z1m2Z2m3Z3e rddl4m5Z5gdZ6GddeZ7ede8 d2dede,de ejrde:d e;d!e e$d"e!fd#Z<ed$%e1dddddd$d&ded'e e=ej|dfde e,d!e e$de ejrd e;d(e;d"e!fd)Z?eGd*d+Z@ed$%dddde7jddd$d,ded'e e=ej|dfde e,d!e e$de ejrd-e7d.e e&d e;d(e;d"e ee@ffd/ZBed$%ded"efd0ZC d3dede,de ejrde:d e;d!e e$d(e;d"e!fd1ZDy)4N)Future) dataclass)Enum)castOptional TYPE_CHECKINGUnion) 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_saveAsyncCheckpointerTypeAsyncSaveResponseceZdZdZdZdZy)r#z!Enum for async checkpointer type.threadprocessN)__name__ __module__ __qualname____doc__THREADPROCESSs/home/ubuntu/crypto_trading_bot/.venv/lib/python3.12/site-packages/torch/distributed/checkpoint/state_dict_saver.pyr#r#3s+ FGr/r#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 |jt5t||||||cdddS#1swYyxYw)z3This method is deprecated. Please switch to 'save'.N)resetr_save_state_dict)r2r3r4r5r6r7s r0r r :sF             s5>Tlog_exceptions checkpoint_idr3r7r4r6use_collectivesr?r@c ~tjjd|xs,tj xstj }|rt jddt5ttt||d}tt||||||cdddS#1swYyxYw) 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)r2r3r4r6r7r@N)torch_C_log_api_usage_oncedist is_availableis_initializedwarningswarnrrrrr;_stateful_to_state_dict)r2r?r3r7r4r6r@s r0r!r!Vs~ HH  !DEQd//11Q4;N;N;P7PG ~    >.-PUV  .z:)'+     s 36B33B<c2eZdZUdZeded<eded<y)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/r0r$r$st $d|#r/r$)r?r3r7r4async_checkpointer_type async_stagerr6r@rSrTc tjjdtjrNtj r:|xs t } tjd| jvr td,|t|tr|nttddddttddt t"t$t$fffd } | } |t&j(k(r t+n t-} | j/| |||||| } t| t"rh| }t#}|fd t"t$d t"dfd }|j1s|j3|n|j5dt7|| Stdfd}|| 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'NFTr<r8c&jSN)stage)rTr2sr0stage_state_dictz$async_save..stage_state_dict>s!!*--r/r>original_staging_futurereturn_staging_futurec |j|jdy#t$r}|j|Yd}~yd}~wwxYwrX)result set_result Exception set_exception)r[r\es r0callbackzasync_save..callbackYsB 7'..0%006 7%33A66 7s!$ AAA)rPrQc@jrjyyrX) should_synchronize_after_executesynchronize_staging)rTsr0maybe_synchronize_stagingz-async_save..maybe_synchronize_stagingns<<002=r/)rFrGrHrIrJrKrdevice _device_typesAssertionError isinstancerrrrNrr rr r#r-r r execute_savedoneadd_done_callbackr_r$)r2r?r3r7r4rSrTr6r@pgrZstaging_future_or_state_dictupload_executor upload_futurestaging_futurer\rcrgs` ` r0r"r"sT HH  !JK t224  202 << b&6&6 6 x   %*^[*Q)L( L)4Jt,.eF?$;_$LM.-.$4#5  #&;&C&C C -. 0 2 ,88$#%#'9 M.75.4h3H 7%+O%< 7#)$< 7""$  , ,X 6 ! , ,T 2!4   4 0 3 1 3 "#r/ci}|jD]-\}}d}d||_td||||</|S)z]Creates a shallow copy of `state_dict` where `state_dict` is called for each Stateful object.cFt|tr|jS|SrX)rkrr2)elems r0_elem_to_state_dictz4_stateful_to_state_dict.._elem_to_state_dict}s(24(B4??$ L Lr/z_stateful_to_state_dict.Tr<)itemsr(r)r2stateful_state_dictkeyrvrws r0rNrNwso%%'  T M*B#'G$$ #J#5T#J $ $C   r/c`tjjdt|| | t  t ddi}t ddx}||d<j|d<td i|fd} td i|fd} drjd| | n| } | | g} | dtd i|fd } td i|fd }rjd | |}|S| }||g}j|S) Nz,torch.distributed.checkpoint.save_state_dictplanner is Noner?r4c\ tdj}dtjjj vr4t jddj jnj |jdtjjj vr)jjjnjjj}j|}|S) 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.rBrC)r2r~is_coordinatorkwargs)rankr@) rjr~inspect signatureset_up_planner parametersrLrMrset_up_storage_writerrcreate_local_planprepare_local_plan)r~ local_plandistWr7r2r3r@s r0 local_stepz$_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 r/cl tdj|\}j|}|S)Nr|)rjcreate_global_planprepare_global_plan)all_local_plansglobal_metadatar7r3s r0 global_stepz%_save_state_dict..global_stepsA ? !23 3+2+E+Eo+V((<<_Mr/planrc td tdj}j|}|j|j S)Nr|zcentral_plan is None)rj finish_plan write_datawaitvalue)final_local_plan all_writes central_planr7r3s r0rz$_save_state_dict..write_datasc ? !23 3   !78 8"..|<#../?I !!r/cH tdj|S)Nzglobal_metadata is None)metadataresults)rjfinish) all_resultsrr3s r0finish_checkpointz+_save_state_dict..finish_checkpoints.  " !:; ; Lr/writer.) rFrGrHrrrjgetattrgrouprreduce_scatter all_reducebarrier)r2r3r4r5r6r7r@ ckpt_kwargsckpt_idrrr global_planrrr write_resultsrrrs`` `` @@@r0r;r;sh HH  !OP G 5E FE$&.//OK>?DAAN'. O$',{{ O$&+&!'!F&+&'(,L++FJ L )| &1:,&? "1~ &+& "' "&+&' ##GZ9JK O ,6< $m_5  Or/)NrFN)NrFNT)ErosrLconcurrent.futuresr dataclassesrenumrtypingrrrr typing_extensionsr rFtorch.distributed distributedrI#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 ProcessGroupintboolr strPathLiker!r$r,r"rNr;r.r/r0rsI %!77( ?GKB:F ;KA88U D ! 26%)   ! D--.     k "     .4(48.2%)15 r r bkk4/0r ]+ r k " r D--. r r r r )r j  $ $  $4(48.2%)155J5Q5Q*. XXbkk4/0X]+ X k " X D--. X3X;'XXX 6$ $%X)Xv4(  O ) &26%) hh!hD--.h h  h k " hhhr/