o ߗi0e@sUddlZddlZddlZddlZddlZddlZddlZddlZddlZddl Z ddl m Z m Z ddl mZddlmZddlmZddlmZmZmZmZmZmZmZmZmZmZmZmZddlZddlm Z ddl!m"Z"m#Z#dd l$m%Z%dd l&m'Z'm(Z(m)Z)m*Z*dd l+m,Z,m-Z-m.Z.m/Z/m0Z0m1Z1m2Z2m3Z3dd l4m5Z5dd l6m7Z7m8Z8m9Z9ddl:m;Z;ddldZ?e@eAd<eGdddZBeGdddZCdZDde@fddZEGddde ZFGdddeFZGGdd d eFZHd!e2deIfd"d#ZJd$eId%ee2deee2fd&d'ZKd(ejLd)eejMej fd*e2d+e@de9f d,d-ZNd.ed/ejOd0ejOd1e1d2eId3ePd4eIddfd5d6ZQGd7d8d8e ZRGd9d:d:eRZSGd;d<dd>e7ZUGd?d@d@eTe5ZVdS)AN)ABCabstractmethod)contextmanager) dataclass)Path) AnyCallablecastDict GeneratorIOIterableIteratorListOptionalTupleUnion)Tensor)_get_available_device_type_get_device_module)narrow_tensor_by_index)Metadata MetadataIndexSTATE_DICT_TYPE StorageMeta) LoadItemTypeLoadPlan LoadPlannerReadItemSavePlan SavePlanner WriteItem WriteItemType)BlockingAsyncStager) StorageReader StorageWriter WriteResult)_create_file_view)Future)FileSystemWriterFileSystemReader FileSystemFileSystemBase .metadata _metadata_fnc@s*eZdZUdZeed<eed<eed<dS) _StorageInfoz#This is the per entry storage info. relative_pathoffsetlengthN)__name__ __module__ __qualname____doc__str__annotations__intr:r:e/home/ubuntu/transcripts/venv/lib/python3.10/site-packages/torch/distributed/checkpoint/filesystem.pyr/Bs  r/c@seZdZUeed<dS)_StoragePrefixprefixN)r3r4r5r7r8r:r:r:r;r<Ks r<z.distcpreturncCs ttSN)r7uuiduuid4r:r:r:r;_generate_uuidSs rBc@sTeZdZedededdfddZed ddZedee e j effd d Z dS) _TensorLoadersizeobjr>NcCdSr?r:selfrDrEr:r:r;addXz_TensorLoader.addcCrFr?r:rHr:r:r; start_loading\rJz_TensorLoader.start_loadingcCrFr?r:rKr:r:r;values`rJz_TensorLoader.valuesr>N) r3r4r5rr9objectrIrLrrtorchrrMr:r:r:r;rCWs "rCc@sZeZdZdeddfddZdededdfdd Zdd d Zde e e j effd d Z dS)_SerialCpuLoader resolve_funr>NcCs||_g|_dSr?)rRitems)rHrRr:r:r;__init__fs z_SerialCpuLoader.__init__rDrEcCs|j||fdSr?)rSappendrGr:r:r;rIjsz_SerialCpuLoader.addcCrFr?r:rKr:r:r;rLmz_SerialCpuLoader.start_loadingccsR|jD]"\}}||}|}||kr!|}||fVqdSr?)rSrRdetachcpustoragerDnumelclonerH_rEtensorr:r:r;rMpsz_SerialCpuLoader.valuesrN)r3r4r5rrTr9rOrIrLrrrPrrMr:r:r:r;rQes   rQc @seZdZ  ddedeejdeddfddZe de fd d Z de e ejeffd d Zdd dZdee ejeffddZdededdfddZdddZdee ejeffddZdS)_OverlappingCpuLoaderN@BrRstreaminflight_threshholdr>cCs||_g|_||_d|_t|_d|_d|_|r|j nt |_ t |j |_ t tjj|p0|j |_|j|j krF|j|j dSdS)NrF)rRrSrbin_flight_data collectionsdeque current_itemsidxstarted device_typerr device_moduler rPcudaStreamcurrent_streamra wait_stream)rHrRrarbr:r:r;rT}s   z_OverlappingCpuLoader.__init__cCs|jt|jkSr?)rglenrSrKr:r:r;_donesz_OverlappingCpuLoader._donecCslg}|j|jkr |j|j|jkr4|j}|j|d|d8_|||j|jks|SNr) rcrbra synchronizerfpopleftrZ element_sizerU)rHdrainedvalr:r:r;_drains    "  z_OverlappingCpuLoader._draincCs.|j|j|jsu|j|jkr}|j|j\}}|jd7_||}|j j |j kr6|j ddd}n|j t dkrO|||jkrO|}|j||f|j||7_|js|j|jksWddSWddSWddSWddS1swYdS)NrXT)device non_blocking)rjrarprcrbrSrgrRrWrytyperitorPuntyped_storagerDrZitemsizer[rfrUrtr\r:r:r;_refills4  "z_OverlappingCpuLoader._refillcCs(|jsJt|jdkr|j|jSrq)rprorfrarrrKr:r:r;_finishs  z_OverlappingCpuLoader._finishrDrEcCs"|jrtd|j||fdS)Nz&cannot add items after loading started)rh RuntimeErrorrSrUrGr:r:r;rIsz_OverlappingCpuLoader.addcCs0|jrdSd|_|jjtdd|dS)NTrkey)rhrSsortoperator itemgetterrrKr:r:r;rLs  z#_OverlappingCpuLoader.start_loadingccsB||js|}||EdH|jr|EdHdSr?)rLrprwrr)rHrur:r:r;rMs z_OverlappingCpuLoader.values)Nr`rN)r3r4r5rrrPrlr9rTpropertyboolrprrrrOrwrr rrIrLrrMr:r:r:r;r_|s(    r_itemcCsBd}|jdus J|jjD]}||9}q |jjj}|tj|SNrx) tensor_datarD propertiesdtyperP_utils _element_size)rrDsrr:r:r; _item_sizes    rbinsrSc Cs|dkr|gSdd|D}dd|D}ddt|D}ddt|D}|jtddt|D] \}}||||q2|D] }tt|tdd d }|||||t|7<qB|S) NrxcSg|] }|jtjkr|qSr:r{r"BYTE_IO.0wir:r:r; z+_split_by_size_and_type..cSg|] }|jtjkr|qSr:rrr:r:r;rrcSsg|]}gqSr:r:rr]r:r:r;rcSsg|]}dqS)rr:rr:r:r;rrT)rreverserr)rangerr enumeraterUminrr) rrSbytes_wtensor_wbuckets bucket_sizesirrgr:r:r;_split_by_size_and_typesrradata write_item storage_keycCs|}|jtjkrt|tjsJ||nt|t j s"J|j t dks,Jt |t tt|||}t|j|t|||dS)NrX)index size_in_bytes storage_data)tellr{r"r isinstanceioBytesIOwrite getbufferrPrrysaver r bytesr&rr/)rarrrr1r2r:r:r; _write_items   r create_stream file_queue result_queueplannerrb use_fsync thread_countc Csz |\}}} tj} tt| d} |dkr/tjs#| r/| r/|dkr/t|j|d} nt |j} dd| D} | D] }| t ||q=| dd| D}g}||dM}|D]}||}| t||||q]| D]\}}|js|J| t||||qs|rz t|Wn tytYnwWdn1swY||qtjyYdSw) NTrxr)rbcSrr:rrr:r:r;r6rz+_write_files_from_queue..cSrr:rrr:r:r;r;rwb) get_nowaitrP_C_get_privateuse1_backend_namegetattrrk is_availabler_ resolve_datarQrIrrLrUrrMis_cpuosfsyncfilenoAttributeErrorsyncputqueueEmpty)rrrrrbrr file_namer write_itemscustom_backend_namecustom_device_modloaderrrr write_resultsrarr^r:r:r;_write_files_from_queuesd           6rc @sJeZdZeedeeejfdede e j ddffddZ edeeejfdedeeejffdd Z edeeejfd eeejfddfd d Zedeeejfdeeejffd dZedeeejfddfddZeedeeejfdefddZedeeejfdefddZedeeejfddfddZdS)r,pathmoder>NcCrFr?r:)rHrrr:r:r;rVszFileSystemBase.create_streamsuffixcCrFr?r:rHrrr:r:r; concat_path]zFileSystemBase.concat_pathnew_pathcCrFr?r:rHrrr:r:r;renamecrzFileSystemBase.renamecCrFr?r:rHrr:r:r; init_pathirJzFileSystemBase.init_pathcCrFr?r:rr:r:r;mkdirmrJzFileSystemBase.mkdir checkpoint_idcCrFr?r:clsrr:r:r;validate_checkpoint_idqsz%FileSystemBase.validate_checkpoint_idcCrFr?r:rr:r:r;existsvrJzFileSystemBase.existscCrFr?r:rr:r:r;rm_filezrJzFileSystemBase.rm_file)r3r4r5rrrr7rPathLiker rIOBaserrrrr classmethodrrrrr:r:r:r;r,UsJ       ( "r,c @s*eZdZedeeejfdedee j ddffddZ deeejfdedeeejffdd Z deeejfdeeejffd d Z deeejfd eeejfddfd dZdeeejfddfddZedeeejfdefddZdeeejfdefddZdeeejfddfddZdS)r+rrr>NccsFtt||}ttj|VWddS1swYdSr?)r ropenrr)rHrrrar:r:r;rs"zFileSystem.create_streamrcCstt||Sr?)r rrr:r:r;rszFileSystem.concat_pathcCst|ts t|}|Sr?)rrrr:r:r;rs zFileSystem.init_pathrcCstt|tt|dSr?)r rrrr:r:r;rszFileSystem.renamecCstt|jddddS)NT)parentsexist_ok)r rrrr:r:r;rszFileSystem.mkdirrcCsRt|trdSdt|vrdSt|jD]}|r&tt|tjr&dSqdS)NTz://F)rrr7rrraccessW_OK)rrpr:r:r;rs  z!FileSystem.validate_checkpoint_idcCstt|Sr?)r rrrr:r:r;rszFileSystem.existscCstt|dSr?)r runlinkrr:r:r;rszFileSystem.rm_file)r3r4r5rrr7rrr rrrrrrrrrrrrr:r:r:r;r+s:     &    r+cs`eZdZdZ     d,deeejfdedede d e d ed e d e d dffdd Z d-deeejdfd dfddZ ded dfddZ ded efddZdeed eefddZdeded eeefddZd ed!eeed dfd"d#Zd eefd$d%Zed eeejffd&d'Zed eeejffd(d)Zedeeejfd efd*d+ZZ S)._FileSystemWritera Basic implementation of StorageWriter using file IO. This implementation makes the following assumptions and simplifications: * The checkpoint path is an empty or non-existing directory. * File creation is atomic The checkpoint consist of one file per write request plus a `.metadata` file with the serialized metadata. Trx逖rsingle_file_per_rank sync_filesrper_thread_copy_ahead overwriteargskwargsr>Nc sJtt|_|j||_||_||_||_||_ t |_ ||_ dS)a Initialize the writer pointing to `path`. Args: path: directory where the checkpoint will be written to. single_file_per_rank: Produce one file per rank instead of one file per tensor/blob. Default to True. sync_files : force files to be synced to permanent storage. Default to True. thread_count: Number of IO threads to use to write. Default to 1. per_thread_copy_ahead: How many bytes to copy from the GPU ahead of saving then. Default 10Mb. overwrite: Whether to allow overwriting existing checkpoints. Defaults to True. N. B. If sync_files is disabled, there's no guarantee that the checkpoint will be consistent in the case of a failure. N) superrTr+fsrrrrrrrBsave_idr) rHrrrrrrrr __class__r:r;rTs  z_FileSystemWriter.__init__rcCs|r |j||_t|_dSr?)rrrrBrrHrr:r:r;resets z_FileSystemWriter.resetis_coordinatorcCrFr?r:)rHrr:r:r;set_up_storage_writerrVz'_FileSystemWriter.set_up_storage_writerplancCsX|j|j|j|jr*|jr!td|jd|jd|Std|jd|S)Nz#Detected an existing checkpoint in z#, overwriting since self.overwrite=z. Past version 2.5 of PyTorch, `overwrite` will default to False. Set this variable to True to maintain this functionality or False to raise when an existing checkpoint is found.z-Checkpoint already exists and self.overwrite=.) rrrr metadata_pathrwarningswarnrrHrr:r:r;prepare_local_plansz$_FileSystemWriter.prepare_local_planplanscCsddt|D}|S)NcSs*g|]\}}tj|td|ddqS)__r]r) dataclassesreplacer<)rrrr:r:r;rsz9_FileSystemWriter.prepare_global_plan..)r)rHr  new_plansr:r:r;prepare_global_plansz%_FileSystemWriter.prepare_global_planrc s\|jdfdd}t}|jr1t|j|jD]}|}|j|j |}| |||fqn|jD]}|}|j|j |}| |||gfq4t} g} t d|jD]} t j t|jj|| ||j|j|jfd} | | | qWt|jj|| ||j|j|jd| D]} | qg} z | | 7} qtjyt}|| |YSw)Nrcsjt}d7|Sr)r=DEFAULT_SUFFIX)r file_count storage_planr:r;gen_filesz._FileSystemWriter.write_data..gen_filerx)targetr)rrrrrbrr)rrQueuerrrrSrrrrr threadingThreadrrrrstartrUjoinrrr( set_result)rHrrrrbucketrrrrthreadsr]tresfutr:rr; write_datasf     z_FileSystemWriter.write_datametadataresultsc Csi}|D] }|dd|Dq||_||_tt|j|jtd}|j |d(}t |||j rOz t |Wn tyNt YnwWdn1sYwY|j|jrl|j|j|j||jdS)NcSsi|]}|j|jqSr:)rr)rwrr:r:r; Asz,_FileSystemWriter.finish..z.tmpr)updater storage_metar rrrrr.rpickledumprrrrrrrrrr)rHr"r# storage_mdwr_listtmp_path metadata_filer:r:r;finish>s&     z_FileSystemWriter.finishcCst|j|jdS)N)rr)rrrrKr:r:r;r'Uz_FileSystemWriter.storage_metacCstt|j|jtSr?)r rrrrr.rKr:r:r;rXsz_FileSystemWriter.metadata_pathcC|jS)zT return the checkpoint_id that will be used to save the checkpoint. rrKr:r:r;r\z_FileSystemWriter.checkpoint_idcC t|Sr?r+rrr:r:r;rc z(_FileSystemWriter.validate_checkpoint_id)TTrxrTr?)!r3r4r5r6rr7rrrr9rrTrrrrrrr r(r&r!rr.rrr'rrrrr __classcell__r:r:rr;rsZ    "  C&rcs eZdZdeeejfddffdd Zdede j fddZ dd eeejdfddfd d Z d e d ededfddZdefddZdededdfddZd e de fddZdee dee fddZedeeejffddZed eeejfdefddZZS) r*rr>Ncs2tt|_|j||_i|_t|_dSr?) rrTr+rrrrrBload_idrrr:r;rTis  zFileSystemReader.__init__sinfocCst||j|jSr?)r'r1r2)rHfiler8r:r:r; _slice_filepr/zFileSystemReader._slice_filercCs$i|_|r |j||_t|_dSr?)rrrrrBr7rr:r:r;rss zFileSystemReader.resetrrc Csxi}|jD]}|j|j}|j}||g|q|D]\}}|j|j|} |j | dx} |D]m} |j| j}| | |} | j t j kr]t| |j} | d|| | q4tttjttt| ddd}t|| j| j}|| }||ksJd| jd|d|| ||!| |q4Wdn1swYqt"}|#d|S) NrbrrXT) map_location weights_onlyzreq z mismatch sizes z vs )$rSr storage_indexr0 setdefaultrUrrrrr:r{rrrrreadr2seek load_bytesr rrPloadr rrstorage_offsetslengthsresolve_tensorrWrDcopy_ commit_tensorr(r)rHrrper_file read_itemitem_mdrr0reqsrrareq file_slice read_bytesr^ target_tensorr r:r:r; read_dataysL          zFileSystemReader.read_datacCsn|j|jd}|j|d }t|}Wdn1swYt|dddur0t|_|j |j_ |S)Nr-r;r') rrrrr(rCrrr'r7)rHrr-r"r:r:r; read_metadatas  zFileSystemReader.read_metadatar"rcCs|j|_|jdus JdSr?r )rHr"rr:r:r;set_up_storage_readersz&FileSystemReader.set_up_storage_readercC|Sr?r:rr:r:r;rrVz#FileSystemReader.prepare_local_planr cCrTr?r:)rHr r:r:r;rrVz$FileSystemReader.prepare_global_plancCr0)zT return the checkpoint_id that will be used to load the checkpoint. r1rKr:r:r;rr2zFileSystemReader.checkpoint_idcCr3r?r4rr:r:r;rr5z'FileSystemReader.validate_checkpoint_idr?)r3r4r5rr7rrrTr/rrr:rrrr(rQrrRrrSrrrrrrrr6r:r:rr;r*hs  , &r*csleZdZdZ      ddeeejfdeded e d e d ed ed dfddZ de d e ffdd Z Z S)r)rTrxrFrrrrrcache_staged_state_dictrr>Nc Cs*tj|||||||dtj||ddS)aM Initialize the writer pointing to `path`. Args: path: directory where the checkpoint will be written to. single_file_per_rank: Produce one file per rank instead of one file per tensor/blob. Default to True. sync_files : force files to be synced to permanent storage. Default to True. thread_count: Number of IO threads to use to write. Default to 1. per_thread_copy_ahead: How many bytes to copy from the GPU ahead of saving then. Default 10Mb. cache_staged_state_dict: Whether to cache the staged state_dict. This option decreases staging latency at the cost of increases memory usage. Additionally, if this parameter is set to True, it's the expectation that the stager is maintained and re-used for multiple dcp.async_save calls. Default to False. overwrite: Whether to allow overwriting existing checkpoints. Defaults to True. N. B. If sync_files is disabled, there's no guarantee that the checkpoint will be consistent in the case of a failure. )rrrrrr)rUN)rrTr#)rHrrrrrrUrr:r:r;rTs  zFileSystemWriter.__init__ state_dictcsd|_t|S)zOverride of AsyncStager.stager)rrstage)rHrVrr:r;rWs zFileSystemWriter.stage)TTrxrFT)r3r4r5r6rr7rrrr9rTrrWr6r:r:rr;r)s4  (r))Wrdr rrrr(rrr@rabcrr contextlibrrpathlibrtypingrrr r r r r rrrrrrPr torch._utilsrrtorch.distributed._shard._utilsr%torch.distributed.checkpoint.metadatarrrr$torch.distributed.checkpoint.plannerrrrrrr r!r"$torch.distributed.checkpoint.stagingr#$torch.distributed.checkpoint.storager$r%r&"torch.distributed.checkpoint.utilsr' torch.futuresr(__all__r.r7r8r/r<rrBrCrQr_r9rrrrrrrrr,r+rr*r)r:r:r:r;s    8  (    Z"   D*/;^