a h @s0UddlZddlZddlZddlZddlZddlZddlZddlZddlZddl Z ddl Z ddl m Z m Z ddlmZmZmZmZddlmZddlmZddlmZddlmZddlmZdd lmZmZmZmZm Z m!Z!dd l"m#Z#ddl$Z$dd l$m%Z%dd l&m'Z'm(Z(dd l)m*Z*ddl+m,Z,m-Z-ddl.m/Z/m0Z0m1Z1m2Z2m3Z3ddl4m5Z5m6Z6m7Z7ddl8m9Z9m:Z:m;Z;mZ>m?Z?m@Z@ddlAmBZBddlCmDZDmEZEmFZFddlGmHZHddlImJZJgdZKdZLeMeNd<eGdddZOeGdddZPGdddeZQdZReMd d!d"ZSGd#d$d$e ZTGd%d&d&eTZUGd'd(d(eTZVGd)d*d*ZWe?eXd+d,d-ZYeXeZe?eZeZe?d.d/d0Z[eWej\e!ej]e$j%fe?eMeQeFd1d2d3Z^eej_ej_e>eWeXe`eXeQdd4 d5d6ZaGd7d8d8e ZbGd9d:d:ebZcGd;d<dd>ZeGd?d@d@eDZfGdAdBdBedeBZgdS)CN)ABCabstractmethod) GeneratorIterableIteratorSequence)contextmanager) dataclass)Enum)UnsupportedOperation)Path)AnyCallablecastIOOptionalUnion)Buffer)Tensor)_get_available_device_type_get_device_module)narrow_tensor_by_index)ExtensionRegistryStreamTransformExtension)CUSTOM_METADATA_KEYDCP_VERSION_KEY FORMAT_KEY FORMAT_VALUEHF_DCP_VERSION)MetadataSTATE_DICT_TYPE StorageMeta) LoadItemTypeLoadPlan LoadPlannerReadItemSavePlan SavePlanner WriteItem WriteItemType)BlockingAsyncStager) StorageReader StorageWriter WriteResult)_create_file_view)Future)FileSystemWriterFileSystemReader FileSystemFileSystemBaseSerializationFormat .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).0kvr=r=U/var/www/auris/lib/python3.9/site-packages/torch/distributed/checkpoint/filesystem.py Rz-_StorageInfo.__getstate__..)__dict__itemsselfr=r=rA __getstate__Qsz_StorageInfo.__getstate__) __name__ __module__ __qualname____doc__str__annotations__intr;rrrHr=r=r=rAr7Hs r7c@seZdZUeed<dS)_StoragePrefixprefixN)rIrJrKrMrNr=r=r=rArPUs rPc@seZdZdZdZdS)r4Z torch_saveZ safetensorsN)rIrJrK TORCH_SAVE SAFETENSORSr=r=r=rAr4Zsr4z.distcpreturncCs ttSr<)rMuuiduuid4r=r=r=rA_generate_uuidbsrXc@sTeZdZeeeddddZeddddZeee e j efddd Z dS) _TensorLoaderNsizeobjrUcCsdSr<r=rGr[r\r=r=rAaddgsz_TensorLoader.addrTcCsdSr<r=rFr=r=rA start_loadingksz_TensorLoader.start_loadingcCsdSr<r=rFr=r=rAvaluesosz_TensorLoader.values) rIrJrKrrOobjectr^r_rtupletorchrr`r=r=r=rArYfs rYc@sXeZdZeddddZeeddddZddd d Ze e e j efdd d Z dS) _SerialCpuLoaderN) resolve_funrUcCs||_g|_dSr<)rerE)rGrer=r=rA__init__usz_SerialCpuLoader.__init__rZcCs|j||fdSr<)rEappendr]r=r=rAr^ysz_SerialCpuLoader.addrTcCsdSr<r=rFr=r=rAr_|sz_SerialCpuLoader.start_loadingccsP|jD]D\}}||}|}||kr@|}||fVqdSr<)rEredetachcpuZstorager[numelclonerG_r\tensorr=r=rAr`sz_SerialCpuLoader.values)rIrJrKrrfrOrar^r_rrbrcrr`r=r=r=rArdtsrdc@seZdZdeeejeddddZe e dddZ e e ejefdd d Zddd d Zee ejefdd dZeeddddZddddZee ejefdddZdS)_OverlappingCpuLoaderN@B)restreaminflight_threshholdrUcCs||_g|_||_d|_t|_d|_d|_|r8|j nt |_ t |j |_ t tjj|p`|j |_|j|j kr|j|j dS)NrF)rerErrin_flight_data collectionsdeque current_itemsidxstarted device_typerr device_modulerrccudaStreamZcurrent_streamrqZ wait_stream)rGrerqrrr=r=rArfs  z_OverlappingCpuLoader.__init__rTcCs|jt|jkSr<)rwlenrErFr=r=rA_donesz_OverlappingCpuLoader._donecCsbg}|j|jkr|j|j|jkr^|j}|j|d|d8_||q|SNr) rsrrrq synchronizervpopleftrj element_sizerg)rGdrainedvalr=r=rA_drains    " z_OverlappingCpuLoader._draincCs|j|j|js|j|jkr|j|j\}}|jd7_||}|j j |j krl|j ddd}n2|j t dkr|||jkr|}|j||f|j||7_qWdn1s0YdS)NriT)deviceZ non_blocking)rzrqr~rsrrrErwrerhrtyperytorcZuntyped_storager[rjitemsizerkrvrgrrlr=r=rA_refills&  z_OverlappingCpuLoader._refillcCs(|js Jt|jdkr"|j|jSr)r~r}rvrqrrFr=r=rA_finishs  z_OverlappingCpuLoader._finishrZcCs"|jrtd|j||fdS)Nz&cannot add items after loading started)rx RuntimeErrorrErgr]r=r=rAr^sz_OverlappingCpuLoader.addcCs0|jr dSd|_|jjtdd|dS)NTrkey)rxrEsortoperator itemgetterrrFr=r=rAr_s z#_OverlappingCpuLoader.start_loadingccs<||js*|}||EdHq|EdHdSr<)r_r~rrr)rGrr=r=rAr`s  z_OverlappingCpuLoader.values)Nrp)rIrJrKrrrcr|rOrfpropertyboolr~listrbrrarrrrr^r_rr`r=r=r=rAros   roc@sNeZdZdZd eeeddddZee j e e e eefdddZdS) _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) extensionsrUcCs|dur dn||_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. Nr=)r)rGrr=r=rArfs z!_StorageWriterTransforms.__init__) write_item raw_streamrUcCsRGdddtj}ttt||}|jD]}||}q*|ddt|jDfS)Nc@sBeZdZejdddZedddZee ddd Z d d Z d S) zE_StorageWriterTransforms.transform_save_stream..NoCloseWriterrawcSs ||_dSr<r)rGrr=r=rArfszN_StorageWriterTransforms.transform_save_stream..NoCloseWriter.__init__rTcSsdS)NTr=rFr=r=rA writeableszO_StorageWriterTransforms.transform_save_stream..NoCloseWriter.writeable)brUcSs |j|Sr<)rwrite)rGrr=r=rArszK_StorageWriterTransforms.transform_save_stream..NoCloseWriter.writecSs||jdSr<)flushrrFr=r=rAclose szK_StorageWriterTransforms.transform_save_stream..NoCloseWriter.closeN) rIrJrKioIOBaserfrrrrOrrr=r=r=rA NoCloseWritersrcSsg|] }|qSr=)Zget_descriptor)r>exr=r=rA rCzB_StorageWriterTransforms.transform_save_stream..)rrrrbytesr transform_toreversed)rGrrrrrr=r=rAtransform_save_streams   z._StorageWriterTransforms.transform_save_stream)N)rIrJrKrLrrrrfr(rrrbrrrrMrr=r=r=rArs  r)itemrUcCsBd}|jdusJ|jjD] }||9}q|jjj}|tj|SNr) tensor_datar[Z propertiesdtyperc_utilsZ _element_size)rr[srr=r=rA _item_sizes    r)binsrErUc Cs|dkr|gSdd|D}dd|D}ddt|D}ddt|D}|jtddt|D]\}}||||qd|D]@}tt|tdd d }|||||t|7<q|S) NrcSsg|]}|jtjkr|qSr=rr)BYTE_IOr>wir=r=rAr'rCz+_split_by_size_and_type..cSsg|]}|jtjkr|qSr=rrr=r=rAr(rCcSsg|]}gqSr=r=r>rmr=r=rAr*rCcSsg|]}dqS)rr=rr=r=rAr+rCT)rreverserr)rangerr enumeratergminrr) rrEbytes_wtensor_wZbucketsZ bucket_sizesirrwr=r=rA_split_by_size_and_type#sr) transformsrqdatar storage_keyserialization_formatrUc Cs|}|||\}}|jtjkrDt|tjs4J|| n:t|t j sTJ|j t dkshJ|t jkr~t ||||t jkst|tjr||} n||} t|dkrdn|} t|j| t||| | ddS)Nrir)r;)indexZ size_in_bytes storage_data)tellrrr)r isinstancerBytesIOr getbufferrcrrr4rRsaverrjrr}r-rr7) rrqrrrrr9rr;r:Zinfo_transform_descriptorsr=r=rA _write_item;s:    r) create_stream file_queue result_queueplannerrrr use_fsync thread_countrrUc  Csz|\} } } tj} tt| d} |dkr^tjsF| r^| r^|dkr^t|j|d}n t |j}dd| D}|D]}| t ||qz| dd| D}g}|| d}|D]&}||}| t||||| |qi}i}|D]N\}}|jsJ| t||||| ||||jj<d|jjji||jj<q|tjkrdd lm}|||tt|ttt t!t"id |rzt#$|%Wn t&t'fyt#(Yn0|)Wdn1s0Y|*|qWnt+j,y Yn0dS) Nrr)rrcSsg|]}|jtjkr|qSr=rrr=r=rArrCz+_write_files_from_queue..cSsg|]}|jtjkr|qSr=rrr=r=rArrCwbZ saved_offsets)r)metadata)- get_nowaitrcZ_CZ_get_privateuse1_backend_namegetattrr{Z is_availableroZ resolve_datardr^rr_rgrr`Zis_cpurZfqnrchunkoffsetsr4rSZsafetensors.torchrrrjsondumpsrrMrrrosfsyncfilenoAttributeErrorr syncrputqueueEmpty)rrrrrrrrrr file_namerZ write_itemsZcustom_backend_nameZcustom_device_modloaderrrrZ write_resultsrqrZ tensor_dictZ metadata_dictrnrr=r=rA_write_files_from_queuems            (rc@s4eZdZeeeeejfee e j ddfdddZ eeeejfeeeejfdddZ eeeejfeeejfddd d Zeeeejfeeejfd d d Zeeeejfdd ddZeeeeejfedddZeeeejfed ddZeeeejfdd ddZdS)r3NpathmoderUcCsdSr<r=)rGrrr=r=rArszFileSystemBase.create_streamrsuffixrUcCsdSr<r=rGrrr=r=rA concat_pathszFileSystemBase.concat_pathrnew_pathrUcCsdSr<r=rGrrr=r=rArenameszFileSystemBase.renamerrUcCsdSr<r=rGrr=r=rA init_pathszFileSystemBase.init_pathcCsdSr<r=rr=r=rAmkdirszFileSystemBase.mkdir checkpoint_idrUcCsdSr<r=clsrr=r=rAvalidate_checkpoint_idsz%FileSystemBase.validate_checkpoint_idcCsdSr<r=rr=r=rAexistsszFileSystemBase.existscCsdSr<r=rr=r=rArm_fileszFileSystemBase.rm_file)rIrJrKrrrrMrPathLikerrrrrrrr classmethodrrrrr=r=r=rAr3s0 &r3c@s2eZdZeeeejfeee j ddfdddZ eeejfeeeejfdddZ eeejfeeejfdd d Z eeejfeeejfdd d d ZeeejfddddZeeeejfedddZeeejfedddZeeejfddddZeeejfeedddZdS)r2NrccsNt|tst|}||}ttj|VWdn1s@0YdSr<)rr openrrr)rGrrrqr=r=rArs  zFileSystem.create_streamrcCst|tst|}||Sr<rr rr=r=rArs zFileSystem.concat_pathrcCst|tst|}|Sr<rrr=r=rAr s zFileSystem.init_pathrcCs&t|tst|}|tt|dSr<)rr rrrr=r=rArs zFileSystem.renamecCs$t|tst|}|jddddS)NT)parentsexist_ok)rr rrr=r=rArs zFileSystem.mkdirrcCsRt|trdSdt|vrdSt|jD]$}|r(tt|tjr(dSq(dS)NTz://F)rr rMrrraccessW_OK)rrpr=r=rArs  z!FileSystem.validate_checkpoint_idcCst|tst|}|Sr<)rr rrr=r=rAr)s zFileSystem.existscCst|tst|}|dSr<)rr unlinkrr=r=rAr.s zFileSystem.rm_filecCs$t|tst|}dd|DS)NcSsg|] }t|qSr=)rM)r>rr=r=rAr6rCz!FileSystem.ls..)rr iterdirrr=r=rAls3s z FileSystem.ls)rIrJrKrrrMrrrrrrrrrrrrrrrrrr=r=r=rAr2s"  $  r2c sleZdZdZddddddejfeeej fe e e e e e e eeeedd fdd Zd'eeej dfdd d d Ze dd d dZeedddZeeeedddZeeeeedddZeejeeedddZeeeeddddZe e ddd Z!e"eeej fdd!d"Z#e"eeej fdd#d$Z$e%eeej fe d d%d&Z&Z'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. Tr逖N) rsingle_file_per_rank sync_filesrper_thread_copy_ahead overwrite _extensionsrargskwargsrUc sZtt|_|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)superrfr2fsrrr r rr rXsave_idr rrr) rGrr r rr r rrrr __class__r=rArfGs  z_FileSystemWriter.__init__rcCs|r|j||_t|_dSr<)rrrrXrrGrr=r=rAresetnsz_FileSystemWriter.reset)is_coordinatorrUcCsdSr<r=)rGrr=r=rAset_up_storage_writerssz'_FileSystemWriter.set_up_storage_writerplanrUcCsV|j|j|j|jrR|jr@td|jd|jdntd|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_pathr warningswarnrrGrr=r=rAprepare_local_planvsz$_FileSystemWriter.prepare_local_planplansrUcCsddt|D}|S)NcSs*g|]"\}}tj|td|ddqS)__rmr) dataclassesreplacerP)r>rrr=r=rArsz9_FileSystemWriter.prepare_global_plan..)r)rGr#Z new_plansr=r=rAprepare_global_plansz%_FileSystemWriter.prepare_global_planrrrUc s|jdfdd}t}|jrbt|j|jD]*}|}|j|j |}| |||fq4n4|jD],}|}|j|j |}| |||gfqh| ||S)Nrcsjt}d7|Sr)rQDEFAULT_SUFFIX)rZ file_countZ storage_planr=rAgen_filesz._FileSystemWriter.write_data..gen_file) rrQueuer rrrErrrr _write_data) rGrrr,rZbucketrrrr=r+rA write_datas z_FileSystemWriter.write_data)rrrUc Cst}g}td|jD]D}tjt|jj||||j |j |j |j|j f d}| ||qt|jj||||j |j |j |j|j d |D] }|qg}z||7}qWn(tjyt}|||YS0dS)Nr)targetr) rrrrrrrrrr)rr-rr threadingThreadrrrrr r rstartrgjoinrrr/ set_result) rGrrrthreadsrmtresfutr=r=rAr.sL    z_FileSystemWriter._write_data)rresultsrUc Csi}|D]}|dd|Dq||_||_tt|j|jtd}|j |dT}t |||j rzt |Wnttfyt Yn0Wdn1s0Y|j|jr|j|j|j||jdS)NcSsi|]}|j|jqSr=)rr)r>wrr=r=rArBrCz,_FileSystemWriter.finish..z.tmpr)updater storage_metarr rrrr6rpickledumpr rrrrr rrrrr)rGrr:Z storage_mdZwr_listZtmp_path metadata_filer=r=rAfinishs   ,z_FileSystemWriter.finishrTcCst|j|jdS)N)rr)r!rrrFr=r=rAr=sz_FileSystemWriter.storage_metacCstt|j|jtSr<)rr rrrr6rFr=r=rArsz_FileSystemWriter.metadata_pathcCs|jS)zT return the checkpoint_id that will be used to save the checkpoint. rrFr=r=rArsz_FileSystemWriter.checkpoint_idcCs t|Sr<r2rrr=r=rArsz(_FileSystemWriter.validate_checkpoint_id)N)(rIrJrKrLr4rRrrMrrrrOrrrr rfrrr&r!rr(r'r/r-r/rr-r.rrAr!r=rrrrr __classcell__r=r=rrAr9sR  '   2rc@sFeZdZdZd eeddddZeee e e e e dddZ dS) _StorageReaderTransformsrN)extension_registryrUcCs|durtn||_dSr<)rrF)rGrFr=r=rArf sz!_StorageReaderTransforms.__init__) read_itemr;rrUcCs2|j|}|}|D]}t|tr||}q|Sr<)rFZfrom_descriptor_listrrtransform_from)rGrGr;rrrHrr=r=rAtransform_load_streams    z._StorageReaderTransforms.transform_load_stream)N) rIrJrKrLrrrfr%rrMrrrIr=r=r=rArEsrEcseZdZdeeejfeeddfdd Z e e e dddZ deeejdfddd d Zeeedd d d ZedddZeeddddZeedddZeeeedddZeeeejfdddZeeeejfedddZZS) r1N)r_extension_registryrUcs<tt|_|j||_i|_t|_t ||_ dSr<) rrfr2rrrrrXload_idrEr)rGrrJrr=rArfs  zFileSystemReader.__init__)sinforUcCstttt||j|jSr<)rrrr.r9r:)rGfilerLr=r=rA _slice_file*szFileSystemReader._slice_filercCs$i|_|r|j||_t|_dSr<)rrrrrXrKrr=r=rAr-szFileSystemReader.resetr)c Csi}|jD](}|j|j}|j}||g|q |D]`\}}|j|j|} |j | d&} |D]} |j| j}| | |} |j | |j pd| } | jtjkrt| d}|d|| |ql| r| }nt| d}|dtttj|ddd}t|| j| j}|| }| | ksfJd| jd | d | |!||"| |qlWdq<1s0Yq0Yt|dddur`t|_|j |j_ |S)Nr5rOr=) rrrrr>rUrr!r=rK)rGrr@rr=r=rA read_metadataos( zFileSystemReader.read_metadata)rrrUcCs|j|_|jdusJdSr<r%)rGrrr=r=rAset_up_storage_readerzsz&FileSystemReader.set_up_storage_readerrcCs|Sr<r=r r=r=rAr!~sz#FileSystemReader.prepare_local_planr"cCs|Sr<r=)rGr#r=r=rAr(sz$FileSystemReader.prepare_global_plancCs|jS)zT return the checkpoint_id that will be used to load the checkpoint. rBrFr=r=rArszFileSystemReader.checkpoint_idcCs t|Sr<rCrr=r=rArsz'FileSystemReader.validate_checkpoint_id)N)N)rIrJrKrrMrrrrrfr7rrrNrr#r$r/rZrr[rr\r!rr(rrrrrDr=r=rrAr1s"  < r1c sneZdZdZdddddddejfeeej fe e e e e e e e eedd dd Zeed fd d ZZS) r0rTrr FN) rr r rr cache_staged_state_dictr rrrUc Cs.tj||||||||| d tj||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 reused 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. )rr r rr r rr)r]N)rrfr*) rGrr r rr r]r rrr=r=rArfs zFileSystemWriter.__init__) state_dictrUcsd|_t|S)zOverride of AsyncStager.stager)r rstage)rGr^rr=rAr_szFileSystemWriter.stage)rIrJrKrLr4rRrrMrrrrOrrrrfr r_rDr=r=rrAr0s,   -r0)hrtr&rrrrr>rr1rVrabcrrcollections.abcrrrr contextlibrr enumr r pathlibr typingr rrrrrZtyping_extensionsrrcrZ torch._utilsrrZtorch.distributed._shard._utilsrZ'torch.distributed.checkpoint._extensionrrZ&torch.distributed.checkpoint._hf_utilsrrrrrZ%torch.distributed.checkpoint.metadatarr r!Z$torch.distributed.checkpoint.plannerr"r#r$r%r&r'r(r)Z$torch.distributed.checkpoint.stagingr*Z$torch.distributed.checkpoint.storager+r,r-Z"torch.distributed.checkpoint.utilsr.Z torch.futuresr/__all__r6rMrNr7rPr4r*rXrYrdrorrOrrrrrrr-rrr3r2rrEr1r0r=r=r=rAs          (     Z3  3 i"AKs