o }oi2@sddlZddlZddlmZddlmZddlmZddlm Z ddl m Z ddl m Z mZmZmZmZddlZddlmZdd lmZdd lmZmZmZmZmZGd d d eZd eddfddZddZ ddZ!ddZ"dS)N)ProcessPoolExecutor)BytesIO)get_start_method)Path)NamedTemporaryFile)AnyCallableDictOptionalUnion) CheckpointIO)logging)DEFAULT_CHUNK_SIZE_MBDEFAULT_MAX_READ_CONCURRENCYDEFAULT_MAX_WRITE_CONCURRENCYSHARED_MEM_DIRS3Utilsc seZdZdZeeedfdeffdd Ze ddZ de d ed efd d Z de d ed e fd dZddZ dde eefd eeefdeed dfddZddfd eeefdeed e eeffddZd eeefd dffdd ZdddZZS) S3CheckpointIOzsA custom S3CheckpointIO module that supports checkpoint reading/writing with s3 when filepath is a s3 url. Fdirpathc st|s td|d||_||_||_||_ g|_|jrk|jr(t ddnd|_ t dkr;t j dur;tdtd |j t}z |td Wntyg}z td ||d}~wwg|_tdS) a Initialize the transfer configuration with custom values. This method overrides the default TransferConfig values in boto3. See https://boto3.amazonaws.com/v1/documentation/api/latest/_modules/boto3/s3/transfer.html#TransferConfig Args: chunk_size_MB (int, optional): The size of chunks to use when transferring files. Default is 64 (MB). max_read_concurrency (int, optional): The maximum number of threads that will be making requests to perform a download. Default is 15. max_write_concurrency (int, optional): The maximum number of threads that will be making requests to perform an upload. Default is 10. async_checkpointing (bool, optional): Uses a ProcessPoolExecutor to do the main saving logic. This feature should be used with save_top_k as it's possible a previous checkpoint is removed while the current checkpoint write fails. z6Error attempting to initialize an S3CheckpointIO when zL is not an S3 url. Please use TorchCheckpointIO when using a non-S3 dirpath.) max_workersNforkTz\torch.cuda should not be initialized when checkpointing subprocess is created by fork methodz.Creating asynchronous checkpointing subprocessz9Asynchronous heckpointing subprocess created successfullyzCFailed to create asynchronous checkpointing subprocess, exception: )r is_s3_urlAssertionError chunk_size_MBmax_read_concurrencymax_write_concurrency_async_checkpointing _temp_filesasync_checkpointingr _executorrtorchcudais_initialized Exceptionr infosubmit dummy_funcresulterror_futuressuper__init__)selfrrrrrfuturee __class__Y/home/ubuntu/.local/lib/python3.10/site-packages/nemo/utils/callbacks/s3_checkpoint_io.pyr,*s8    zS3CheckpointIO.__init__cCs|jSN)r)r-r2r2r3rgsz"S3CheckpointIO.async_checkpointing checkpointpathreturnc Cs\t}ttdd}t||td|jd|dt|ddtj ~|jS)zW Returns: filename of the temporary file in shared memory. Fdirdeletez'Time elapsed saving checkpoint dict to z for : .2f seconds, rank ) time perf_counterrrr!saver r%name distributedget_rank)r-r5r6 start_timetempfiler2r2r3_serialize_checkpoint_to_shmks  ,z+S3CheckpointIO._serialize_checkpoint_to_shmc CsPt}t}t||t|}td|d|ddtj~|S)zH Returns: The bytestring of the checkpoint. z1Time elapsed saving checkpoint dict to bytes for r;r<r=) r>r?rr!r@r r%rBrC)r-r5r6ssbytesttr2r2r3_serialize_checkpoint_to_bytesys  z-S3CheckpointIO._serialize_checkpoint_to_bytesc Cs|jsdSt}g}g}|jD]}|dr||q||q|D]:}z|dWn tyP}ztd|dd|dd||d}~wwz |j |dWq&Yq&||_t dt|d d dS) aV self._future is a list of tuples of form (future, destination path, source path) This function checks the result of all the futures, and updates the self._futures list appropriately. It also updates the list of self._temp_files, which is used to clean up leaked temporary files in SHARED_MEM during teardown. NrzFailed to upload z to rz , exception: z0Time elapsed checking uploading future results: r<z seconds) r*r>r?doneappendr(r$r r)rremovedebug)r-rD done_futuresin_progress_futuresitemr/r2r2r3_check_uploading_results_so_fars2    $z.S3CheckpointIO._check_uploading_results_so_farNstorage_optionscCs tjtr|||}|j|d}n|||}d}|jrc| t d|dt j |rL|jt|||j|jd}|j|||fdS|jt|||j|j}|j||dfdSt d|dt j |rt|||j|jd|j|dSt|||j|jdS)NTFzUploading checkpoint to z in asynchronous mode, rank rHz in synchronous mode, rank )osr6existsrrFrrMrJrrSr r%r!rBrCr r&_upload_file_to_s3rrr*_upload_bytes_to_s3rN)r-r5r6rT localfile saved_as_filerHr.r2r2r3save_checkpoints.    zS3CheckpointIO.save_checkpointcCs|Sr4r2)storagelocr2r2r3szS3CheckpointIO. map_locationc CstjtrCttdd-}td|d|jdtj t j ||j|j |jdt|j}Wd|S1sr?r shutdownrrUr6rVrNr$)r-rankrDtfiler/r2r2r3teardowns&    "zS3CheckpointIO.teardownr4)r7N)__name__ __module__ __qualname____doc__rrrstrr,propertyrr rFrrJrSrr rr r[rrgrirn __classcell__r2r2r0r3r%sD= "   "   rfilepathr7cCsft|r/t|}td|tj|ddd}|D]}td|d|t|qdSdS)a before saving to s3, clean up any existing object with the same prefix megatron_gpt+step_count e.g. before we save "megatron_gpt--step=1400-validation_loss=6.32-consumed_samples=55920.0-last.ckpt" we need to clean up "megatron_gpt--step=1400-validation_loss=xxx-consumed_samples=yyy-last.ckpt" so that in case later we need to resume from step 1400, it has a single checkpoint file at step 1400 z0Looking for conflicting checkpoint under prefix z last.ckptF) base_pathsuffixreturn_key_onlyz"Cleaning up conflicting last ckpt z before saving N)rrparse_prefix_with_stepr r%find_files_with_suffixrh)rvprefix_with_stepconflict_last_ckpts last_ckptr2r2r3 _clean_up_conflicting_checkpoints   rc Cs>zt|t|||||WdSty}z|d}~wwr4)rr upload_filer$)rYr6rr remove_filer/r2r2r3rWsrWc Cs<zt|t||||WdSty}z|d}~wwr4)rrupload_file_stream_to_s3r$)rHr6rrr/r2r2r3rXsrXcCstddS)Ng{Gz?)r>sleepr2r2r2r3r' sr')#rUr>concurrent.futuresriormultiprocessingrpathlibrrErtypingrrr r r r!)lightning.fabric.plugins.io.checkpoint_ior nemo.utilsr nemo.utils.s3_utilsrrrrrrrsrrWrXr'r2r2r2r3s$        X