g^(h? UddlZddlmZmZmZddlZddlmZddl 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)fdZ*de&ede&ede&efdZ+dejXdefdZ-de defdZ.dede defdZ/de'dede fdZ0de'dede de fdZ1de'dejXde fdZ2de'defd Z3d!Z4d"Z5de'd#ed$e&ede&efd%Z6d&edefd'Z7de'd(ede&e fd)Z8dedefd*Z9dejXde&efd+Z:de'd,ed-ede&efd.Z;d&ey)5N)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"|j|jk7ryt|jt|jk7ryt|j|jD]%\}}|j|jk7ry|j }|j }|j |j k7s2|j|jk7s|j |j k7ry|j}|j}|r|r|s|ry|s|s|j|jk7ry|j}|j} |r| r|s| ry|s| s|j| jk7s|j| jk7s&yy)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_chunks j/var/www/skyplay_api_hub/venv/lib/python3.12/site-packages/torch/distributed/checkpoint/planner_helpers.py_compare_save_plansr5'sn {{j''' 4::#j..//'*$**j6F6F&G)!" ? >>_11 1'oo$3$9$9! # #'@'D'D D"))-F-M-MM"((,E,K,KK ++ +77  1 1 ,#4#9#99 %%E+11Kk5[MM[%8%88{{k&7&77 S)!V  cached_plans delta_planscg}t||D]6\}}|r|js|j|&|j|8|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$r!append)r7r8 merged_plans cached_plan delta_plans r4_merge_delta_local_plansr>hsSL#&|[#A, Z j//    ,    + , r6tensorc ttjdgt|j z|j S)Nrr,r-)r torchSizer"r*)r?s r4_create_chunk_from_tensorrDs5  A3V[[]!334FKKM r6shard_mdcttj|jtj|jSNrA)r rBrC shard_offsets shard_sizes)rEs r4_chunk_for_shardrJs3  8112jj--. r6sharded_tensorc|jj}t|j|j|j |j |j}tt|||jjS)N)dtypelayout requires_grad memory_format pin_memoryr+ propertiesr*) metadatatensor_propertiesrrMrNrOrPrQrrJr*)rKrEshard_propertiesrSs r4_sharded_tensor_metadatarWs|&..0BB!$$&&&44&44#.. J x(  $ $ & + + r6r'c t|j|j|j\}}t j |t j |}}t t||tjtt||tj|j|jS)NrArRr&r%r))r shape device_mesh placementsrBrCrrrSHARDrr rcreate_from_tensorto_localr*)r'r?r-r,s r4_create_write_items_for_dtensorr`s: f((&*;*;NE7ZZ& 7(;7E C)  #&(::6??;LM    r6ctj|j}tt ||t j t||S)NrY)rBrCrHrrrr]rW)r'rKrEr,s r4_create_write_item_for_shardrbsBjj//0G C)  ,^XF r6c :tjdgt|jz}t t ||t jtt||jtj||jS)NrrArRrY) rBrCr"r*rrrTENSORrr rr^)r'r?r,s r4_create_write_item_for_tensorresqjj!s6;;=112G C)  ! !#&wfkkmL'::6B  r6bytescJtt|tjS)N)r&r%)rrrBYTE_IO)r'rfs r4_create_write_item_for_bytesioris C  " " r6c ttj|tj|f|tj|ftj|fSN)r% dest_index dest_offsets storage_indexstorage_offsetslengths)rrrhrBrCrl dest_offsetrnstorage_offsetlengths r4_create_read_item_for_byteiorusK   ! !ZZ/# N#45 F9%  r6c ttj|tj||tj|tj|Srk)rrrdrBrCrlrmrnrorps r4_create_read_item_for_tensorrxsD   ZZ -# ?3 7#  r6 checkpoint_md local_chunksc g}t|D]\}}t|jD]\}}t||sg}g} g} t||D]:\} } } }|j | | j | | j |<|j t t ||j|| t ||j||| |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_shardrw) enumeratechunksrrr:rxrr,)r'ryrz read_itemsidxshard storage_idx storage_mdrormrp_dimoffset_for_saved_tensoroffset_for_current_tensorrts r4rrs*J - U'01E1E'F  #K5eZH OLG =&e '') &&'>?##$=>v& '   ,,S%--E!-"/Z5G5G"U$3#  ' : r6 state_dictcg}|jD]\ttr|jt 2tt r4|j fdjjDvttjr|jt|jtt|S)Nc38K|]}t|yw)N)rb).0rEr'objs r4 z5_create_default_metadata_only_plan..,s"-S#x@s)r# isinstancer r:r`rextendrTshards_metadatarBTensorrerir)rrequestsr'rs @@r4"_create_default_metadata_only_planr&sH$$& FS c7 # OO;CE F ] + OO #  > > U\\ * OO9#sC D OO:3D E F H r6objectc6t|dr|j||St|tr3|j Dcgc]}t |||j c}St|tjr t||gSt||gScc}w)N__create_write_items__) hasattrrrr local_shardsrbrTrBrreri)r'rrs r4_create_write_itemsr7sv/0,,S&99 FM * ,,.  )fenn E   FELL )-c6:;;.sF;<< sBct|j|j|j\}}t j |t j |}}t ||SrG)r rZr[r\rBrCr )r?r-r,s r4_create_chunk_from_dtensorrFsW: f((&*;*;NE7ZZ& 7(;7E  r6cJt|dr|j}|St|tr2|j Dcgc]}t |j }}|St|tjrt|g}|Stdt|cc}w)N__create_chunk_list__zMUnsupported Type, expecting one of [Tensor, DTensor, ShardedTensor] ,but got ) rrrrrrJrTrBrrD ValueErrorr%)r?rzrs r4_create_chunk_listrQsv./335   FM *:@:M:M:O 16 U^^ ,    FELL )1&9:    V ~ '   sB mdrct|ts t|}t |||St t|dt|ddgS#t$r$}td|ddt |z|d}~wwxYw)Nz Invalid checkpoint metadata for z, z(expected BytesStorageMetadata but found rrq)rr rrr%rrur)r'rrrzexs r4_create_read_itemsrds b. / -c2L0RFF )(-+C0      23%r:.dtensor_funcs$/ U\\&) )//FFHMMK 0=LLNF %//0@P (( !-- ++jjl||~ GNLr6ct|dd}|tjdk(rtdt |d|S)NrrzFound unsupported type z for meta device loading.)rrBr RuntimeErrorr%)rrs r4sharded_tensor_funcz-_init_state_dict..sharded_tensor_funcsF$/ U\\&) ))$u+6OP Lr6c4t|dd}|tjdk(rrtjj j }ttjt|j}tj||}|S|S)Nrrr) rrBrrrrr%rrrr)rrrr?s r4 tensor_funcz%_init_state_dict..tensor_funcs{$/ U\\&) )//FFHMMK 0=LLNF%%eF;FMLr6N)r rrBr_iterate_state_dict)rrrrs r4_init_state_dictr{s@ G*3 5<<  r6 iter_objectrrrc *t|tr||St|tr||St|tjr||St|t t tttjfs||St|tr+|jD]\}}t||||||<|St|ttfr8|Dcgc]}t||||}}t|tr t|}|Sycc}w)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. N)rr rrBrintfloatstrrfioBytesIOdictr#rlisttuple)rrrrkeyrvrets r4rrs(+w'K(( K /";// K .;'';eS% DE   K &%++- JC2|%8+ K   K$ /!  <1Dk R   k5 )*C  0 sD)?rtypingrrrrBtorch.distributed distributedr torch._utilsr!torch.distributed._shard.metadatar'torch.distributed._shard.sharded_tensorrtorch.distributed.tensorr torch.distributed.tensor._utilsr rTr r rrrrrplannerrrrrrr reshardingrrrrr__annotations__boolr5r>rrDrJrWr`rbrerirurxrrrrrrrrrr6r4rs] && +;A,Q 99c9>h>H>>Bx.h (^2ell7K }1E!-:(g)( +7D s ELL Y C  4 4(4+,4 (^ 4n?x" =S =# =$y/ =w3Gu||5I0J& C ]  h .4c3h4C4n,,,", ,r6