a h@@sVUddlZddlmZmZmZddlZddlmZddl m Z ddl m Z ddl mZddlmZddlmZdd lmZmZmZmZmZmZmZdd lmZmZmZmZm Z m!Z!dd l"m#Z#m$Z$d gZ%e&e'e(d <eee)dddZ*e&ee)dddZ+e&ee&ee&edddZ,ej-edddZ.e edddZ/ee edddZ0e'ee d d!d"Z1e'ee e d#d$d%Z2e'ej-e d d&d'Z3e'ed(d)d*Z4d+d,Z5d-d.Z6e'ee&ee&ed/d0d Z7eed1d2d3Z8e'ee&e d4d5d6Z9eedd7d8Z:ej-e&edd9d:Z;e'eee&ed;dd?Z>eeeed@dAdBZ?dS)CN)AnyCallablecast)_get_device_module) ShardMetadata) ShardedTensor)DTensor)%compute_local_shape_and_global_offset)BytesStorageMetadataChunkStorageMetadata MetadataIndexSTATE_DICT_TYPE STORAGE_TYPESTensorPropertiesTensorStorageMetadata) LoadItemTypeReadItemSavePlanTensorWriteData WriteItem WriteItemType)"_check_shard_metadata_pair_overlap+_shards_get_overlap_region_wrt_saved_tensor create_read_items_for_chunk_list__all__)plan other_planreturnc Cs|j|jkrdSt|jt|jkr(dSt|j|jD]\}}|j|jkrPdS|j}|j}|j|jks|j|jks|j|jkrdS|j}|j}|r|r|s|rdS|r6|r6|j |j krdS|j }|j } |r| r|s| rdS|r6| r6|j | j ks|j | j kr6dSq6dS)a Compare the two Save plans and return True if they are equal. Args: plan (SavePlan): First SavePlan to compare. other_plan (SavePlan): Second SavePlan to compare. Returns: True if the two plans are equal, False otherwise. FT) usablelenitemsziptypeindexfqnoffset tensor_datasizechunkoffsetssizes) rrZ plan_itemZother_plan_itemZplan_metadata_indexZother_plan_metadata_indexr'Zother_tensor_datar)Z other_chunkr,Z/var/www/auris/lib/python3.9/site-packages/torch/distributed/checkpoint/planner_helpers.py_compare_save_plans'sL        r.) delta_plansrcCstdd|DS)z Check if any delta plan is usable, indicating the plan has changed. Args: delta_plans (List[SavePlan]): A list of delta plans to check. Returns: True if any delta plan is usable, False otherwise. css|]}|o|jVqdSN)r).0 delta_planr,r,r- qz(_contains_usable_plan..)any)r/r,r,r-_contains_usable_planhs r6) cached_plansr/rcCs<g}t||D](\}}|r,|js,||q||q|S)ac Merge a list of delta plans into a single plan. Args: cached_plans (List[SavePlan]): A list of cached plans. delta_plans (List[SavePlan]): A list of delta plans to merge. It can contain empty plans Returns: A single merged plan. If a delta plan is not usable, use the cached plan. Otherwise, use the delta plan. )r"rappend)r7r/Z merged_plansZ cached_planr2r,r,r-_merge_delta_local_plansts    r9)tensorrcCs$ttdgt||dS)Nrr*r+)r torchSizer r()r:r,r,r-_create_chunk_from_tensorsr>)shard_mdrcCstt|jt|jdSNr;)r r<r= shard_offsetsZ shard_sizes)r?r,r,r-_chunk_for_shards  rB)sharded_tensorr?rcCs>|j}t|j|j|j|j|jd}tt |||j dS)N)dtypelayout requires_grad memory_format pin_memoryr) propertiesr() metadataZtensor_propertiesrrDrErFrGrHrrBr()rCr?Zshard_propertiesrJr,r,r-_sharded_tensor_metadatas rL)r%r:rc Csbt|j|j|j\}}t|t|}}tt||tj t t ||dt ||ddS)Nr;rIr$r#r')r shape device_mesh placementsr<r=rr rSHARDrr rcreate_from_tensorto_localr()r%r:r+r*r,r,r-_create_write_items_for_dtensors   rT)r%rCr?rcCs(t|j}tt||tjt||dS)NrM)r<r=rArr rrQrL)r%rCr?r*r,r,r-_create_write_item_for_shards  rUcCsNtdgt|}tt||tjtt ||dt ||ddS)Nrr;rIrM) r<r=r r(rr rTENSORrr rrR)r%r:r*r,r,r-_create_write_item_for_tensorsrWr%bytescCstt|tjdS)N)r$r#)rr rBYTE_IOrXr,r,r-_create_write_item_for_bytesiosr[c Cs.ttj|t|f|t|ft|fdSN)r# dest_index dest_offsets storage_indexstorage_offsetslengths)rrrZr<r=r]Z dest_offsetr_Zstorage_offsetlengthr,r,r-_create_read_item_for_byteios   rdc Cs(ttj|t||t|t|dSr\)rrrVr<r=r]r^r_r`rar,r,r-_create_read_item_for_tensorsrf)r% checkpoint_md local_chunksrc Csg}t|D]\}}t|jD]\}}t||s2qg}g} g} t||dD]*\} } } }|| | | | |qJ|tt||j|| t||j||| dqq |S)aW Create a list of ``ReadItem`` based on the checkpoint and local chunks. This applies the resharding algorithm and computes the reads needed to satisfy ``local_chunks`` with a checkpoint described by ``checkpoint_md``. Args: fqn (str) : The state_dict FQN to pass to ``ReadItem``. checkpoint_md (TensorStorageMetadata): metadata for a given tensor from a checkpoint. local_chunks (List[ChunkStorageMetadata]): Local chunks that needs to be loaded. Returns: A list of ``ReadItem`` that will satisfy all input chunks. )Z saved_shardZ current_shardre) enumeratechunksrrr8rfr r*)r%rgrhZ read_itemsidxshardZ storage_idxZ storage_mdr`r^raZ_dimZoffset_for_saved_tensorZoffset_for_current_tensorrcr,r,r-rs:       ) state_dictrcsg}|D]~\ttr0|tq ttr\|fddjDq tt j rz|t q |t q t |S)Nc3s|]}t|VqdSr0)rU)r1r?r%objr,r-r38sz5_create_default_metadata_only_plan..)r! isinstancerr8rTrextendrKZshards_metadatar<TensorrWr[r)rmrequestsr,rnr-"_create_default_metadata_only_plan2s    rt)r%objectrcs`tdrSttr8fddDSttjrPtgStgSdS)N__create_write_items__csg|]}t|jqSr,)rUrKr1rlr%rur,r- Hsz'_create_write_items..) hasattrrvrpr local_shardsr<rrrWr[rxr,rxr-_create_write_itemsCs      r|cCs8t|j|j|j\}}t|t|}}t||dSr@)r rNrOrPr<r=r )r:r+r*r,r,r-_create_chunk_from_dtensorRs r}cCs`t|dr|}nHt|tr2dd|D}n*t|tjrJt|g}ntdt ||S)N__create_chunk_list__cSsg|]}t|jqSr,)rBrKrwr,r,r-rybsz&_create_chunk_list..zMUnsupported Type, expecting one of [Tensor, DTensor, ShardedTensor] ,but got ) rzr~rprr{r<rrr> ValueErrorr#)r:rhr,r,r-_create_chunk_list]s     r)r%mdrorc Cst|tshz t|}WnDtyZ}z,td|ddt||WYd}~n d}~00t|||Stt|dt|dddgSdS)Nz Invalid checkpoint metadata for z, z(expected BytesStorageMetadata but found rrb)rpr rrr#rrdr )r%rrorhexr,r,r-_create_read_itemsps(     rcCs>tddd}tddd}tjddd}t||||dS) zP Initializes meta tensor if the meta tensor is DTensor or torch.Tensor. )valuecSsxt|dd}|tdkrptjj}ttjt| }tj | |d}t j ||j|j||d}|S|SdS)Ndevicemetar)rOrPrNstride)getattrr<rdistdistributed_c10d_get_pg_default_devicer#rrcurrent_device empty_likerSrZ from_localrOrPr(r)rr device_typeZnew_local_tensorZdtensorr,r,r- dtensor_funcs   z&_init_state_dict..dtensor_funccSs8t|dd}|tdkr0tdt|dn|SdS)NrrzFound unsupported type z for meta device loading.)rr<r RuntimeErrorr#)rrr,r,r-sharded_tensor_funcs  z-_init_state_dict..sharded_tensor_funccSsTt|dd}|tdkrLtjj}ttjt| }tj ||d}|S|SdS)Nrrr) rr<rrrrr#rrrr)rrrr:r,r,r- tensor_funcs  z%_init_state_dict..tensor_funcN)rrr<rr_iterate_state_dict)rmrrrr,r,r-_init_state_dicts  r) iter_objectrrrcst|tr|St|tr$|St|tjr8|St|ttttt j fsV|durZ|St|t r| D]\}}t |||<ql|St|ttfrƇfdd|D}t|trt|}|SdS)a$ Iterate through the state dict, applying the given functions to each tensor type and update the state dict in place. Args: iter_object (Any): the target state_dict. sharded_tensor_func (Callable): the function to apply to ShardedTensor dtensor_func (Callable): the function to apply to DTensor tensor_func (Callable): the function to apply to Tensor # TODO: let state_dict_util._iterate_state_dict() to support in place option so we don't need to have two versions of _iterate_state_dict. Ncsg|]}t|qSr,)r)r1vrrrr,r-rysz'_iterate_state_dict..)rprrr<rrintfloatstrrYioBytesIOdictr!rlisttuple)rrrrkeyrretr,rr-rs0      r)@rtypingrrrr<Ztorch.distributedZ distributedrZ torch._utilsrZ!torch.distributed._shard.metadatarZ'torch.distributed._shard.sharded_tensorrZtorch.distributed.tensorrZtorch.distributed.tensor._utilsr rKr r r rrrrZplannerrrrrrrZ reshardingrrrrr__annotations__boolr.r6r9rrr>rBrLrTrUrWr[rdrfrrtr|r}rrrrrr,r,r,r-s^       $ A       7 8