o h@sUddlZddlZddlZddlZddlZddlZddlZddlZddlZddl Z ddl m Z m Z ddl mZmZmZmZddlmZddlmZddlmZddlmZddlmZmZmZmZmZmZdd lm Z ddl!Z!dd l!m"Z"dd l#m$Z$m%Z%dd l&m'Z'dd l(m)Z)m*Z*ddl+m,Z,m-Z-m.Z.ddl/m0Z0m1Z1m2Z2m3Z3m4Z4m5Z5m6Z6m7Z7ddl8m9Z9ddl:m;Z;mm?Z?ddl@mAZAgdZBdZCeDeEd<eGdddZFeGdddZGdZHdeDfddZIGdd d e ZJGd!d"d"eJZKGd#d$d$eJZLGd%d&d&ZMd'e6deNfd(d)ZOd*eNd+ePe6dePePe6fd,d-ZQ .dLd/eMd0ejRd1eejSe!j"fd2e6d3eDd4eTde=fd5d6ZUd7ed8ejVd9ejVd:e5d/eMd;eNdd?ZWGd@dAdAe ZXGdBdCdCeXZYGdDdEdEe<ZZGdFdGdGZ[GdHdIdIe;Z\GdJdKdKeZe9Z]dS)MN)ABCabstractmethod) GeneratorIterableIteratorSequence)contextmanager) dataclass)UnsupportedOperation)Path)AnyCallablecastIOOptionalUnion)Buffer)Tensor)_get_available_device_type_get_device_module)narrow_tensor_by_index)ExtensionRegistryStreamTransformExtension)MetadataSTATE_DICT_TYPE StorageMeta) LoadItemTypeLoadPlan LoadPlannerReadItemSavePlan SavePlanner WriteItem WriteItemType)BlockingAsyncStager) StorageReader StorageWriter WriteResult)_create_file_view)Future)FileSystemWriterFileSystemReader FileSystemFileSystemBase .metadata _metadata_fnc@sFeZdZUdZeed<eed<eed<dZee eed<ddZ dS) _StorageInfoz#This is the per entry storage info. relative_pathoffsetlengthNtransform_descriptorscCsdd|jDS)NcSsi|] \}}|dur||qSN).0kvr6r6{/var/www/html/construction_image-detection-poc/venv/lib/python3.10/site-packages/torch/distributed/checkpoint/filesystem.py Csz-_StorageInfo.__getstate__..)__dict__itemsselfr6r6r: __getstate__Bz_StorageInfo.__getstate__) __name__ __module__ __qualname____doc__str__annotations__intr4rrr@r6r6r6r:r09s  r0c@seZdZUeed<dS)_StoragePrefixprefixN)rBrCrDrFrGr6r6r6r:rIFs rIz.distcpreturncCs ttSr5)rFuuiduuid4r6r6r6r:_generate_uuidN rNc@sTeZdZedededdfddZed ddZedee e j effd d Z dS) _TensorLoadersizeobjrKNcCdSr5r6r?rQrRr6r6r:addSz_TensorLoader.addcCrSr5r6r>r6r6r: start_loadingWrVz_TensorLoader.start_loadingcCrSr5r6r>r6r6r:values[rVz_TensorLoader.valuesrKN) rBrCrDrrHobjectrUrWrtupletorchrrXr6r6r6r:rPRs "rPc@sZeZdZdeddfddZdededdfdd Zdd d Zde e e j effd d Z dS)_SerialCpuLoader resolve_funrKNcCs||_g|_dSr5)r^r=)r?r^r6r6r:__init__as z_SerialCpuLoader.__init__rQrRcCs|j||fdSr5)r=appendrTr6r6r:rUerAz_SerialCpuLoader.addcCrSr5r6r>r6r6r:rWhz_SerialCpuLoader.start_loadingccsR|jD]"\}}||}|}||kr!|}||fVqdSr5)r=r^detachcpustoragerQnumelcloner?_rRtensorr6r6r:rXksz_SerialCpuLoader.valuesrY)rBrCrDr r_rHrZrUrWrr[r\rrXr6r6r6r:r]`s   r]c @seZdZ  ddedeejdeddfddZe de fd d Z de e ejeffd d Zdd dZdee ejeffddZdededdfddZdddZdee ejeffddZdS)_OverlappingCpuLoaderN@Br^streaminflight_threshholdrKcCs||_g|_||_d|_t|_d|_d|_|r|j nt |_ t |j |_ t tjj|p0|j |_|j|j krF|j|j dSdS)NrF)r^r=rmin_flight_data collectionsdeque current_itemsidxstarted device_typerr device_modulerr\cudaStreamcurrent_streamrl wait_stream)r?r^rlrmr6r6r:r_xs   z_OverlappingCpuLoader.__init__cCs|jt|jkSr5)rrlenr=r>r6r6r:_donesz_OverlappingCpuLoader._donecCslg}|j|jkr |j|j|jkr4|j}|j|d|d8_|||j|jks|SNr) rnrmrl synchronizerqpopleftre element_sizer`)r?drainedvalr6r6r:_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)NrcT)device non_blocking)rurlr{rnrmr=rrr^rbrtyperttor\untyped_storagerQreitemsizerfrqr`rrgr6r6r:_refills4  "z_OverlappingCpuLoader._refillcCs(|jsJt|jdkr|j|jSr|)r{rzrqrlr}r>r6r6r:_finishs  z_OverlappingCpuLoader._finishrQrRcCs"|jrtd|j||fdS)Nz&cannot add items after loading started)rs RuntimeErrorr=r`rTr6r6r:rUsz_OverlappingCpuLoader.addcCs0|jrdSd|_|jjtdd|dS)NTrkey)rsr=sortoperator itemgetterrr>r6r6r:rWs  z#_OverlappingCpuLoader.start_loadingccsB||js|}||EdH|jr|EdHdSr5)rWr{rrr)r?rr6r6r:rXs z_OverlappingCpuLoader.values)NrkrY)rBrCrDr rr\rwrHr_propertyboolr{listr[rrZrrrrrUrWrrXr6r6r6r:rjws(    rjc @sVeZdZdZ d deeeddfddZdede j de e e eeffd d ZdS) _StorageWriterTransforms This is experimental, and will likely move elsewhere in the future. It lives here to minimize changes while we are still learning and gathering feedback. N extensionsrKcCs|dur d|_dS||_dS)a If the extensions arg is None, this means the implementation should provide whatever defaults it chooses. An empty sequence indicates no extensions should be used. At this time, the default extensions sequence is empty. Nr6)r)r?rr6r6r:r_s z!_StorageWriterTransforms.__init__ write_item raw_streamcCsRGdddtj}ttt||}|jD]}||}q|ddt|jDfS)Nc@sDeZdZdejfddZdefddZdede fdd Z d d Z d S) zE_StorageWriterTransforms.transform_save_stream..NoCloseWriterrawcSs ||_dSr5)r)r?rr6r6r:r_s zN_StorageWriterTransforms.transform_save_stream..NoCloseWriter.__init__rKcSsdS)NTr6r>r6r6r: writeablerazO_StorageWriterTransforms.transform_save_stream..NoCloseWriter.writeablebcSs |j|Sr5)rwrite)r?rr6r6r:rrOzK_StorageWriterTransforms.transform_save_stream..NoCloseWriter.writecSs||jdSr5)flushrr>r6r6r:closeszK_StorageWriterTransforms.transform_save_stream..NoCloseWriter.closeN) rBrCrDioIOBaser_rrrrHrrr6r6r6r: NoCloseWriters  rcSsg|]}|qSr6)get_descriptor)r7exr6r6r: szB_StorageWriterTransforms.transform_save_stream..)rrrrbytesr transform_toreversed)r?rrrrrr6r6r:transform_save_streams    z._StorageWriterTransforms.transform_save_streamr5)rBrCrDrErrrr_r"rrr[rrrrFrr6r6r6r:rs   ritemcCsBd}|jdus J|jjD]}||9}q |jjj}|tj|SNr) tensor_datarQ propertiesdtyper\_utils _element_size)rrQsrr6r6r: _item_sizes    rbinsr=c Cs|dkr|gSdd|D}dd|D}ddt|D}ddt|D}|jtddt|D] \}}||||q2|D] }tt|tdd d }|||||t|7<qB|S) NrcSg|] }|jtjkr|qSr6rr#BYTE_IOr7wir6r6r:rz+_split_by_size_and_type..cSg|] }|jtjkr|qSr6rrr6r6r:rrcSsg|]}gqSr6r6r7rhr6r6r:rcSsg|]}dqS)rr6rr6r6r:rrT)rreverserr)rangerr enumerater`minrr) rr=bytes_wtensor_wbuckets bucket_sizesirrrr6r6r:_split_by_size_and_typesrF transformsrldatar storage_key safe_tensorsc Cs|}|||\}}|jtjkr"t|tjsJ|| nt|t j s*J|j t dks4J|s.cSrr6rrr6r6r:rrwb)r) get_nowaitr\_C_get_privateuse1_backend_namegetattrrv is_availablerj resolve_datar]rUrrWr`rrXis_cpurfqnsafetensors.torchrrosfsyncfilenoAttributeErrorr syncrputqueueEmpty)rrrrrrmrrr file_namer write_itemscustom_backend_namecustom_device_modloaderrrr write_resultsrlr tensor_dictrirr6r6r:_write_files_from_queueWs            (Lrc @sJeZdZeedeeejfdede e j ddffddZ edeeejfdedeeejffdd Z edeeejfd eeejfddfd d Zedeeejfdeeejffd dZedeeejfddfddZeedeeejfdefddZedeeejfdefddZedeeejfddfddZdS)r-pathmoderKNcCrSr5r6)r?rrr6r6r:rszFileSystemBase.create_streamsuffixcCrSr5r6r?rrr6r6r: concat_pathzFileSystemBase.concat_pathnew_pathcCrSr5r6r?rrr6r6r:renamerzFileSystemBase.renamecCrSr5r6r?rr6r6r: init_pathrazFileSystemBase.init_pathcCrSr5r6rr6r6r:mkdirrazFileSystemBase.mkdir checkpoint_idcCrSr5r6clsrr6r6r:validate_checkpoint_idrVz%FileSystemBase.validate_checkpoint_idcCrSr5r6rr6r6r:existsrazFileSystemBase.existscCrSr5r6rr6r6r:rm_filerazFileSystemBase.rm_file)rBrCrDrrrrFrPathLikerrrrrrrr classmethodrrrrr6r6r6r:r-sJ       ( "r-c @s*eZdZedeeejfdedee j ddffddZ deeejfdedeeejffdd Z deeejfdeeejffd d Z deeejfd eeejfddfd dZdeeejfddfddZedeeejfdefddZdeeejfdefddZdeeejfddfddZdS)r,rrrKNccsRt|ts t|}||}ttj|VWddS1s"wYdSr5)rr openrrr)r?rrrlr6r6r:rs   "zFileSystem.create_streamrcCst|ts t|}||Sr5rr rr6r6r:rs zFileSystem.concat_pathcCst|ts t|}|Sr5r rr6r6r:rs zFileSystem.init_pathrcCs&t|ts t|}|tt|dSr5)rr rrrr6r6r:rs zFileSystem.renamecCs$t|ts t|}|jddddS)NT)parentsexist_ok)rr rrr6r6r:rs zFileSystem.mkdirrcCsRt|trdSdt|vrdSt|jD]}|r&tt|tjr&dSqdS)NTz://F)rr rFr rraccessW_OK)rrpr6r6r:rs  z!FileSystem.validate_checkpoint_idcCst|ts t|}|Sr5)rr rrr6r6r:rs zFileSystem.existscCst|ts t|}|dSr5)rr unlinkrr6r6r:r s  zFileSystem.rm_file)rBrCrDrrrFrrrrrrrrrrrrrrrr6r6r6r:r,s:     &    r,cseZdZdZ      d2deeejfdeded e d e d ed e e e d e de ddffdd Zd3deeejdfddfddZdeddfddZdedefddZdeedeefddZdededeeefdd Z !d4ded"ejd#edeeefd$d%Zd&ed'eeeddfd(d)Zde efd*d+Ze deeejffd,d-Z!e deeejffd.d/Z"e#deeejfdefd0d1Z$Z%S)5_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. Tr逖Nrsingle_file_per_rank sync_filesrper_thread_copy_ahead overwrite _extensionsargskwargsrKc sTtt|_|j||_||_||_||_||_ t |_ ||_ 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. _extensions: Extensions to apply to output streams (EXPERIMENTAL) N. B. If sync_files is disabled, there's no guarantee that the checkpoint will be consistent in the case of a failure. N)superr_r,fsrrrrrrrNsave_idrrr) r?rrrrrrrrr __class__r6r:r_s z_FileSystemWriter.__init__rcCs|r |j||_t|_dSr5)rrrrNrr?rr6r6r:resetDs z_FileSystemWriter.resetis_coordinatorcCrSr5r6)r?r"r6r6r:set_up_storage_writerIraz'_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_pathrwarningswarnrr?r$r6r6r:prepare_local_planLsz$_FileSystemWriter.prepare_local_planplanscCsddt|D}|S)NcSs*g|]\}}tj|td|ddqS)__rhr) dataclassesreplacerI)r7rr$r6r6r:r[sz9_FileSystemWriter.prepare_global_plan..)r)r?r+ new_plansr6r6r:prepare_global_planZsz%_FileSystemWriter.prepare_global_planrc s|jdfdd}t}|jr1t|j|jD]}|}|j|j |}| |||fqn|jD]}|}|j|j |}| |||gfq4| ||S)Nrcsjt}d7|Sr)rJDEFAULT_SUFFIX)r file_count storage_planr6r:gen_fileisz._FileSystemWriter.write_data..gen_file) rrQueuerrrr=rrrr _write_data) r?r$rr6rbucketrrrr6r3r: write_dataas  z_FileSystemWriter.write_dataFrrc Cst}g}td|jD]!}tjt|jj||||j |j |j |j|f d}| | |q t|jj||||j |j |j |j|d |D]}|qCg}z ||7}qNtjyht} | || YSw)Nr)targetr) rrrrrrmrrr)rr7rr threadingThreadrrrrrrstartr`joinrrr) set_result) r?rrrrthreadsrhtresfutr6r6r:r8}sR    z_FileSystemWriter._write_datametadataresultsc Csi}|D] }|dd|Dq||_||_tt|j|jtd}|j |d*}t |||j rQz t |WnttfyPt YnwWdn1s[wY|j|jrn|j|j|j||jdS)NcSsi|]}|j|jqSr6)rr)r7wrr6r6r:r;sz,_FileSystemWriter.finish..z.tmpr)updater storage_metarr rrrr/rpickledumprrrrrr rrr&rr)r?rErF storage_mdwr_listtmp_path metadata_filer6r6r:finishs&    z_FileSystemWriter.finishcCst|j|jdS)N)rr)rrrr>r6r6r:rIsz_FileSystemWriter.storage_metacCstt|j|jtSr5)rr rrrr/r>r6r6r:r&sz_FileSystemWriter.metadata_pathcC|jS)zT return the checkpoint_id that will be used to save the checkpoint. rr>r6r6r:rz_FileSystemWriter.checkpoint_idcC t|Sr5r,rrr6r6r:r z(_FileSystemWriter.validate_checkpoint_id)TTrrTNr5F)&rBrCrDrErrFrrrrHrrrr r_r!r#r r*rr1r!r)r'r:rr7r8rrPrrIrr&rrr __classcell__r6r6rr:rst      %     3&rc @sNeZdZdZd deeddfddZdedee d e e de e fd d Z dS) _StorageReaderTransformsrNextension_registryrKcCs|dur t|_dS||_dSr5)rrZ)r?rZr6r6r:r_s z!_StorageReaderTransforms.__init__ read_itemr4rcCs2|j|}|}|D] }t|tr||}q |Sr5)rZfrom_descriptor_listrrtransform_from)r?r[r4rrr]rr6r6r:transform_load_streams   z._StorageReaderTransforms.transform_load_streamr5) rBrCrDrErrr_rrrFrrr^r6r6r6r:rYsrYcseZdZ d deeejfdeeddffdd Z de de e fdd Z d d eeejdfddfd d Zd edededfddZdefddZdededdfddZd edefddZdeedeefddZedeeejffddZed eeejfdefddZZS)!r+Nr_extension_registryrKcs<tt|_|j||_i|_t|_t ||_ dSr5) rr_r,rrrrrNload_idrYr)r?rr_rr6r:r_s zFileSystemReader.__init__sinfocCstttt||j|jSr5)rrrr(r2r3)r?filerar6r6r: _slice_fileszFileSystemReader._slice_filercCs$i|_|r |j||_t|_dSr5)rrrrrNr`r r6r6r:r!s zFileSystemReader.resetr$rc Csi}|jD]}|j|j}|j}||g|q|D]\}}|j|j|} |j | d} |D]} |j| j}| | |} |j | |j pJd| } | jtjkrgt| d}|d|| |q4| rn| }n t| d}|dtttj|ddd}t|| j| j}|| }| | ksJd| jd | d | |!||"| |q4Wdn1swYqt#}|$d|S) Nrbr6rrcT) map_location weights_onlyzreq z mismatch sizes z vs )%r=r storage_indexr1 setdefaultr`rrrrrcrr^r4rrrrrreadseek load_bytesseekablerrr\loadrstorage_offsetslengthsresolve_tensorrbrQcopy_ commit_tensorr)r@)r?r$rper_filer[item_mdrr1reqsrrlreq file_slicer] read_bytesrmri target_tensorrDr6r6r: read_data s^         - zFileSystemReader.read_datacCsn|j|jd}|j|d }t|}Wdn1swYt|dddur0t|_|j |j_ |S)Nr.rdrI) rrrrrJrnrrrIr`)r?rrOrEr6r6r: read_metadataFs  zFileSystemReader.read_metadatarEr"cCs|j|_|jdus JdSr5r-)r?rEr"r6r6r:set_up_storage_readerQsz&FileSystemReader.set_up_storage_readercC|Sr5r6r)r6r6r:r*Uraz#FileSystemReader.prepare_local_planr+cCr~r5r6)r?r+r6r6r:r1Xraz$FileSystemReader.prepare_global_plancCrQ)zT return the checkpoint_id that will be used to load the checkpoint. rRr>r6r6r:r[rSzFileSystemReader.checkpoint_idcCrTr5rUrr6r6r:rbrVz'FileSystemReader.validate_checkpoint_idr5)rBrCrDrrFrrrrr_r0rrrcr!rrr)r{rr|rr}r*rr1rrrrrXr6r6rr:r+s(  < &r+cszeZdZdZ       ddeeejfded ed e d e d ed ede e e ddfddZ dedeffdd ZZS)r*rTrrFNrrrrrcache_staged_state_dictrrrKc Cs,tj||||||||dtj||ddS)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. 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. _extensions: Extensions to apply to output streams (EXPERIMENTAL) N. B. If sync_files is disabled, there's no guarantee that the checkpoint will be consistent in the case of a failure. )rrrrrrr)rN)rr_r$) r?rrrrrrrrr6r6r:r_us  zFileSystemWriter.__init__ state_dictcsd|_t|S)zOverride of AsyncStager.stager)rrstage)r?rrr6r:rs zFileSystemWriter.stage)TTrrFTN)rBrCrDrErrFrrrrHrrrr_rrrXr6r6rr:r*gs:   +r*rW)^ror.rrrrJrr<rLr'abcrrcollections.abcrrrr contextlibrr r pathlibr typingr r rrrrtyping_extensionsrr\r torch._utilsrrtorch.distributed._shard._utilsr'torch.distributed.checkpoint._extensionrr%torch.distributed.checkpoint.metadatarrr$torch.distributed.checkpoint.plannerrrrrr r!r"r#$torch.distributed.checkpoint.stagingr$$torch.distributed.checkpoint.storager%r&r'"torch.distributed.checkpoint.utilsr( torch.futuresr)__all__r/rFrGr0rIr2rNrPr]rjrrHrrrrrrrr7rr-r,rrYr+r*r6r6r6r:s         (     Z3"  0  \"<Js