U "Afbd @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:K/tmp/pip-unpacked-wheel-ak3y1t3j/torch/distributed/checkpoint/filesystem.pyr/Bs r/c@seZdZUeed<dS)_StoragePrefixprefixN)r3r4r5r7r8r:r:r:r;r<Ks r<z.distcpreturncCs ttSN)r7uuidZuuid4r:r:r:r;_generate_uuidSsrBc@sTeZdZeeeddddZeddddZeee e j efddd Z dS) _TensorLoaderNsizeobjr?cCsdSr@r:selfrErFr:r:r;addXsz_TensorLoader.addr>cCsdSr@r:rHr:r:r; start_loading\sz_TensorLoader.start_loadingcCsdSr@r:rJr:r:r;values`sz_TensorLoader.values) r3r4r5rr9objectrIrKrrtorchrrLr:r:r:r;rCWs rCc@sXeZdZeddddZeeddddZddd d Ze e e j efdd d Z dS) _SerialCpuLoaderN) resolve_funr?cCs||_g|_dSr@)rPitems)rHrPr:r:r;__init__fsz_SerialCpuLoader.__init__rDcCs|j||fdSr@)rQappendrGr:r:r;rIjsz_SerialCpuLoader.addr>cCsdSr@r:rJr:r:r;rKmsz_SerialCpuLoader.start_loadingccsP|jD]D\}}||}|}||kr@|}||fVqdSr@)rQrPdetachcpuZstoragerEnumelclonerH_rFtensorr:r:r;rLpsz_SerialCpuLoader.values)r3r4r5rrRr9rMrIrKrrrNrrLr:r:r:r;rOesrOc@seZdZdeeejeddddZe e dddZ e e ejefdd d Zddd d Zee ejefdd dZeeddddZddddZee ejefdddZdS)_OverlappingCpuLoaderN@B)rPstreaminflight_threshholdr?cCs||_g|_||_d|_t|_d|_d|_|r8|j nt |_ t |j |_ t tjj|p`|j |_|j|j kr|j|j dS)NrF)rPrQr^in_flight_data collectionsdeque current_itemsidxstarted device_typerr device_moduler rNcudaStreamZcurrent_streamr]Z wait_stream)rHrPr]r^r:r:r;rR}s    z_OverlappingCpuLoader.__init__r>cCs|jt|jkSr@)rclenrQrJr:r:r;_donesz_OverlappingCpuLoader._donecCsbg}|j|jkr|j|j|jkr^|j}|j|d|d8_||q|SNr) r_r^r] synchronizerbpopleftrV element_sizerS)rHdrainedvalr: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)NrUT)deviceZ non_blocking)rfr]rjr_r^rQrcrPrTrstyperetorNZuntyped_storagerErVitemsizerWrbrSrnrXr:r:r;_refills&  z_OverlappingCpuLoader._refillcCs(|js tt|jdkr"|j|jSrk)rjAssertionErrorrirbr]rlrJr:r:r;_finishs  z_OverlappingCpuLoader._finishrDcCs"|jrtd|j||fdS)Nz&cannot add items after loading started)rd RuntimeErrorrQrSrGr:r:r;rIsz_OverlappingCpuLoader.addcCs0|jr dSd|_|jjtdd|dS)NTrkey)rdrQsortoperator itemgetterrwrJr:r:r;rKs z#_OverlappingCpuLoader.start_loadingccs<||js*|}||EdHq|EdHdSr@)rKrjrqrwry)rHror:r:r;rLs  z_OverlappingCpuLoader.values)Nr\)r3r4r5rrrNrhr9rRpropertyboolrjrrrrMrqrwr ryrIrKrrLr:r:r:r;r[|s   r[)itemr?cCsBd}|jdk st|jjD] }||9}q|jjj}|tj|SNrr)Z tensor_datarxrEZ propertiesdtyperN_utilsZ _element_size)rrEsrr:r:r; _item_sizes    r)binsrQr?c Cs|dkr|gSdd|D}dd|D}ddt|D}ddt|D}|jtddt|D]\}}||||qd|D]@}tt|tdd d }|||||t|7<q|S) NrrcSsg|]}|jtjkr|qSr:rtr"BYTE_IO.0wir:r:r; s z+_split_by_size_and_type..cSsg|]}|jtjkr|qSr:rrr:r:r;rs cSsg|]}gqSr:r:rrYr:r:r;rscSsg|]}dqS)rr:rr:r:r;rsT)r|reverser{r)ranger}r enumeraterSminr~r) rrQbytes_wtensor_wZbucketsZ bucket_sizesirrcr: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)NrU)indexZ size_in_bytes storage_data)tellrtr"r isinstanceioBytesIOrxwrite getbufferrNrrssaver 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)Nrrr)r^cSsg|]}|jtjkr|qSr:rrr:r:r;r6s z+_write_files_from_queue..cSsg|]}|jtjkr|qSr:rrr:r:r;r;s wb) get_nowaitrNZ_CZ_get_privateuse1_backend_namegetattrrgZ is_availabler[Z resolve_datarOrIrrKrSrrLZis_cpurxosfsyncfilenoAttributeErrorsyncputqueueEmpty)rrrrr^rr file_namerZ write_itemsZcustom_backend_nameZcustom_device_modloaderrrrZ write_resultsr]rrZr: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:)rHrrr:r:r;rVszFileSystemBase.create_streamrsuffixr?cCsdSr@r:rHrrr:r:r; concat_path]szFileSystemBase.concat_pathrnew_pathr?cCsdSr@r:rHrrr:r:r;renamecszFileSystemBase.renamerr?cCsdSr@r:rHrr: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)rHrrr]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. Trr逖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) superrRr+fsrrrrrrrBsave_idr) rHrrrrrrrr __class__r:r;rRs z_FileSystemWriter.__init__rcCs|r|j||_t|_dSr@)rrrrBrrHrr:r:r;resetsz_FileSystemWriter.reset)is_coordinatorr?cCsdSr@r:)rHrr: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_pathrwarningswarnrzrHrr:r:r;prepare_local_plansz$_FileSystemWriter.prepare_local_planplansr?cCsddt|D}|S)NcSs*g|]"\}}tj|td|ddqS)__rY)r) dataclassesreplacer<)rrrr:r:r;rsz9_FileSystemWriter.prepare_global_plan..)r)rHrZ 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_filerr)targetr)rrrrr^rr)rrQueuerrrrQrrrrr threadingThreadrrrrstartrSjoinrrr( set_result)rHrrrrZbucketrrrrthreadsrYtresfutr: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)rHr r Z storage_mdZwr_listZtmp_path metadata_filer:r:r;finish@s   z_FileSystemWriter.finishr>cCst|j|jdS)N)rr)rrrrJr:r:r;rWsz_FileSystemWriter.storage_metacCstt|j|jtSr@)r rrrrr.rJr:r:r;rZsz_FileSystemWriter.metadata_pathcCs|jSzT return the checkpoint_id that will be used to save the checkpoint. rrJr:r:r;r^sz_FileSystemWriter.checkpoint_idcCs t|Sr@r+rrr:r:r;resz(_FileSystemWriter.validate_checkpoint_id)TTrrrT)N)!r3r4r5r6rr7rrrr9rrRrrrrrrr 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@) rrRr+rrrrrrBload_idrrr:r;rRks  zFileSystemReader.__init__)sinfor?cCst||j|jSr@)r'r1r2)rHfilerr:r:r; _slice_filerszFileSystemReader._slice_filercCs&t|_|r|j||_t|_dSr@)rrrrrrBrrr:r:r;ruszFileSystemReader.resetrc Csht}|jD](}|j|j}|j}||g|q |D]\}}|j|j |} |j | d} |D]} |j| j}| | |} | j t jkrt| |j} | d|| | qltttjttt| dd}t|| j| j}|| }| | ks0t!d| jd| d| |"||#| |qlW5QRXq>t$}|%d|S)NrbrrU)Z map_locationzreq z mismatch sizes z vs )&rrQrZ storage_indexr0 setdefaultrSrrrrrrtrrrrreadr2seekZ load_bytesr rrNloadr rrZstorage_offsetslengthsZresolve_tensorrTrErxZcopy_Z commit_tensorr(r)rHrrZper_fileZ read_itemZitem_mdrr0reqsrr]reqZ file_slice read_bytesrZZ target_tensorrr:r:r; read_data{sD        zFileSystemReader.read_datar>c CsZ|j|jd}|j|d}t|}W5QRXt|dddkrLt|_|j |j_ |S)Nr-rr) rrrrrr!rrrr)rHrrr r:r:r; read_metadatas zFileSystemReader.read_metadata)r rr?cCs|j|_|jdk stdSr@)rrx)rHr rr:r:r;set_up_storage_readersz&FileSystemReader.set_up_storage_readerrcCs|Sr@r:rr:r:r;rsz#FileSystemReader.prepare_local_planrcCs|Sr@r:)rHrr:r:r;rsz$FileSystemReader.prepare_global_plancCs|jSrrrJr:r:r;rszFileSystemReader.checkpoint_idcCs t|Sr@rrr:r:r;rsz'FileSystemReader.validate_checkpoint_id)N)r3r4r5rr7rrrRr/rrrrrrr(r&rr'rr(rrrrrrrrr:r:rr;r*js( c sTeZdZdZd eeejfeee e eeddfdd Z e e d fd d Z Z S)r)rTrrrFN)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)rrR)rHrrrrrr)rrr:r;rRszFileSystemWriter.__init__) state_dictr?csd|_t|S)zOverride of AsyncStager.stager)rrstage)rHr*rr:r;r+szFileSystemWriter.stage)TTrrrFT)r3r4r5r6rr7rrrr9rRrr+rr:r:rr;r)s$ $)Wr`rrr~rrrrrArabcrr contextlibrrpathlibrtypingrrr r r r r rrrrrrNrZ 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<rrBrCrOr[r9rrrrrrrrr,r+rr*r)r:r:r:r;sv   8  (     Z   D*/=Z