o
    j9:jn                  	   @   sh  U d dl Z d dlZd dlZd dlZd dlZd dlmZmZ d dlm	Z	 d dl
mZmZ d dl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mZmZmZmZmZm Z m!Z! d dl"m#Z#m$Z$m%Z%m&Z&m'Z'm(Z(m)Z) d dl*m+Z+m,Z,m-Z-m.Z.m/Z/m0Z0m1Z1 d dl2m3Z3 d dl4m5Z5 ddl6m7Z7 e8e9Z:ej;e<d< g dZ=G dd de'Z>G dd de$Z?G dd de?Z@	d6deAeBef dedeCde#fddZDd eEe# deEe# fd!d"ZFdeAeBef d#eCde&fd$d%ZG	d6d eEe& d&eCdeHeEe& ef fd'd(ZIdedefd)d*ZJd+ed,edeCfd-d.ZKd/ejLd0edeCfd1d2ZMd3eEe& dedeEeB fd4d5ZNdS )7    N)bisect_rightinsort)ChainMap)Anycast)narrow_tensor_by_index)dedup_save_plans)FLATTEN_MAPPINGflatten_state_dict)_flatten_sharded_tensors)set_element)BytesStorageMetadataChunkStorageMetadataMetadataMetadataIndexSTATE_DICT_TYPESTORAGE_TYPESStorageMetaTensorStorageMetadata)LoadPlanLoadPlannerReadItemSavePlanSavePlanner	WriteItemWriteItemType)_compare_save_plans_contains_usable_plan"_create_default_metadata_only_plan_create_read_items_create_write_items_init_state_dict_merge_delta_local_plans)find_state_dict_object)DTensor   )_versionlogger)DefaultSavePlannerDefaultLoadPlannercreate_default_local_load_plancreate_default_global_load_plancreate_default_local_save_plancreate_default_global_save_planc                   @   sb  e Zd ZU eed< 					d+dedededB ded	ed
dfddZ		d,dededB ded
dfddZ	d
e
fddZdee
 d
ee
 fddZdee
 d
eee
 ef fddZdee
 d
eee
 ee
 ef fddZdee
 d
eee
 ef fddZde
d
e
fddZde
d
e
fd d!Zd"ed
ejejB fd#d$Zd%ed
efd&d'Zd"ed(efd)d*ZdS )-r(   mappingsTNFr
   flatten_sharded_tensorsdedup_replicated_tensorsdedup_save_to_lowest_rankenable_plan_cachingreturnc                 C   s>   || _ || _i | _|| _|d urtd | jj| _|| _	d S )NzDefaultSavePlanner's `dedup_replicated_tensors` argument is being deprecated, and no longer has any effect. Please remove this argument from your call.)
r
   r/   r.   r1   r'   warning	__class____name___cached_plans_key_enable_plan_caching)selfr
   r/   r0   r1   r2    r:   s/home/nk/hobo-godmode/plappi-mvp/.venv/lib/python3.10/site-packages/torch/distributed/checkpoint/default_planner.py__init__I   s   

zDefaultSavePlanner.__init__
state_dictstorage_metais_coordinatorc                 C   s2   | j r
t |\}| _| jrt|}|| _|| _d S N)r
   r.   r/   r   r=   r?   )r9   r=   r>   r?   r:   r:   r;   set_up_planner^   s   
z!DefaultSavePlanner.set_up_plannerc                 C   sp   t | j| j}| jrtj|| jd}|| _| jr5| j	t
jv r2t|t
j| j	 r2td tg ddS || _| jS )Nplanner_datazINo change in the local plan. Skipping sending the plan to the coordinatorFusable)r,   r=   r?   r
   dataclassesreplacer.   planr8   r7   r   _cached_save_planr   r'   infor   _pending_local_plan)r9   rH   r:   r:   r;   create_local_plank   s    z$DefaultSavePlanner.create_local_plan	all_plansc                 C   s   t || jS r@   )r   r1   )r9   rM   r:   r:   r;   _dedup_save_plans   s   z$DefaultSavePlanner._dedup_save_plansc           	      C   s   |  |}t|\}}| jr"dd |D }tt| }tj||d}t||}|rCd|}t	|dkr<|d d d }t
d| ||fS )Nc                 S   s   g | ]}|j qS r:   rB   ).0pr:   r:   r;   
<listcomp>   s    z:DefaultSavePlanner._create_global_plan.<locals>.<listcomp>rB   z; i  z... (truncated)z Failed to validate global plan: )rN   r-   r
   dictr   rF   rG   _validate_global_planjoinlen
ValueError)	r9   rM   deduped_plansglobal_planmetadataplanner_data_dictmerged_mappingsvalidation_errorserror_summaryr:   r:   r;   _create_global_plan   s   


z&DefaultSavePlanner._create_global_planc                 C   s"  g }| j tjvr&| |\}}|tj| j < |tj| j < |tj| j < |||fS t|sBtg ddgt| }tj| j  }tj| j  }nJt	tj| j  |}|tj| j < | |\}}| j | jv rt
tj| j  |D ]\}}t||rz|tg dd qg|| qg|tj| j < |tj| j < |||fS )zw
        Create global plan with caching.
        Returns a tuple of global_plan_delta, global_plan, metadata.
        FrD   )r7   r   _cached_all_plansr^   _cached_global_plan_cached_metadatar   r   rU   r"   zipr   append)r9   rM   global_plan_deltarX   rY   merged_planscached_plannew_planr:   r:   r;    _create_global_plan_with_caching   s4   


z3DefaultSavePlanner._create_global_plan_with_cachingc                 C   sD   g }| j r| |\}}}n	| |\}}|}|| _|| _|| jfS r@   )r8   rh   r^   rX   rY   )r9   rM   rd   rX   rY   r:   r:   r;   create_global_plan   s   
z%DefaultSavePlanner.create_global_planrg   c                 C   s.   |}|j stj| j }|S |}|tj| j< |S r@   )rE   r   _cached_final_save_planr7   r9   rg   finished_planr:   r:   r;   _finish_plan_with_caching   s   z,DefaultSavePlanner._finish_plan_with_cachingc                 C   s<   |}| j r| |}t| dr| jtj| j< | `|| _| jS )NrK   )r8   rm   hasattrrK   r   rI   r7   rH   rk   r:   r:   r;   finish_plan   s   


zDefaultSavePlanner.finish_plan
write_itemc                 C      |  |j}| ||S r@   )lookup_objectindextransform_object)r9   rp   objectr:   r:   r;   resolve_data     zDefaultSavePlanner.resolve_datars   c                 C      t | j|S zSExtension from the planner interface to make it easy to extend the default planner.r#   r=   r9   rs   r:   r:   r;   rr        z DefaultSavePlanner.lookup_objectru   c                 C   s(   |j tjkrt }t|| |}|S ry   )typer   BYTE_IOioBytesIOtorchsave)r9   rp   ru   bytesr:   r:   r;   rt     s
   z#DefaultSavePlanner.transform_object)TTNFFNF)r6   
__module____qualname__r	   __annotations__boolr<   r   r   rA   r   rL   listrN   tupler   r^   rh   ri   rm   ro   r   r   Tensorr   r   rv   r   r   rr   rt   r:   r:   r:   r;   r(   F   sj   
 



<

r(   c                	   @   s  e Zd ZU dZeed< eed< 			d(dededed	d
fddZ	
	d)dede	d
B ded	d
fddZ
d	efddZdee d	ee fddZded	efddZdedejd	d
fddZdefddZded ejd	d
fd!d"Zd#ed	ejfd$d%Zded ejfd&d'Zd
S )*r)   ak  
    DefaultLoadPlanner that adds multiple features on top of LoadPlanner.

    In particular it adds the following:

    flatten_state_dict: Handle state_dict with nested dicts
    flatten_sharded_tensors: For FSDP in 2D parallel mode
    allow_partial_load: If False, will raise a runtime error if a key is present in state_dict, but not in the checkpoint.
    original_state_dictr.   TFr
   r/   allow_partial_loadr3   Nc                 C   s"   || _ || _i | _i | _|| _d S r@   )r
   r/   r   r.   r   )r9   r
   r/   r   r:   r:   r;   r<   1  s
   
zDefaultLoadPlanner.__init__r=   rY   r?   c                 C   sF   t | || _| jrt|}| jrt|\}| _|| _|| _|| _d S r@   )	r!   r   r/   r   r
   r.   r=   rY   r?   )r9   r=   rY   r?   r:   r:   r;   rA   =  s   
z!DefaultLoadPlanner.set_up_plannerc                 C   s   | j d u r	td| jr?t| j }t| j j }|| }|r?dt_t| j	\}}t| }||@ r<||| _| _
d t_t| j| j | j S )Nzself.metadata is not None2_3)rY   AssertionErrorr
   setr=   keysstate_dict_metadatar&   _derived_versionr   r.   r*   r   )r9   current_keys	load_keysmissing_keysold_state_dictold_mappingsold_keysr:   r:   r;   rL   P  s$   
z$DefaultLoadPlanner.create_local_planrX   c                 C   s   t |S r@   )r+   )r9   rX   r:   r:   r;   ri   x  s   z%DefaultLoadPlanner.create_global_planrg   c                 C   s   |S r@   r:   )r9   rg   r:   r:   r;   ro   {     zDefaultLoadPlanner.finish_plan	read_itemvaluec                 C   sH   | j rt| j| j|jj tj|dd d S tj|dd| j|jj< d S )NF)weights_only)	r
   r   r   r.   
dest_indexfqnr   loadr=   )r9   r   r   r:   r:   r;   
load_bytes~  s   zDefaultLoadPlanner.load_bytesc                 C   rq   r@   )lookup_tensorr   transform_tensorr9   r   tensorr:   r:   r;   resolve_tensor  rw   z!DefaultLoadPlanner.resolve_tensorr   c                 C   s   d S r@   r:   r   r:   r:   r;   commit_tensor  r   z DefaultLoadPlanner.commit_tensorrs   c                 C   rx   ry   rz   r{   r:   r:   r;   r     r|   z DefaultLoadPlanner.lookup_tensorc                 C   s   t ||j|jS ry   )r   dest_offsetslengthsr   r:   r:   r;   r     s   z#DefaultLoadPlanner.transform_tensor)TTFr   )r6   r   r   __doc__r   r   r	   r   r<   r   rA   r   rL   r   ri   ro   r   r   r   r   r   r   r   r   r   r   r   r:   r:   r:   r;   r)   #  sF   
 


(r)   c                	       s`   e Zd ZdZd fdd	Zdededefdd	Z		
dde	dedB deddf fddZ
  ZS )_EmptyStateDictLoadPlannera  
    Extension of DefaultLoadPlanner, which rebuilds state_dict from the saved metadata.
    Useful for loading in state_dict without first initializing a model, such as
    when converting a DCP checkpoint into a Torch save file.

    . N.B. `state_dict` must be an empty dictionary when used with this LoadPlanner

    .. warning::
        Because the entire state dict is initialized, It's recommended to only utilize
        this LoadPlanner on a single rank or process to avoid OOM.

    Nc                    s   || _ t j|i | d S r@   )r   superr<   )r9   r   argskwargsr5   r:   r;   r<     s   z#_EmptyStateDictLoadPlanner.__init__keyrY   r3   c                    s    j d u rdS | j v rdS g }|j|}|D ]}|r+|d|d t|g q|| qt fdd|D r>dS dS )NT.c                 3   s    | ]}| j v V  qd S r@   )r   )rO   unflattened_keyr9   r:   r;   	<genexpr>  s    zA_EmptyStateDictLoadPlanner._should_include_key.<locals>.<genexpr>F)r   rC   getrc   rT   strany)r9   r   rY   unflattened_keysrC   r   r:   r   r;   _should_include_key  s   

z._EmptyStateDictLoadPlanner._should_include_keyFr=   r?   c                    s   |rt d|d u rt d|j D ]2\}}| ||sqt|tr-tj|j|j	j
d}|jd urA||jv rAt||j| | q|||< qt ||| d S )Nznot state_dictzmetadata is not None)dtype)r   r   itemsr   
isinstancer   r   emptysize
propertiesr   rC   r   r   rA   )r9   r=   rY   r?   kvr   r:   r;   rA     s   

z)_EmptyStateDictLoadPlanner.set_up_plannerr@   r   )r6   r   r   r   r<   r   r   r   r   r   rA   __classcell__r:   r:   r   r;   r     s    r   Tr=   rY   strictr3   c                 C   s   g }	 |   D ]Z\}}||jvr|rtd| dq|j| }t|trDt|dd d urD|j| krDtd|j d|  d| t|trY|j	
 d urX|t|||7 }q|t|||7 }qt|S )Nz&Missing key in checkpoint state_dict: r   r   zSize mismatch between saved z and current: z for )r   r   RuntimeErrorr   r   getattrr   rV   r$   device_meshget_coordinater   r   )r=   rY   r   requestsr   objmdr:   r:   r;   r*     s*   



r*   rM   c                 C   s   | S )z
    Create global load plan used by DefaultLoadPlanner.

    The default load behavior involved no global coordination and this function
    currently doesn't change the local plans.
    r:   )rM   r:   r:   r;   r+     s   	r+   r?   c                 C   sT   g }|   D ]\}}t|tr|j dur|t||7 }q|t||7 }qt|S )a  
    Create the ``SavePlan`` used by DefaultSavePlanner.

    On non-coordinator ranks, this function ignores tensors and non-tensor objects,
    only producing writes for ShardedTensor objects.

    On the coordinator rank, produce writes for all values.
    N)r   r   r$   r   r   r    r   )r=   r?   r   r   r   r:   r:   r;   r,     s   
r,   rewrite_index_hintsc           
      C   s(  i }g }| D ]}g }|j D ]u}|jtjkr|jj|v rtd|jtjkr2t ||jj< |	| q|j
du r;tdtt||jjt|j
j|j
jg d}|}|retj|jt|jd}	tj||	d}|	| |j
jdu rztd|jj d|j	|j
j q|	tj||d q|t|fS )	a6  
    Create the global plan and metadata used by DefaultSavePlanner.

    Metadata is produced by concatenating the metadata of all ``WriteItem`` from the supplied plans.

    The only global planning change is to update index hints in all ``MetadataIndex`` objects if
    ``rewrite_index_hints`` is True.
    zitem.index.fqn not in mdNzitem.tensor_data is not None)r   r   chunks)rs   zZ
                    Cannot create MD for tensor without bounds.
                    FQN: z
                )r   )r   r}   r   SHARDrs   r   r   r~   r   rc   tensor_datar   r   
setdefaultr   r   rF   rG   rU   r   chunkr   )
rM   r   r   	new_plansrH   	new_itemsitem	tensor_mdnew_item	new_indexr:   r:   r;   r-   .  sL   



r-   c                 C   s   t | }t|g\}}|S )zTReturn the ``Metadata`` if DefaultSavePlanner was used to checkpoint ``state_dict``.)r   r-   )r=   rH   _r   r:   r:   r;   _create_default_local_metadataf  s   r   box0box1c                 C   sd   t | j}t|D ]&}| j| |j| |j|  kr dS |j| | j| | j|  kr/ dS q	dS )z9Check if two boxes overlap. Tuples are (offset, lengths).FT)rU   offsetsrangesizes)r   r   ndimsir:   r:   r;   _check_box_overlapm  s   
r   outer_box_size	inner_boxc                 C   s`   t t| D ]'}|j| dk r dS |j| dk r dS |j| |j|  | | kr- dS qdS )Nr   FT)r   rU   r   r   )r   r   r   r:   r:   r;   _check_box_bounds}  s   r   rX   c              	      s  g }|j  D ]\}ttrqtjdkrqj d} D ]&}tj|s>d| dj d| }t	| |
| |t|j7 }q t dkrtjttdfdddttt  fd	dd
}g }|D ]M}	 |	 }
|
j }||
j  }t||tjf}|r|d|= |D ]"\}} | }t|
|rd| d|
 d| }t	| |
| qt|||	f qstj}t| dkr||krd| d| d| }t	| |
| q|S )zNValidate the global plan and return a list of error messages (empty if valid).r   zkey:z& has out of bounds chunk: tensor-size:z chunk: r%   c                    s
    j |  S r@   )r   )d)r   r:   r;   <lambda>  s   
 z'_validate_global_plan.<locals>.<lambda>)defaultr   c                    s*     j  g fddtD R S )Nc                 3   s    | ]
}  j | V  qd S r@   )r   )rO   r   )r   idxr:   r;   r     s    z:_validate_global_plan.<locals>.<lambda>.<locals>.<genexpr>)r   r   r   )r   dims	sweep_dimr   r;   r     s   )r   Nz has overlapping chunks:  z invalid fill tensor-volume: z chunks-volume: )r   r   r   r   rU   r   r   r   r'   r4   rc   mathprodr   maxr   sortedr   r   sysmaxsizer   r   )rX   rY   errorsr   chunks_volumer   msgsorted_indicesactiver   currentstartendcutoffr   	other_idxothertensor_volumer:   )r   r   r   r   r;   rS     sl   











rS   )T)OrF   r   loggingr   r   bisectr   r   collectionsr   typingr   r   r   torch.distributed._shard._utilsr   .torch.distributed.checkpoint._dedup_save_plansr   )torch.distributed.checkpoint._nested_dictr	   r
   2torch.distributed.checkpoint._sharded_tensor_utilsr   &torch.distributed.checkpoint._traverser   %torch.distributed.checkpoint.metadatar   r   r   r   r   r   r   r   $torch.distributed.checkpoint.plannerr   r   r   r   r   r   r   ,torch.distributed.checkpoint.planner_helpersr   r   r   r   r    r!   r"   "torch.distributed.checkpoint.utilsr#   torch.distributed.tensorr$    r&   	getLoggerr6   r'   Loggerr   __all__r(   r)   r   rR   r   r   r*   r   r+   r,   r   r-   r   r   Sizer   rS   r:   r:   r:   r;   <module>   s   
($
$		 ^wD

*



8
"