U yhd @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<eGdddZBeGdddZCdZDe@dddZEGddde ZFGd d!d!eFZGGd"d#d#eFZHe2eId$d%d&ZJeIee2eee2d'd(d)ZKejLeejMej fe2e@e9d*d+d,ZNeejOejOe1eIePeIdd-d.d/ZQGd0dde ZRGd1ddeRZSGd2d3d3e8ZTGd4dde7ZUGd5ddeTe5ZVdS)6N)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)FutureFileSystemWriterFileSystemReader 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:Y/var/www/html/venv/lib/python3.8/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_uuidSsrCc@sTeZdZeeeddddZeddddZeee e j efddd Z dS) _TensorLoaderNsizeobjr?cCsdSr@r:selfrFrGr:r:r;addXsz_TensorLoader.addr>cCsdSr@r:rIr:r:r; start_loading\sz_TensorLoader.start_loadingcCsdSr@r:rKr:r:r;values`sz_TensorLoader.values) r3r4r5rr9objectrJrLrrtorchrrMr:r:r:r;rDWs rDc@sXeZdZeddddZeeddddZddd d Ze e e j efdd d Z dS) _SerialCpuLoaderN) resolve_funr?cCs||_g|_dSr@)rQitems)rIrQr:r:r;__init__fsz_SerialCpuLoader.__init__rEcCs|j||fdSr@)rRappendrHr:r:r;rJjsz_SerialCpuLoader.addr>cCsdSr@r:rKr:r:r;rLmsz_SerialCpuLoader.start_loadingccsP|jD]D\}}||}|}||kr@|}||fVqdSr@)rRrQdetachcpuZstoragerFnumelclonerI_rGtensorr:r:r;rMpsz_SerialCpuLoader.values)r3r4r5rrSr9rNrJrLrrrOrrMr:r:r:r;rPesrPc@seZdZdeeejeddddZe e dddZ e e ejefdd d Zddd d Zee ejefdd dZeeddddZddddZee ejefdddZdS)_OverlappingCpuLoaderN@B)rQstreaminflight_threshholdr?cCs||_g|_||_d|_t|_d|_d|_|r8|j nt |_ t |j |_ t tjj|p`|j |_|j|j kr|j|j dS)NrF)rQrRr_in_flight_data collectionsdeque current_itemsidxstarted device_typerr device_moduler rOcudaStreamZcurrent_streamr^Z wait_stream)rIrQr^r_r:r:r;rS}s    z_OverlappingCpuLoader.__init__r>cCs|jt|jkSr@)rdlenrRrKr:r:r;_donesz_OverlappingCpuLoader._donecCsbg}|j|jkr|j|j|jkr^|j}|j|d|d8_||q|SNr) r`r_r^ synchronizercpopleftrW element_sizerT)rIdrainedvalr:r:r;_drains    " z_OverlappingCpuLoader._drainc Cs|j|j|js|j|jkr|j|j\}}|jd7_||}|j j |j krl|j ddd}n2|j t dkr|||jkr|}|j||f|j||7_qW5QRXdS)NrVT)deviceZ non_blocking)rgr^rkr`r_rRrdrQrUrttyperftorOZuntyped_storagerFrWitemsizerXrcrTrorYr:r:r;_refills&  z_OverlappingCpuLoader._refillcCs(|js tt|jdkr"|j|jSrl)rkAssertionErrorrjrcr^rmrKr:r:r;_finishs  z_OverlappingCpuLoader._finishrEcCs"|jrtd|j||fdS)Nz&cannot add items after loading started)re RuntimeErrorrRrTrHr:r:r;rJsz_OverlappingCpuLoader.addcCs0|jr dSd|_|jjtdd|dS)NTrkey)rerRsortoperator itemgetterrxrKr:r:r;rLs z#_OverlappingCpuLoader.start_loadingccs<||js*|}||EdHq|EdHdSr@)rLrkrrrxrz)rIrpr:r:r;rMs  z_OverlappingCpuLoader.values)Nr])r3r4r5rrrOrir9rSpropertyboolrkrrrrNrrrxr rzrJrLrrMr:r:r:r;r\|s   r\)itemr?cCsBd}|jdk st|jjD] }||9}q|jjj}|tj|SNrs)Z tensor_dataryrF propertiesdtyperOZ_utilsZ _element_size)rrFsrr:r:r; _item_sizes    r)binsrRr?c Cs|dkr|gSdd|D}dd|D}ddt|D}ddt|D}|jtddt|D]\}}||||qd|D]@}tt|tdd d }|||||t|7<q|S) NrscSsg|]}|jtjkr|qSr:rur"BYTE_IO.0wir:r:r; s z+_split_by_size_and_type..cSsg|]}|jtjkr|qSr:rrr:r:r;rs cSsg|]}gqSr:r:rrZr:r:r;rscSsg|]}dqS)rr:rr:r:r;rsT)r}reverser|r)ranger~r enumeraterTminrr) rrRbytes_wtensor_wZbucketsZ bucket_sizesirrdr:r:r;_split_by_size_and_typesr)r^data write_item storage_keyr?cCs|}|jtjkr4t|tjs$t|| n:t|t j sDt|j t dksXtt |ttt|||}t|j|t|||dS)NrV)indexZ size_in_bytes storage_data)tellrur"r isinstanceioBytesIOrywrite getbufferrOrrtsaver r bytesr&rr/)r^rrrr1r2r:r:r; _write_items   r) create_stream file_queue result_queueplannerr_ use_fsync thread_countr?c Cs|z\|\}}} 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||||q| D]&\}}|jst| t||||q|rFzt|Wntk rDtYnXW5QRX||qWntjk rvYnXdS)Nrsr)r_cSsg|]}|jtjkr|qSr:rrr:r:r;r6s z+_write_files_from_queue..cSsg|]}|jtjkr|qSr:rrr:r:r;r;s wb) get_nowaitrOZ_CZ_get_privateuse1_backend_namegetattrrhZ is_availabler\Z resolve_datarPrJrrLrTrrMZis_cpuryosfsyncfilenoAttributeErrorsyncputqueueEmpty)rrrrr_rr file_namerZ write_itemsZcustom_backend_nameZcustom_device_modloaderrrrZ write_resultsr^rr[r:r:r;_write_files_from_queues\        rc@s4eZdZeeeeejfee e j ddfdddZ eeeejfeeeejfdddZ eeeejfeeejfddd d Zeeeejfeeejfd d d Zeeeejfdd ddZeeeeejfedddZeeeejfed ddZeeeejfdd ddZdS)r,Npathmoder?cCsdSr@r:)rIrrr:r:r;rVszFileSystemBase.create_streamrsuffixr?cCsdSr@r:rIrrr:r:r; concat_path]szFileSystemBase.concat_pathrnew_pathr?cCsdSr@r:rIrrr:r:r;renamecszFileSystemBase.renamerr?cCsdSr@r:rIrr:r:r; init_pathiszFileSystemBase.init_pathcCsdSr@r:rr:r:r;mkdirmszFileSystemBase.mkdir checkpoint_idr?cCsdSr@r:clsrr:r:r;validate_checkpoint_idqsz%FileSystemBase.validate_checkpoint_idcCsdSr@r:rr:r:r;existsvszFileSystemBase.existscCsdSr@r:rr:r:r;rm_filezszFileSystemBase.rm_file)r3r4r5rrrr7rPathLiker rIOBaserrrrr classmethodrrrrr:r:r:r;r,Us6    &c@seZdZeeeejfeee j ddfdddZ eeejfeeeejfdddZ eeejfeeejfdd d Z eeejfeeejfdd d d ZeeejfddddZeeeejfedddZeeejfedddZeeejfddddZdS)r+Nrc cs.tt||}ttj|VW5QRXdSr@)r ropenrr)rIrrr^r:r:r;rszFileSystem.create_streamrcCstt||Sr@)r rrr:r:r;rszFileSystem.concat_pathrcCst|tst|}|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|krdSt|jD]$}|r(tt|tjr(dSq(dS)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&   $   c s4eZdZdZd$eeejfeee e ee e dd fdd Z d%eeejdfdd d d Z edd d dZ eedddZeeeedddZeeeeedddZeeeeddddZeedddZeeeejfdddZeeeejfdd d!Zeeeejfed d"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. Trs逖N) rsingle_file_per_rank sync_filesrper_thread_copy_ahead overwriteargskwargsr?c 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) superrSr+fsrrrrrrrCsave_idr) rIrrrrrrrr __class__r:r;rSs z_FileSystemWriter.__init__rcCs|r|j||_t|_dSr@)rrrrCrrIrr:r:r;resetsz_FileSystemWriter.reset)is_coordinatorr?cCsdSr@r:)rIrr:r:r;set_up_storage_writersz'_FileSystemWriter.set_up_storage_writerplanr?cCsV|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_pathrwarningswarnr{rIrr:r:r;prepare_local_plansz$_FileSystemWriter.prepare_local_planplansr?cCsddt|D}|S)NcSs*g|]"\}}tj|td|ddqS)__rZ)r) dataclassesreplacer<)rrrr:r:r;rsz9_FileSystemWriter.prepare_global_plan..)r)rIrZ new_plansr:r:r;prepare_global_plansz%_FileSystemWriter.prepare_global_planrrr?c sj|jdfdd}t}|jrbt|j|jD]*}|}|j|j |}| |||fq4n4|jD],}|}|j|j |}| |||gfqht} g} t d|jD]<} t j t|jj|| ||j|j|jfd} | | | qt|jj|| ||j|j|jd| D]} | qg} z| | 7} q&Wn,tjk rdt}|| |YSXdS)Nrcsjt}d7|Sr)r=DEFAULT_SUFFIX)rZ file_countZ storage_planr:r;gen_filesz._FileSystemWriter.write_data..gen_filers)targetr)rrrrr_rr)rrQueuerrrrRrrrrr threadingThreadrrrrstartrTjoinrrr( set_result)rIrrrrZbucketrrrrthreadsrZtresfutr:rr; write_datas^      z_FileSystemWriter.write_data)metadataresultsr?c Cst}|D]}|dd|Dq ||_||_tt|j|jt d}|j |dH}t |||j rzt|Wntk rtYnXW5QRX|j|jr|j|j|j||jdS)NcSsi|]}|j|jqSr:)rr)rwrr:r:r; Csz,_FileSystemWriter.finish..z.tmpr)dictupdater storage_metar rrrrr.rpickledumprrrrrrrrrr)rIr r Z storage_mdZwr_listZtmp_path metadata_filer:r:r;finish@s   z_FileSystemWriter.finishr>cCst|j|jdS)N)rr)rrrrKr:r:r;rWsz_FileSystemWriter.storage_metacCstt|j|jtSr@)r rrrrr.rKr:r:r;rZsz_FileSystemWriter.metadata_pathcCs|jSzT return the checkpoint_id that will be used to save the checkpoint. rrKr:r:r;r^sz_FileSystemWriter.checkpoint_idcCs t|Sr@r+rrr:r:r;resz(_FileSystemWriter.validate_checkpoint_id)TTrsrT)N)!r3r4r5r6rr7rrrr9rrSrrrrrrr r(r&r rrrrrrrrrr __classcell__r:r:rr;rsB "  ErcseZdZeeejfddfdd Zee j dddZ deeejdfddd d Z e eedd d d ZedddZeeddddZe e dddZee ee dddZeeeejfdddZeeeejfedddZZS)r*Nrcs4tt|_|j||_t|_t|_ dSr@) rrSr+rrrrrrCload_idrrr:r;rSks  zFileSystemReader.__init__)sinfor?cCst||j|jSr@)r'r1r2)rIfilerr:r:r; _slice_filerszFileSystemReader._slice_filercCs&t|_|r|j||_t|_dSr@)rrrrrrCrrr:r:r;ruszFileSystemReader.resetrc Csjt}|jD](}|j|j}|j}||g|q |D]\}}|j|j |} |j | d} |D]} |j| j}| | |} | j t jkrt| |j} | d|| | qltttjttt| ddd}t|| j| j}|| }| | ks2t!d| jd| d| |"||#| |qlW5QRXq>t$}|%d|S) NrbrrVT)Z map_locationZ weights_onlyzreq z mismatch sizes z vs )&rrRrZ storage_indexr0 setdefaultrTrrrrrrurrrrreadr2seekZ load_bytesr rrOloadr rrZstorage_offsetslengthsZresolve_tensorrUrFryZcopy_Z commit_tensorr(r)rIrrZper_fileZ read_itemZitem_mdrr0reqsrr^reqZ file_slice read_bytesr[Z target_tensorr r:r:r; read_data{sL         zFileSystemReader.read_datar>c CsZ|j|jd}|j|d}t|}W5QRXt|dddkrLt|_|j |j_ |S)Nr-rr) rrrrrr"rrrr)rIrrr r:r:r; read_metadatas zFileSystemReader.read_metadata)r rr?cCs|j|_|jdk stdSr@)rry)rIr rr:r:r;set_up_storage_readersz&FileSystemReader.set_up_storage_readerrcCs|Sr@r:rr:r:r;rsz#FileSystemReader.prepare_local_planrcCs|Sr@r:)rIrr:r:r;rsz$FileSystemReader.prepare_global_plancCs|jSrrrKr:r:r;rszFileSystemReader.checkpoint_idcCs t|Sr@rrr:r:r;rsz'FileSystemReader.validate_checkpoint_id)N)r3r4r5rr7rrrSr/rrrrrrr(r'rr(rr)rrrrrrrrr:r:rr;r*js, c sTeZdZdZd eeejfeee e eeddfdd Z e e d fd d Z Z S)r)rTrsrFN)rrrrrcache_staged_state_dictrr?c stj|||||||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*rN)rrS)rIrrrrrr*rrr:r;rSszFileSystemWriter.__init__) state_dictr?csd|_t|S)zOverride of AsyncStager.stager)rrstage)rIr+rr:r;r,szFileSystemWriter.stage)TTrsrFT)r3r4r5r6rr7rrrr9rSrr,rr:r:rr;r)s$ $)WrarrrrrrrrArabcrr contextlibrrpathlibrtypingrrr r r r r rrrrrrOrZ torch._utilsrrZtorch.distributed._shard._utilsrZ%torch.distributed.checkpoint.metadatarrrrZ$torch.distributed.checkpoint.plannerrrrrrr r!r"Z$torch.distributed.checkpoint.stagingr#Z$torch.distributed.checkpoint.storager$r%r&Z"torch.distributed.checkpoint.utilsr'Z torch.futuresr(__all__r.r7r8r/r<rrCrDrPr\r9rrrrrrrrr,r+rr*r)r:r:r:r;sv   8  (     Z   D*/=^