a d?@sddlmZmZddlZddlZddlZddlmZddlZddlZddl Z ddl Z ddl m Z m Z mZmZddlZddlmZddlmZddlmZdd lmZmZdd lmZmZmZdd lmZmZm Z m!Z!m"Z"m#Z#m$Z$m%Z%dd l&m'Z'gd Z(eGdddZ)eGdddZ*dZ+ejejdddZ,e$edddZ-GdddeZ.Gddde.Z/Gddde.Z0e$e1ddd Z2e e$e e e$d!d"d#Z3d$d%Z4ej5ej5e"e1e6d&d'd(Z7Gd)d*d*eZ8Gd+d,d,e j9Z:Gd-d.d.eZ;dS)/)ABCabstractmethodN) dataclass)ListUnionDictcast)Tensor)Future)Path)Metadata MetadataIndex) StorageReader StorageWriter WriteResult) LoadItemType LoadPlannerLoadPlanSavePlan SavePlannerReadItem WriteItem WriteItemType)narrow_tensor_by_index)FileSystemWriterSlicedBufferedReaderFileSystemReaderc@s*eZdZUdZeed<eed<eed<dS) _StorageInfoz, This is the per entry storage info relative_pathoffsetlengthN)__name__ __module__ __qualname____doc__str__annotations__intr)r)p/var/www/html/stable-diffusion-webui/venv/lib/python3.9/site-packages/torch/distributed/checkpoint/filesystem.pyr0s rc@seZdZUeed<dS)_StoragePrefixprefixN)r"r#r$r&r'r)r)r)r*r+;s r+z.distcp)tensorreturncCs,|}||kr(|}|SN)detachcpuZ_typed_storage_sizenumelclone)r-r)r)r*_trimCs r5)itemr.cCst|j||dS)N)index size_in_bytes storage_data)rr7)r6r8r9r)r)r*_result_from_write_itemJsr:c@s,eZdZeddZddZeddZdS) _TensorLoadercCsdSr/r)selfsizeobjr)r)r*addSsz_TensorLoader.addcCsdSr/r)r=r)r)r* start_loadingWsz_TensorLoader.start_loadingcCsdSr/r)rAr)r)r*valuesZsz_TensorLoader.valuesN)r"r#r$rr@rBrCr)r)r)r*r;Rs  r;c@s,eZdZddZddZddZddZd S) _SerialCpuLoadercCs||_g|_dSr/) resolve_funitems)r=rEr)r)r*__init__`sz_SerialCpuLoader.__init__cCs|j||fdSr/)rFappendr<r)r)r*r@dsz_SerialCpuLoader.addcCsdSr/r)rAr)r)r*rBgsz_SerialCpuLoader.start_loadingccsP|jD]D\}}||}|}||kr@|}||fVqdSr/)rFrEr0r1storager>r3r4r=_r?r-r)r)r*rCjsz_SerialCpuLoader.valuesN)r"r#r$rGr@rBrCr)r)r)r*rD_srDc@sReZdZdddZeddZddZd d Zd d Zd dZ ddZ ddZ dS)_OverlappingCpuLoaderN@BcCsd||_g|_||_d|_t|_d|_d|_|p:t j |_ |j t j kr`|j t j dS)NrF)rErFinflight_threshholdin_flight_data collectionsdeque current_itemsidxstartedtorchcudaZcurrent_streamstreamZ wait_stream)r=rErWrNr)r)r*rGws z_OverlappingCpuLoader.__init__cCs|jt|jkSr/)rSlenrFrAr)r)r*_donesz_OverlappingCpuLoader._donecCsbg}|j|jkr|j|j|jkr^|j}|j|d|d8_||q|SNr) rOrNrW synchronizerRpopleftr3 element_sizerH)r=drainedvalr)r)r*_drains    " z_OverlappingCpuLoader._draincCstj|j|js|j|jkr|j|j\}}|jd7_|| }|j rd|j ddd}n,|j t dkr| |kr|}|j||f|j||7_qWdn1s0YdS)Nr r1T)deviceZ non_blocking)rUrVrWrYrOrNrFrSrEr0is_cudatorarIr>r3r4rRrHr]rJr)r)r*_refills& z_OverlappingCpuLoader._refillcCs(|js Jt|jdkr"|j|jSrZ)rYrXrRrWr[rAr)r)r*_finishs  z_OverlappingCpuLoader._finishcCs"|jrtd|j||fdS)Nz&cannot add items after loading started)rT RuntimeErrorrFrHr<r)r)r*r@sz_OverlappingCpuLoader.addcCs.|jr dSd|_|jjddd|dS)NTcSs|dSrZr)xr)r)r*z5_OverlappingCpuLoader.start_loading..key)rTrFsortrdrAr)r)r*rBs z#_OverlappingCpuLoader.start_loadingccs<||js*|}||EdHq|EdHdSr/)rBrYr`rdre)r=r^r)r)r*rCs  z_OverlappingCpuLoader.values)NrM) r"r#r$rGpropertyrYr`rdrer@rBrCr)r)r)r*rLvs   rLcCsBd}|jdusJ|jjD] }||9}q|jjj}|tj|SNr )Z tensor_datar>Z propertiesdtyperU_utilsZ _element_size)r6r>srpr)r)r* _item_sizes    rs)rFr.c Cs|dkr|gSdd|D}dd|D}ddt|D}ddt|D}|jtddt|D]\}}||||qd|D]>}tt|d d d d }|||||t|7<q|S) Nr cSsg|]}|jtjkr|qSr)typerBYTE_IO.0wir)r)r* rjz+_split_by_size_and_type..cSsg|]}|jtjkr|qSr)rtrwr)r)r*rzrjcSsg|]}gqSr)r)rxrKr)r)r*rzrjcSsg|]}dqS)rr)r{r)r)r*rzrjT)rlreversecSs|dSror)rgr)r)r*rirjz)_split_by_size_and_type..rkr)rangermrs enumeraterHmin) ZbinsrFbytes_wtensor_wZbucketsZ bucket_sizesiryrSr)r)r*_split_by_size_and_typesrcCs|}|jtjkr4t|tjs$J||n0t|t j sDJ|j t dksXJt ||||}t ||t|||S)Nr1)tellrurrv isinstanceioBytesIOwrite getbufferrUr rasaver:r)rWdata write_item storage_keyr r!r)r)r* _write_items   r file_queue result_queueplannerrN use_fsyncc sNz0|\}}}tjr:|dkr:tfdd|d}ntfdd}dd|D} | D]} |t| | q\|dd|D} g} t |d |} | D]"} | }| t | || |q| D]&\}} |jrJ| t | || |q|rt| Wdn1s0Y|| qWntjyHYn0dS) Nrcs |Sr/ resolve_datargrr)r*ri rjz)_write_files_from_queue..)rNcs |Sr/rrgrr)r*rirjcSsg|]}|jtjkr|qSr)rtrwr)r)r*rzsz+_write_files_from_queue..cSsg|]}|jtjkr|qSr)rtrwr)r)r*rzswb) get_nowaitrUrVZ is_availablerLrDr@rsrBopenrrHrrCrbosfsyncfilenoputqueueEmpty)rrrrNr file_namerZ write_itemsloaderrrrZ write_resultsrWrr-r)rr*_write_files_from_queuesH       .rcseZdZdZdeeejfeee e ddfdd Z edd d d Z e e d d dZ ee ee dddZe eeeedddZeeeeddddZZS)raa 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)pathsingle_file_per_rank sync_files thread_countper_thread_copy_aheadr.cs0tt||_||_||_||_||_dS)a Initialize the writer pointing to `path` Args: path: diretory where the checkpoint will be writen 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. N. B. If sync_files is disabled, there's no guarantee that the checkpoint will be consistent in the case of a failure. N)superrGr rrrrr)r=rrrrr __class__r)r*rG?s   zFileSystemWriter.__init__)is_coordinatorr.cCsdSr/r))r=rr)r)r*set_up_storage_writerZsz&FileSystemWriter.set_up_storage_writerplanr.cCs|jjddd|S)NT)parentsexist_ok)rmkdirr=rr)r)r*prepare_local_plan]sz#FileSystemWriter.prepare_local_plan global_planr.cCsddt|D}|S)NcSs*g|]"\}}tj|td|ddqS)__rKr9) dataclassesreplacer+)rxrrr)r)r*rzdsz8FileSystemWriter.prepare_global_plan..)r~)r=rZ new_plansr)r)r*prepare_global_planasz$FileSystemWriter.prepare_global_planrrr.cs<|jdfdd}t}|jrXt|j|jD] }|}||j|||fq4n*|jD]"}|}||j|||gfq^t}g} t d|jD]2} t j t ||||j |jfd} | | | qt ||||j |jd| D] } | qg} z| |7} qWn*tjy6t} | | | YS0dS)Nrcsjt}d7|Sro)r,DEFAULT_SUFFIX)rZ file_countZ storage_planr)r*gen_filersz-FileSystemWriter.write_data..gen_filer )targetargsr)r9rQueuerrrrFrrr} threadingThreadrrrstartrHjoinrrr set_result)r=rrrrZbucketrr6rthreadsrKtresfutr)rr* write_datajsV     zFileSystemWriter.write_data)metadataresultsr.cCst}|D]}|dd|Dq ||_|jdd*}t||t| Wdn1sj0Y|jd |jddS)NcSsi|]}|j|jqSr))r7r9)rxwrr)r)r* rjz+FileSystemWriter.finish..z .metadata.tmpr .metadata) dictupdater9rrpickledumprrrrename)r=rrZ storage_mdZwr_list metadata_filer)r)r*finishs ,zFileSystemWriter.finish)TTr r)r"r#r$r%rr&rPathLikeboolr(rGrrrrrrr rrr r __classcell__r)r)rr*r1s2   B rcsVeZdZejeedfdd Zejfeeedfdd Z edfdd Z Z S) r) base_streamr rXcs&t|||_||_|ddSrZ)rrGr rXseek)r=rr rXrr)r*rGs zSlicedBufferedReader.__init__)_SlicedBufferedReader__offset_SlicedBufferedReader__whencer.csD|tjkr|j|}n |tjkr6tj}|j|j|}t||Sr/)rSEEK_SETr SEEK_ENDrXrr)r=rrrr)r*rs    zSlicedBufferedReader.seekr.cst|jSr/)rrr rArr)r*rszSlicedBufferedReader.tell) r"r#r$r RawIOBaser(rGrrrrrr)r)rr*rsrcseZdZeeejfddfdd ZedddZ e e e ddd d Z ed d d ZeeddddZe e dddZee ee dddZZS)rN)rr.cs tt||_t|_dSr/)rrGr rrr9)r=rrr)r*rGs  zFileSystemReader.__init__)sinfocCsttj|dd|j|jS)NF)closefd)rrFileIOrr r!)r=filerr)r)r* _slice_fileszFileSystemReader._slice_filerc Csft}|jD](}|j|j}|j}||g|q |D]\}}|j|d} |D]} |j| j}| | |} | j t j krt | |j} | d|| | q^tttj| dd} t| | j| j} || }|| ksJd| jd|d| || || |q^Wdq>1sF0Yq>t}| d|S)Nrbrr1)Z map_locationzreq z mismatch sizes z vs )!rrFr9Z storage_indexr setdefaultrHrrrrurrvrrreadr!rZ load_bytesrr rUloadrZstorage_offsetslengthsZresolve_tensorr0r>Zcopy_Z commit_tensorr r)r=rrZper_fileZ read_itemZitem_mdrrreqsrreqZ file_slicebytesr-Z target_tensorrr)r)r* read_datas<        0 zFileSystemReader.read_datarcCs>|jdd}t|WdS1s00YdS)Nrr)rrrr)r=rr)r)r* read_metadataszFileSystemReader.read_metadata)rrr.cCs|j|_|jdusJdSr/r)r=rrr)r)r*set_up_storage_readersz&FileSystemReader.set_up_storage_readerrcCs|Sr/r)rr)r)r*rsz#FileSystemReader.prepare_local_planrcCs|Sr/r))r=rr)r)r*r sz$FileSystemReader.prepare_global_plan)r"r#r$rr&rrrGrrrrr rr rrrrrrrr)r)rr*rs&r)<abcrrrrrPrrrrrtypingrrrrrUr Z torch.futuresr pathlibr rr rrIrrrrrrrrrrrrZtorch.distributed._shard._utilsr__all__rr+rr5r:r;rDrLr(rsrrrrrrBufferedReaderrrr)r)r)r*sZ    (    O   6