U
    ©Ê¹j2  ã                   @   s  d dl Z d dlmZmZmZmZmZmZmZ d dl	m
Z
 d dlZd dlmZ d dlmZ d dlmZ d dlmZ d dlmZ d dlm  mZ d dlmZmZmZmZmZm Z  d d	l!m"Z" d d
l#m$Z$ d dl%m&Z&m'Z' d dl(m)Z) d dl*m+Z+ d dl,m-Z- d dl	m.Z. d dl/m0Z0 d dl1m2Z2m3Z3m4Z4 d dl5m6Z6 ee7eeee8  ee8 f f Z9dgZ:d(e8e7e7dœdd„Z;d)eej< edœdd„Z=ej>e?dœdd„Z@d*eee8 e7ej>dœdd „ZAeee9eej< f d!œd"d#„ZBG d$d%„ d%e-ƒZCd+ee7ejDee. ed&œd'd„ZEdS ),é    N)ÚDictÚListÚOptionalÚSequenceÚTupleÚUnionÚcast)ÚLoadPlan)ÚShardedTensor)ÚTensorProperties)ÚShard)ÚChunkShardingSpec)ÚBytesStorageMetadataÚMetadataÚMetadataIndexÚSTATE_DICT_TYPEÚTensorStorageMetadataÚChunkStorageMetadata)Ú_get_default_group)Ú_create_chunk_sharded_tensor)Ú create_read_items_for_chunk_listÚ_create_read_items)Ú_remote_device)ÚDTensor)ÚDefaultLoadPlanner)ÚLoadPlanner)Úunflatten_state_dict)Ú_element_wise_addÚ_element_wise_subÚ_normalize_device_info)Ú_get_device_moduleÚ!load_sharded_optimizer_state_dictÚcuda)Úglobal_rankÚdevice_typeÚreturnc                 C   s2   |dkrdS t |ƒ}| ¡ r.t|| | ¡  ƒS dS )NÚcpu)r    Zis_availabler   Údevice_count)r#   r$   Údevice_module© r)   úe/var/www/html/bwave/python/venv/lib/python3.8/site-packages/torch/distributed/checkpoint/optimizer.pyÚ_gen_rank_device:   s    r+   )Úpgr%   c                    sl   t j ˆ ¡j‰ˆ d kr2‡fdd„tt  ¡ ƒD ƒ}n‡ ‡fdd„tˆ  ¡ ƒD ƒ}tdtt	t
ttf  |ƒd�S )Nc                    s"   g | ]}d |› dt |ˆ ƒ› �‘qS ©úrank:ú/)r+   ©Ú.0Úidx)Úpg_device_typer)   r*   Ú
<listcomp>H   s   ÿz(_create_colwise_spec.<locals>.<listcomp>c              
      s*   g | ]"}d |› dt t ˆ |¡ˆƒ› �‘qS r-   )r+   ÚdistZget_global_rankr0   ©r,   r3   r)   r*   r4   M   s   ÿr   ©ÚdimÚ
placements)r5   Údistributed_c10dÚ_get_pg_default_deviceÚtypeÚrangeÚget_world_sizeÚsizer   r   r   r   r   Ústr)r,   r9   r)   r6   r*   Ú_create_colwise_specC   s    

þ
þþrA   )Úvalr%   c                 C   sŽ   t | ƒtkrZt|  ¡ ƒdkr dS t |  ¡ d jƒtkr:dS t |  ¡ d jƒtkrŠtdƒ‚n0t | ƒtkrŠt | jƒtks‚t | jƒtkrŠtdƒ‚dS )Nr   FTz2Cannot handle DTensor nested insided ShardedTensorzCannot handle nested DTensor)r<   r
   ÚlenÚlocal_shardsÚtensorr   Ú
ValueErrorZ_local_tensor)rB   r)   r)   r*   Ú_is_nested_tensorW   s     ÿÿþrG   )Úpropsr?   r$   r%   c              
   C   s.   t j|| j| j| j| jtt jt|ƒ 	¡ ƒd�S )N)r?   ÚdtypeÚlayoutÚrequires_gradÚ
pin_memoryÚdevice)
ÚtorchÚemptyrI   rJ   rK   rL   r   rM   r    Zcurrent_device)rH   r?   r$   r)   r)   r*   Ú_alloc_tensori   s    úrP   )Ú
state_dictr%   c                 C   sŒ   i }d}|   ¡ D ]r\}}d| ¡ f||< t|ƒrt| ¡ ƒdksHtdƒ‚t|tƒsZtdƒ‚| ¡ d }|jj	|jj
f||< |jj}q||fS )a5  
    We have to load the right TP slice of the optimizer state.
    This is not easy since the per-tensor slicing can't be inferred from checkpoint metadata.
    We take advantage of the model state_dict producing a sliced ST to figure out what we need to load.
    This is pretty fragile and it might be easier for FSDP to compute this info for us.
    Returns a dictionary where keys are the same of the state_dict and the value is a tuple of
    (offset, size) for the current rank TP slice.
    N.B. The state_dict *MUST* come from FSDP.sharded_state_dict.
    Né   z%Cannot handle ST with multiple shardsz$Can only handle nested ShardedTensorr   )Úitemsr?   rG   rC   rD   ÚAssertionErrorÚ
isinstancer
   ÚmetadataÚshard_offsetsÚshard_sizesrE   Z_process_group)rQ   ÚspecsÚdp_pgÚkeyÚvalueZshardr)   r)   r*   Ú_get_state_dict_2d_layoutt   s.    ÿþ ÿþþ
þr]   c                       sv   e Zd ZU eeef ed< eed< eed< eee	e
 f ddœ‡ fdd„Zedœd	d
„Zeejdœ‡ fdd„Z‡  ZS )Ú_ReaderWithOffsetÚtranslationrQ   rV   N)Úfqn_to_offsetr%   c                    s*   t ƒ  ¡  || _ti ƒ| _i | _i | _d S ©N)ÚsuperÚ__init__r`   r   rV   rQ   r_   )Úselfr`   ©Ú	__class__r)   r*   rc   �   s
    

z_ReaderWithOffset.__init__)r%   c                 C   s"  g }i | _ | j ¡ D �]\}}| jj| }t|tƒsF|t|||ƒ7 }q|| jkrb|t|||ƒ7 }q| j| }t	| 
¡ ƒdks€t‚| 
¡ d }tt t|jj|ƒ¡t |jj¡d�g}t|tt|ƒ|ƒ}|D ]D}	|	jjd k	sÞt‚t|	jj|ƒ}
tj|	jt |
¡d�}|| j |	j< qÊ||7 }qt|ƒS )NrR   r   )ÚoffsetsÚsizes)Úoffset)r_   rQ   rS   rV   Ústate_dict_metadatarU   r
   r   r`   rC   rD   rT   r   rN   ÚSizer   rW   rX   r   r   r   Z
dest_indexri   r   ÚdataclassesÚreplacer	   )rd   ÚrequestsZfqnÚobjÚmdri   Zoriginal_shardZlocal_chunksÚreqsÚriZoriginal_offsetZoriginal_indexr)   r)   r*   Úcreate_local_plan¤   sR    


 ÿÿúÿ  ÿ ÿ ÿ
z#_ReaderWithOffset.create_local_plan)Úindexr%   c                    s   t ƒ  | j ||¡¡S ra   )rb   Úlookup_tensorr_   Úget)rd   rt   re   r)   r*   ru   Ò   s    z_ReaderWithOffset.lookup_tensor)Ú__name__Ú
__module__Ú__qualname__r   r   Ú__annotations__r   r   r@   r   Úintrc   r	   rs   rN   ÚTensorru   Ú__classcell__r)   r)   re   r*   r^   ˜   s   
 .r^   )Úmodel_state_dictÚoptimizer_keyÚstorage_readerÚplannerr%   c              	   C   s6  |  ¡ }t| ƒ\}}tj |¡j}t|ƒ}|dkr~g }	tt ¡ ƒD ],}
t	||
| 
¡  ƒ}|	 d|
› d|› �¡ qBtd|	d�}nt|ƒ}i }i }|j ¡ D �]h\}}|j| }|d |krºq˜t|tƒrÎd||< q˜|j ¡ dkròt|j|j|ƒ||< q˜|dk�r.tt|j|j|ƒt ¡ t ¡ | 
¡ tƒ d�||< q˜|d	 }| |d|jf¡d }| t |¡|j¡}g }t |¡}|jD ]>}tt |j!ƒ "¡ |k�r’�qt| t#t|j|j$|ƒ|d
�¡ �qtt%j&|||d�}||k�rú|| d dk	�rútt't( || d ƒ||< |||< q˜t)j*|||dk	�rt+|ƒn|d� t,||jƒ}|S )aç  
    Loads a state_dict in conjunction with FSDP sharded optimizer state.
    This is the current recommended way to checkpoint FSDP.
    >>> # xdoctest: +SKIP
    >>> import torch.distributed.checkpoint as dist_cp
    >>> # Save
    >>> model: torch.nn.Model
    >>> optim_params = model.parameters()
    >>> optim = torch.optim.SGD(optim_params, lr=0.01)
    >>> # Save
    >>> with FSDP.state_dict_type(model, StateDictType.SHARDED_STATE_DICT):
    >>>     state_dict = {
    >>>         "optimizer": FSDP.optim_state_dict(model, optim),
    >>>         "model": model.state_dict()
    >>>     }
    >>>     dist_cp.save_state_dict(
    >>>         state_dict=optim_state,
    >>>         storage_writer=dist_cp.FileSystemWriter("checkpoint"),
    >>>         planner=dist_cp.DefaultSavePlanner(),
    >>>     )
    >>>
    >>> # Load
    >>> with FSDP.state_dict_type(model_tp, StateDictType.SHARDED_STATE_DICT):
    >>>     model_state_dict = model_tp.state_dict()
    >>>     checkpoint = {
    >>>         "model": model_state_dict
    >>>     }
    >>>     dist_cp.load_state_dict(
    >>>         state_dict=checkpoint,
    >>>         storage_reader=dist_cp.FileSystemReader(checkpoint_file),
    >>>         planner=dist_cp.DefaultLoadPlanner(),
    >>>     )
    >>>     model.load_state_dict(checkpoint["model_state"])
    >>>
    >>>     optim_state = dist_cp.load_sharded_optimizer_state_dict(
    >>>         model_state_dict,
    >>>         optimizer_key="optimizer",
    >>>         storage_reader=dist_cp.FileSystemReader("checkpoint"),
    >>>     )
    >>>
    >>>     flattened_osd = FSDP.optim_state_dict_to_load(
    >>>        model, optim, optim_state["optimizer"]
    >>>     )
    >>>
    >>>     optim.load_state_dict(flattened_osd)
    Nr.   r/   r   r7   z
<bytes_io>rR   )ÚrankZ
world_sizeZnum_devices_per_noder,   é   )rE   rV   )Zprocess_group)rQ   r€   r�   )-Zread_metadatar]   r5   r:   r;   r<   r    r=   r>   r   r'   Úappendr   rA   rj   rS   Zplanner_datarU   r   r?   ZnumelrP   Ú
propertiesr   Zget_rankr   rv   Zbuild_metadatarN   rk   Zshards_metadatar   r   Z	placementr‚   r   rX   r
   Z+_init_from_local_shards_and_global_metadatar   r{   Údist_cpZload_state_dictr^   r   )r~   r   r€   r�   rV   Zlayout_specsrZ   Zdp_pg_device_typer(   r9   ÚiZdevice_infoZsharding_specrQ   r`   r[   r\   Zkey_pathZspec_keyZ
alloc_sizeZst_mdrD   Zcurrent_rankZshard_mdÚstr)   r)   r*   r!   Ö   s–    4


û ÿ

ÿÿ  ÿüÿ	  ÿÿþ 
ÿ
ü)r"   )N)r"   )N)Frl   Útypingr   r   r   r   r   r   r   Z$torch.distributed.checkpoint.plannerr	   rN   Ztorch.distributedÚdistributedr5   Z+torch.distributed._shard.sharded_tensor.apir
   Z0torch.distributed._shard.sharded_tensor.metadatar   Z-torch.distributed._shard.sharded_tensor.shardr   Z:torch.distributed._shard.sharding_spec.chunk_sharding_specr   Ztorch.distributed.checkpointÚ
checkpointr†   Z%torch.distributed.checkpoint.metadatar   r   r   r   r   r   Z"torch.distributed.distributed_c10dr   Z#torch.distributed.fsdp._shard_utilsr   Z,torch.distributed.checkpoint.planner_helpersr   r   Ztorch.distributed.remote_devicer   Ztorch.distributed._tensorr   Z,torch.distributed.checkpoint.default_plannerr   r   Z)torch.distributed.checkpoint._nested_dictr   Z"torch.distributed.checkpoint.utilsr   r   r   Ztorch._utilsr    r@   r{   ZSTATE_DICT_2D_LAYOUTÚ__all__r+   ZProcessGrouprA   r|   ÚboolrG   rP   r]   r^   ZStorageReaderr!   r)   r)   r)   r*   Ú<module>   sV   $  ÿ
 ÿþþ$B üû