o !iD= @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'dd l(m)Z)ddl*m+Z+ddgZ,eGdddZ-eGdddZ.dZ/dejdejfddZ0de$defddZ1GdddeZ2Gdd d e2Z3Gd!d"d"e2Z4de$de5fd#d$Z6d%e e$de e e$fd&d'Z7d(d)Z8d*ej9d+ej9d,e"d-e5d.e:f d/d0Z;Gd1ddeZr*r*r+rBgsz_SerialCpuLoader.addcCr=r0r*rDr*r*r+rEjz_SerialCpuLoader.start_loadingccsR|jD]"\}}||}|}||kr!|}||fVqdSr0)rIrHr1r2storager@r4r5r?_rAr.r*r*r+rFmsz_SerialCpuLoader.valuesN)r#r$r%rJrBrErFr*r*r*r+rGbs  rGc@sReZdZdddZeddZddZd d Zd d Zd dZ ddZ ddZ dS)_OverlappingCpuLoaderN@BcCs||_g|_||_d|_t|_d|_d|_|r|j nt dj |_ t |j |_|p/|j|_|j|jkrD|j|jdSdS)NrFcuda)rHrIinflight_threshholdin_flight_data collectionsdeque current_itemsidxstarted device_typetorchdevicetyper device_moduleZcurrent_streamstreamZ wait_stream)r?rHr_rSr*r*r+rJzs  z_OverlappingCpuLoader.__init__cCs|jt|jkSr0)rXlenrIrDr*r*r+_donesz_OverlappingCpuLoader._donecCslg}|j|jkr |j|j|jkr4|j}|j|d|d8_|||j|jks|SNr) rTrSr_ synchronizerWpopleftr4 element_sizerK)r?drainedvalr*r*r+_drains    "  z_OverlappingCpuLoader._draincCs(|j|j|jsr|j|jkrz|j|j\}}|jd7_||}|j j |j kr6|j ddd}n|j t dkrL||krL|}|j||f|j||7_|js|j|jksWddSWddSWddSWddS1swYdS)Nr r2T)r\Z non_blocking)r^r_rarTrSrIrXrHr1r\r]rZtor[rMr@r4r5rWrKrerNr*r*r+_refills8 "z_OverlappingCpuLoader._refillcCs(|jsJt|jdkr|j|jSrb)rar`rWr_rcrDr*r*r+_finishs  z_OverlappingCpuLoader._finishcCs"|jrtd|j||fdS)Nz&cannot add items after loading started)rY RuntimeErrorrIrKr>r*r*r+rBsz_OverlappingCpuLoader.addcCs.|jrdSd|_|jjddd|dS)NTcS|dSrbr*xr*r*r+z5_OverlappingCpuLoader.start_loading..key)rYrIsortrjrDr*r*r+rEs  z#_OverlappingCpuLoader.start_loadingccsB||js|}||EdH|jr|EdHdSr0)rErarhrjrk)r?rfr*r*r+rFs z_OverlappingCpuLoader.values)NrQ) r#r$r%rJpropertyrarhrjrkrBrErFr*r*r*r+rPys    rPcCsBd}|jdus J|jjD]}||9}q |jjj}|tj|SNr )Z tensor_datar@ propertiesdtyper[_utilsZ _element_size)r7r@srxr*r*r+ _item_sizes    r{rIc Cs|dkr|gSdd|D}dd|D}ddt|D}ddt|D}|jtddt|D] \}}||||q2|D]}tt|d d d d }|||||t|7<qB|S) Nr cSg|] }|jtjkr|qSr*r]rBYTE_IO.0wir*r*r+ z+_split_by_size_and_type..cSg|] }|jtjkr|qSr*r}rr*r*r+rrcSsg|]}gqSr*r*rrOr*r*r+rcSsg|]}dqS)rr*rr*r*r+rrT)rsreversecSrmrvr*rnr*r*r+rprqz)_split_by_size_and_type..rrr)rangertr{ enumeraterKmin) ZbinsrIbytes_wtensor_wZbucketsZ bucket_sizesirrXr*r*r+_split_by_size_and_typesrcCs|}|jtjkrt|tjsJ||nt|t j s"J|j t dks,Jt ||||}t ||t|||S)Nr2)tellr]rr~ isinstanceioBytesIOwrite getbufferr[r r\saver;r)r_data write_item storage_keyr!r"r*r*r+ _write_items   r file_queue result_queueplannerrS use_fsyncc sBz |\}}}tjr|dkrtfdd|d}ntfdd}dd|D} | D] } |t| | q.|d d|D} g} t |d =} | D]} | }| t | || |qN| D]\}} |jsmJ| t | || |qd|rt| Wdn1swY|| qtjyYdSw) NTrc |Sr0 resolve_datarnrr*r+rp z)_write_files_from_queue..)rScrr0rrnrr*r+rprcSrr*r}rr*r*r+rz+_write_files_from_queue..cSr|r*r}rr*r*r+rrwb) get_nowaitr[rRZ is_availablerPrGrBr{rEopenrrKrrFZis_cpuosfsyncfilenoputqueueEmpty)rrrrSr file_namerZ write_itemsloaderrrrZ write_resultsr_rr.r*rr+_write_files_from_queuesR        *rc seZdZdZ    ddeeejfdedede d e d d f fd d Z ded d fddZ de d e fddZ dee d ee fddZde ded eeefddZdedeeed d fddZZS)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 逖pathsingle_file_per_rank sync_files thread_countper_thread_copy_aheadr/Ncs0tt||_||_||_||_||_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. N. B. If sync_files is disabled, there's no guarantee that the checkpoint will be consistent in the case of a failure. N)superrJr rrrrr)r?rrrrr __class__r*r+rJDs   zFileSystemWriter.__init__is_coordinatorcCr=r0r*)r?rr*r*r+set_up_storage_writer_rLz&FileSystemWriter.set_up_storage_writerplancCs|jjddd|S)NT)parentsexist_ok)rmkdirr?rr*r*r+prepare_local_planbsz#FileSystemWriter.prepare_local_plan global_plancCsddt|D}|S)NcSs*g|]\}}tj|td|ddqS)__rOr:) dataclassesreplacer,)rrrr*r*r+risz8FileSystemWriter.prepare_global_plan..)r)r?rZ new_plansr*r*r+prepare_global_planfsz$FileSystemWriter.prepare_global_planrcs6|jdfdd}t}|jr,t|j|jD]}|}||j|||fqn|jD]}|}||j|||gfq/t}g} t d|jD]} t j t ||||j |jfd} | | | qMt ||||j |jd| D]} | qtg} z | |7} qtjy t} | | | YSw)Nrcsjt}d7|Srv)r-DEFAULT_SUFFIX)rZ file_countZ storage_planr*r+gen_filewsz-FileSystemWriter.write_data..gen_filer )targetargs)rrrrSr)r:rQueuerrrrIrrr threadingThreadrrrstartrKjoinrrr set_result)r?rrrrZbucketrr7rthreadsrOtresfutr*rr+ write_dataos`     zFileSystemWriter.write_datametadataresultscCst}|D] }|dd|Dq||_|jdd}t||t| Wdn1s5wY|jd |jddS)NcSsi|]}|j|jqSr*)r8r:)rwrr*r*r+ sz+FileSystemWriter.finish..z .metadata.tmpr .metadata) dictupdater:rrpickledumprrrrename)r?rrZ storage_mdZwr_list metadata_filer*r*r+finishs zFileSystemWriter.finish)TTr r)r#r$r%r&rr'rPathLikeboolr)rJrrrrrrr rrr r __classcell__r*r*rr+r6sP     A cseZdZdeeejfddffdd ZdefddZ d e d e de dfd d Z defd dZdededdfddZd e de fddZdee dee fddZZS)rrr/Ncs tt||_t|_dSr0)rrJr rrr:)r?rrr*r+rJs   zFileSystemReader.__init__sinfocCst||j|jSr0)rr!r")r?filerr*r*r+ _slice_fileszFileSystemReader._slice_filerrc Cs`t}|jD]}|j|j}|j}||g|q|D]\}}|j|dr} |D]g} |j| j}| | |} | j t j krWt | |j} | d|| | q.tttj| dd} t| | j| j} || }|| ksJd| jd|d| || || |q.Wdn1swYqt}| d|S)Nrbrr2)Z map_locationzreq z mismatch sizes z vs )!rrIr:Z storage_indexr setdefaultrKrrrr]rr~rrreadr"seekZ load_bytesrr r[loadrZstorage_offsetslengthsZresolve_tensorr1r@Zcopy_Z commit_tensorr r)r?rrZper_fileZ read_itemZitem_mdrr reqsrreqZ file_slicebytesr.Z target_tensorrr*r*r+ read_datas@         zFileSystemReader.read_datacCs>|jdd }t|WdS1swYdS)Nrr)rrrr)r?rr*r*r+ read_metadatas$zFileSystemReader.read_metadatarrcCs|j|_|jdus JdSr0r)r?rrr*r*r+set_up_storage_readersz&FileSystemReader.set_up_storage_readercC|Sr0r*rr*r*r+rrLz#FileSystemReader.prepare_local_planrcCrr0r*)r?rr*r*r+rsz$FileSystemReader.prepare_global_plan)r#r$r%rr'rrrJrrrrr rr rrrrrrrr*r*rr+rs &)>abcrrrrrUrrrrrtypingrrrrr[r Z torch.futuresr pathlibr rr rrMrrrrrrrrrrrrutilsrZtorch.distributed._shard._utilsrZ torch._utilsr__all__rr,rr6r;r<rGrPr)r{rrrrrrrr*r*r*r+st    (    Q    6