o ,h@ @sUddlZddlmZmZmZddlZddlmZddl m Z ddl m Z ddl mZddlmZddlmZdd lmZmZmZmZmZmZmZdd lmZmZmZmZm Z m!Z!dd l"m#Z#m$Z$d gZ%e&e'e(d <dedede)fddZ*de&ede)fddZ+de&ede&ede&efddZ,dej-defddZ.de defddZ/dede defd d!Z0d"e'dede fd#d$Z1d"e'dede de fd%d&Z2d"e'dej-de fd'd(Z3d"e'd)efd*d+Z4d,d-Z5d.d/Z6d"e'd0ed1e&ede&efd2d Z7d3edefd4d5Z8d"e'd6ede&e fd7d8Z9dedefd9d:Z:dej-de&efd;d<Z;d"e'd=ed>ede&efd?d@ZdCedDedEedFefdGdHZ?dS)IN)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|jkrdSt|j|jD]j\}}|j|jkr(dS|j}|j}|j|jks@|j|jks@|j|jkrCdS|j}|j}|rM|rQ|sT|rTdS|r|r|j |j kradS|j }|j } |rk| ro|sr| rrdS|r| r|j | j ks|j | j krdSqdS)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) rr plan_itemother_plan_itemplan_metadata_indexother_plan_metadata_indexr'other_tensor_datar) other_chunkr2g/var/www/html/scripts/venv/lib/python3.10/site-packages/torch/distributed/checkpoint/planner_helpers.py_compare_save_plans'sD        r4 delta_planscCstdd|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_planr2r2r3 qsz(_contains_usable_plan..)any)r5r2r2r3_contains_usable_planhs r; cached_planscCs<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)r<r5 merged_plans cached_planr8r2r2r3_merge_delta_local_plansts    r@tensorcCs$ttdgt||dS)Nrr*r+)r torchSizer r()rAr2r2r3_create_chunk_from_tensorsrEshard_mdcCstt|jt|jdSNrB)r rCrD shard_offsets shard_sizes)rFr2r2r3_chunk_for_shards  rJsharded_tensorcCs>|j}t|j|j|j|j|jd}tt |||j dS)N)dtypelayout requires_grad memory_format pin_memoryr) propertiesr() metadatatensor_propertiesrrLrMrNrOrPrrJr()rKrFshard_propertiesrRr2r2r3_sharded_tensor_metadatas rVr%c Csbt|j|j|j\}}t|t|}}tt||tj t t ||dt ||ddS)NrBrQr$r#r')r shape device_mesh placementsrCrDrr rSHARDrr rcreate_from_tensorto_localr()r%rAr+r*r2r2r3_create_write_items_for_dtensors   r^cCs(t|j}tt||tjt||dS)NrW)rCrDrHrr rr[rV)r%rKrFr*r2r2r3_create_write_item_for_shards r_cCsNtdgt|}tt||tjtt ||dt ||ddS)NrrBrQrW) rCrDr r(rr rTENSORrr rr\)r%rAr*r2r2r3_create_write_item_for_tensorsrabytescCstt|tjdS)N)r$r#)rr rBYTE_IO)r%rbr2r2r3_create_write_item_for_bytesiosrdc Cs.ttj|t|f|t|ft|fdSN)r# dest_index dest_offsets storage_indexstorage_offsetslengths)rrrcrCrDrf dest_offsetrhstorage_offsetlengthr2r2r3_create_read_item_for_byteios   roc Cs(ttj|t||t|t|dSre)rrr`rCrDrfrgrhrirjr2r2r3_create_read_item_for_tensorsrq checkpoint_md local_chunksc Csg}t|D]L\}}t|jD]B\}}t||sqg}g} g} t||dD]\} } } }|| | | | |q%|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. ) saved_shard current_shardrp) enumeratechunksrrr=rqr r*)r%rrrs read_itemsidxshard storage_idx storage_mdrirgrj_dimoffset_for_saved_tensoroffset_for_current_tensorrnr2r2r3rs<       state_dictcsg}|D]?\ttr|tqttr.|fddjDqtt j r=|t q|t qt |S)Nc3s|] }t|VqdSr6)r_)r7rFr%objr2r3r98s  z5_create_default_metadata_only_plan..)r! isinstancerr=r^rextendrSshards_metadatarCTensorrardr)rrequestsr2rr3"_create_default_metadata_only_plan2s   robjectcs\tdr SttrfddDSttjr(tgStgS)N__create_write_items__csg|] }t|jqSr2)r_rSr7rzr%rr2r3 H z'_create_write_items..) hasattrrrr local_shardsrCrrardrr2rr3_create_write_itemsCs       rcCs8t|j|j|j\}}t|t|}}t||dSrG)r rXrYrZrCrDr )rAr+r*r2r2r3_create_chunk_from_dtensorRs rcCsbt|dr |}|St|trdd|D}|St|tjr(t|g}|Stdt |)N__create_chunk_list__cSsg|]}t|jqSr2)rJrSrr2r2r3rbs z&_create_chunk_list..zMUnsupported Type, expecting one of [Tensor, DTensor, ShardedTensor] ,but got ) rrrrrrCrrE ValueErrorr#)rArsr2r2r3_create_chunk_list]s   rmdrc Csxt|ts.zt|}Wnty'}ztd|ddt||d}~wwt|||Stt|dt|dddgS)Nz Invalid checkpoint metadata for z, z(expected BytesStorageMetadata but found rrk)rr rrr#rror )r%rrrsexr2r2r3_create_read_itemsps,     rcCs>dtfdd}dtfdd}dtjfdd}t||||dS) zP Initializes meta tensor if the meta tensor is DTensor or torch.Tensor. valuecSstt|dd}|tdkr8tjj}ttjt| }tj | |d}t j ||j|j||d}|S|S)Ndevicemetar)rYrZrXstride)getattrrCrdistdistributed_c10d_get_pg_default_devicer#rrcurrent_device empty_liker]r from_localrYrZr(r)rr device_typenew_local_tensordtensorr2r2r3 dtensor_funcs  z&_init_state_dict..dtensor_funccSs2t|dd}|tdkrtdt|d|S)NrrzFound unsupported type z for meta device loading.)rrCr RuntimeErrorr#)rrr2r2r3sharded_tensor_funcs z-_init_state_dict..sharded_tensor_funccSsPt|dd}|tdkr&tjj}ttjt| }tj ||d}|S|S)Nrrr) rrCrrrrr#rrrr)rrrrAr2r2r3 tensor_funcs  z%_init_state_dict..tensor_funcN)rrrCr_iterate_state_dict)rrrrr2r2r3_init_state_dicts  r iter_objectrrrcst|tr |St|tr|St|tjr|St|ttttt j fs+|dur-|St|t rF| D] \}}t |||<q6|St|ttfrcfdd|D}t|trat|}|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|qSr2)r)r7vrrrr2r3rrz'_iterate_state_dict..)rrrrCrintfloatstrrbioBytesIOdictr!rlisttuple)rrrrkeyrretr2rr3rs0      r)@rtypingrrrrCtorch.distributed distributedr torch._utilsr!torch.distributed._shard.metadatar'torch.distributed._shard.sharded_tensorrtorch.distributed.tensorrtorch.distributed.tensor._utilsr rSr r r rrrrplannerrrrrrr reshardingrrrrr__annotations__boolr4r;r@rrErJrVr^r_rardrorqrrrrrrrrrr2r2r2r3s       $ A         7 7