o ?߱i` @sdZddlZddlZddlZddlZddlZddlZddlZddlm Z ddlm Z ddl m Z m Z mZmZmZmZmZddlZddlZddlmmZddlmZddlmZmZddlmZdd lmZm Z m!Z!m"Z"m#Z#dd l$m%Z%dd l&m'Z'dd l(m)Z)m*Z*dd l+m,Z,m-Z-ddl.m/Z/ddl0m1Z1mZm2Z2m3Z3ddl4m5Z5z=ddlm6Z7ddlm8Z8m9Z9m:Z:m;Z;mZ>m?Z?d3de@eAe fde>deBde9fddZCGddde7Z6e2DdWn!eEyZFzddlm6Z6e2DeFdWYdZF[FndZF[Fwwe ddd gZGGd!d"d"e%ZHGd#d$d$e%ZIGd%d&d&eAejJZKGd'd(d(ZLGd)d*d*ZMd+ejNd,ejNd-e,d.e-ddf d/d0ZOGd1d2d2e'ZPdS)4uu Distributed checkpoint (DCP) directory structure and storage backends. The checkpointer saves model state in a sharded format across multiple processes: self.save_dirname/ ├── iter_000000005/ # Checkpoint at iteration 5 │ ├── model/ # Model state shards │ │ ├── __0_0.distcp # Shard 0 from rank 0 │ │ └── __1_0.distcp # Shard 1 from rank 1 │ ├── optim/ # Optimizer state shards │ │ ├── __0_0.distcp # Shard 0 from rank 0 │ │ └── __1_0.distcp # Shard 1 from rank 1 │ ├── scheduler/ # Learning rate scheduler state │ │ ├── __0_0.distcp # Shard 0 from rank 0 │ │ └── __1_0.distcp # Shard 1 from rank 1 │ └── trainer/ # Additional training state │ ├── __0_0.distcp # Shard 0 from rank 0 │ └── __1_0.distcp # Shard 1 from rank 1 └── latest_checkpoint.txt # Points to most recent checkpoint folder, e.g. iter_000000005 Storage backends: - Local filesystem: self.save_dirname = "{config_job.path_local}/checkpoints" - S3 object store: self.save_dirname = "s3://{bucket}/{config_job.path}/checkpoints" where bucket = self.config_checkpoint.save_to_object_store.bucket The sharded format enables efficient distributed saving/loading by: 1. Parallelizing I/O across processes 2. Reducing memory usage per process 3. Supporting both local and cloud storage backends N) namedtuple) get_context)AnyDictListOptionalSetTupleUnion)nn)FileSystemReaderFileSystemWriter)DefaultSavePlanner)StateDictOptionsget_model_state_dictget_optimizer_state_dictset_model_state_dictset_optimizer_state_dict)Stateful)AbstractCheckpointer)S3StorageReaderS3StorageWriter)CheckpointConfig JobConfig)ImaginaireModel)callback distributedlogmisc)easy_io)DefaultLoadPlanner)DTensorLoadPlan_create_read_items_versionflatten_state_dict)MetadataTensorStorageMetadataT state_dictmetadatastrictreturnc Csg} |D]t\}}|drq||jvr!|r td|dq|j|}t|tr^t|dddur^|j|kr^|sNt d|jd|d|qt d|jd|d|t|t rs|j durr|t|||7}q|t|||7}qt|S)Nz ._extra_statez&Missing key in checkpoint state_dict: .sizezSize mismatch between saved z and current: z for )itemsendswithstate_dict_metadata RuntimeError isinstancer'getattrr-rcritical ValueErrorr! device_meshget_coordinater#r")r(r)r*requestsfqnobjmdr<[/data/cameron/vidgen/cosmos-predict2.5/cosmos_predict2/_src/interactive/checkpointer/dcp.pycreate_default_local_load_plan_s4   " r>c@seZdZdefddZdS)r r+cCs|jdusJ|jr=t|j}t|jj}||}|r=dt_t|j\}}t|}||@r:|||_|_ dt_t |j|j|j S)N2_3) r)r%setr(keysr0r$_derived_versionoriginal_state_dictmappingsr>allow_partial_load)self current_keys load_keys missing_keysold_state_dict old_mappingsold_keysr<r<r=create_local_plans" z$DefaultLoadPlanner.create_local_planN)__name__ __module__ __qualname__r"rMr<r<r<r=r sr zIfor the back comptiable pytorch! New DefaultLoadPlanner class is created.z, using default plannerStateDictItemPath save_pathc@sdeZdZdZddeejeejfdefddZ de e e ffdd Z d e e e fdd fd d Zd S) ModelWrapperz%Wrapper for model state dict handlingFmodelload_ema_to_regcCst|tjr |gn||_||_|jrGddlm}ddlm}ddl m }ddl m }t||sIt||sKt||sMt||sOJdt |dSdSdSdSdS)Nr)Cosmos2InteractiveModel)DiffusionModel)Text2WorldModelRectifiedFlow)WANDiffusionModelzPModelWrapper only supports DiffusionModel when load_ema_to_reg is True, but got )r2r ModulerTrUZBcosmos_predict2._src.interactive.methods.cosmos2_interactive_modelrVZ5cosmos_predict2._src.predict2.models.text2world_modelrWZDcosmos_predict2._src.predict2.models.text2world_model_rectified_flowrXZ=cosmos_predict2._src.predict2.models.text2world_wan2pt1_modelrYtype)rFrTrUZcosmos2_interactive_modelZpredict2_DiffusionModelZ&predict2_DiffusionModel_rectified_flowZwan2pt1_DiffusionModelr<r<r=__init__s&      zModelWrapper.__init__r+cCsddtt|jD}|jr=|jdjjjrJdt|}t dd|Ds-Jd|D] }| ||| dd <q/t |jdjd r||jdjj r| i|_g}|D]}d |vrk||||j| d d <qW|D] }| ||| d d <qn|S) NcS$i|]}|D]\}}||qqSr<r..0sdkvr<r<r= $z+ModelWrapper.state_dict..rAEMA is enabled, can not load EMA weights to regular model weightscs.|]}|dp|dp|dVqdS)net.net_fake_score.net_discriminator_head.N startswithr`rbr<r<r=  z*ModelWrapper.state_dict..zAll keys must start with net.rhnet_ema.use_loraz base_layer.)maprrTrUconfigemaenabledlistrAallpopreplacehasattrrqcheckpoint_to_model_keyappend)rF _state_dictall_keysrbZkeys_to_updater<r<r=r(s2     zModelWrapper.state_dictr(NcCs"|jr|t|jdjdr.|jdjjr.t|dr*|jD] \}}||||<qntd|jdjj j r:Jdt | }t dd|DsMJd|D],}|d rb||||d d <qO|d ro||||<qO|d r{||||<qOtjt|td dd}t t||jdS)Nrrqr|z2checkpoint_to_model_key is not set by `state_dict`rfcsrg)rprirjNrkrmr<r<r=rn roz/ModelWrapper.load_state_dict..z!All keys must start with net_ema.rprhrirjF)r*)model_state_dictoptions)rUr{rTrtrqr|r.ryr5rurvrwrArxrlrz functoolspartialrrrs)rFr(Zcheckpoint_key model_keyrrbfuncr<r<r=load_state_dicts<       zModelWrapper.load_state_dict)F)rNrOrP__doc__r r rZrboolr\rstrrr(rr<r<r<r=rSs $"rSc@sveZdZdeejeejfdeejj eejj fddfddZ de e e ffddZd e e e fddfd d ZdS) OptimizerWrapperrToptimr+NcCs<t|tjr |gn||_t|tjjr|g|_dS||_dSN)r2r rZrTtorchr Optimizer)rFrTrr<r<r=r\ s$zOptimizerWrapper.__init__cCs.tjttddd}ddt||j|jDS)NTZflatten_optimizer_state_dict)rcSr]r<r^r_r<r<r=rd-rez/OptimizerWrapper.state_dict..)rrrrrsrTr)rFrr<r<r=r((s zOptimizerWrapper.state_dictr(cCs.tjt|tddd}tt||j|jdS)NTr)optim_state_dictr)rrrrrwrsrTr)rFr(rr<r<r=r/s z OptimizerWrapper.load_state_dict)rNrOrPr r rZrrrrr\rrrr(rr<r<r<r=rs rc@seZdZdZdZdS) AsyncModedisabledZasync_with_pinned_memN)rNrOrPDISABLEDASYNC_WITH_PINNED_MEMr<r<r<r=r8src@s eZdZdS) TerminateN)rNrOrPr<r<r<r=r=src@s*eZdZdededefddZddZdS) SaveDone iteration elapsed_time succeededcCs||_||_||_dSrrrr)rFrrrr<r<r=r\Bs zSaveDone.__init__cCsd|jd|jd|jdS)NzSaveDone(iteration=z, elapsed_time=z , succeeded=)rrFr<r<r=__str__GszSaveDone.__str__N)rNrOrPintfloatrr\rr<r<r<r=rAs rreceiver_queue sender_queuecheckpoint_config job_configc Cstttjddtjd<dtjd<tjttjdtt ||ddd}z t d | }t d t |tr[t d |t|Wt d tjdSt |tsdJd |\}}t}|ddd} d} d} zOz|||t|} t d| dd| d} Wnty} zt d|d| WYd} ~ nd} ~ wwW| dkrt|} |t| | | n| dkrt|} |t| | | wq,t d tjw)a~ Handles model checkpoint saving in a separate background process using PyTorch's distributed functionality. This function runs in a dedicated process to avoid blocking the main training loop. Args: receiver_queue: Queue to receive state dictionaries and commands from the main process sender_queue: Queue to send completion signals back to the main process checkpoint_config: Configuration settings for checkpoint saving behavior job_config: Configuration settings for the training job Flow: 1. Initializes distributed processing environment 2. Continuously waits for state dictionaries to save 3. Saves checkpoints asynchronously 4. Signals completion back to main process 5. Terminates when receiving a Terminate signal Raises: AssertionError: If received object is neither Terminate signal nor valid state dict tuple Note: - Uses a different port than the main process to avoid conflicts - Disables TorchElastic agent store for checkpoint operations - Automatically cleans up distributed process group on exit MASTER_PORTFalseTORCHELASTIC_USE_AGENT_STORE LOCAL_RANKNT) disable_asynczPCheckpoint background process is ready for next task, waiting for new state_dictzReceived new state_dictzRReceived termination signal in checkpoint background process, closing sender queuez1Cleaning up: destroying distributed process groupz>Received data must be a tuple of (state_dict, checkpoint_path)trainerrrFzACheckpoint saved successfully in background process. Time taken: z.2fz seconds, iteration: zError saving checkpoint to z: )rrosenvironrcuda set_devicerinitDistributedCheckpointerrdebuggetr2rinfoputclosedestroy_process_grouptupletime monotonicsave_state_dict_worker Exceptionerrorr) rrrrZcheckpoint_handlerZ received_datar(checkpoint_path start_timerrrer<r<r=save_checkpoint_in_backgroundKsX         "   # rc seZdZgdZ  d+dededeejde ffdd Z d e e e edfffd d Zed    d,dedejjdBdejjjdBdejjdBd ef ddZdedeee eeffd dfddZd-ddZd.ded dfddZ ded e e!e"ffdd Z#ded e e$e%ffd!d"Z&d#eee eeffded dfd$d%Z'dedejjdejjjdejjd&ed df d'd(Z(d-fd)d* Z)Z*S)/rrTr schedulerrNFconfig_checkpoint config_job callbacksrcst|||||_|jrtj|_ntj|_|rtj|_|jtjkrVtd}| |_ | |_ |j t |j |j ||fdd|_|jd|_d|_d|_tj|_dSdS)NspawnT)targetargsdaemonF)superr\rdcp_async_mode_enabledrr async_moderrQueue mp_queue_send mp_queue_recvProcessrmpstartcpu_offload_state_dictstagingstaging_ckpt_filerrStreamstaging_stream)rFrrrrctx __class__r<r=r\s4     z DistributedCheckpointer.__init__r+csR}g}|durtjj|}|jndjr{j}jrcdj j j d|}t d|sc|}tj|d}tj|jdrVtj|jd}|d|}n td|d||}jrm|jn|d jrz|d nd}tjd krjD]}|jvsJd |d jqfdd|D}t||fS)Nzs3:///z/checkpoints/iter_\d{9}/?$z!checkpoints/latest_checkpoint.txt) backend_keyz /checkpoints/zLatest checkpoint file z not found, load from rTrrzInvalid key to resume: z not in csg|] }|jvr|qSr<)keys_not_to_resume)r`keyrr<r= szFDistributedCheckpointer.keys_to_resume_during_load..)_read_latest_checkpoint_filerpathjoinZ load_dirnameextend KEYS_TO_SAVE load_pathZload_s3_backend_keyrload_from_object_storebucketresearchrexistsloadstriprwarningload_training_stater}only_load_scheduler_statelenrr@)rFlatest_checkpoint_file resume_keysrZ old_ckpt_pathZlatest_ckpt_pathcheckpoint_filerr<rr=keys_to_resume_during_loadsF    " z2DistributedCheckpointer.keys_to_resume_during_loadzcheckpoint loadingrT optimizerr grad_scalercCs"|jdur |j||\}}t|}td|d|d}|dur|||D]}tdd} tj ||} td|| | } t j tjd| dd |d krvtd t|} | } tj| | | d | | q-|d krtdt||}|} tj| | | d || q-|dkrtd|} tj| | | d || q-|dkrtd||d} tj| | | d || d| d}q-td|d|jdur|jj|| dtd|d|ntdt j|jdur|jj|||d|S)NzResuming ckpt z with keys: rT)rEzStart loading checkpoint from z starting F rank0_onlyrTz- Loading the model...)storage_readerplannerrz- Loading the optimizer...rz- Loading the scheduler...rz- Loading the trainer...rrrrz Invalid key: z. not support to resume.r(zLoaded checkpoint from z in iteration zTraining from scratch.)rr)ron_load_checkpoint_startrsortedrr4_check_checkpoint_existsr rrrget_storage_readerrrbarrierrrSr(dcprrrr5on_load_checkpointr empty_cacheon_load_checkpoint_end)rFrTrrrrrrrZ load_plannercur_key_ckpt_full_pathr_model_wrapperr~Z_optim_wrapperr<r<r=rs                    zDistributedCheckpointer.loadrr(c Csz ddlm}m}Wnty}ztd|d}~ww|jdur4tdtd||ddd|_tdtdt j |j |||jdd |_d|_ ||_Wdn1s`wY|dS) Nr)_copy_state_dict_create_cpu_state_dictzXPlease install the latest PyTorch nightly to use async checkpointing with pinned memory.z+Preparing the CPU memory, time.monotonic()=z.:.2fT) pin_memory share_memoryz)Staging the state_dict, time.monotonic()=) non_blocking)#torch.distributed._state_dict_utilsrr  ImportErrorrrrrrrrstreamrrrmaybe_wait_for_staging)rFrr(rr rr<r<r=_async_with_pinned_memoryBs0  z1DistributedCheckpointer._async_with_pinned_memorycsJjtjkr!jr#jsjfdd}|d_dSdSdS)NcsjjjfdSr)r put_nowaitrrr<rr<r= sync_func^szADistributedCheckpointer.maybe_wait_for_staging..sync_funcF)rrrrrquery synchronize)rFrr<rr=rYs    z.DistributedCheckpointer.maybe_wait_for_stagingrwait_forc Cs|jtjkrzit}|jr|dkrmz4|jjdd}t|t r*t dWWdS|}t j d||j durG|jrG|j j|j|jdWntjy`t|}||kr^YWdSYnw|jr|dksWdSWdSttfyt dYdSwdS) zfGet the results of previously submitted checkpoints and pass them to callbacks if checkpoint succeededr)timeoutz=Received termination event from checkpoint background processz!Received checkpoint save result: Nrrz1Queue was closed by checkpoint background process)rrrrrremptyrr2rrrloggerrron_save_checkpoint_successrrqueueEmptyEOFErrorBrokenPipeError)rFrrretZ save_donerr<r<r=get_previous_checkpoint_resultsds6    z7DistributedCheckpointer.get_previous_checkpoint_resultsrcCs"|jr t|jjj|dSt|dS)Ncredential_pathr)r)save_to_object_storerr credentialsr rFrr<r<r=get_storage_writer|s  z*DistributedCheckpointer.get_storage_writercCs |jr t|jjj|dSt|S)Nr#)rrrr&r r'r<r<r=rs z*DistributedCheckpointer.get_storage_reader to_save_dictcCsz|D]\}\}}||}tj||tdddqtr+td|||t j dt j |j|dddS)NT)dedup_save_to_lowest_rank)storage_writerrzSaving last checkpoint file zSaved checkpoint to r)r.r(rsaverris_rank0print_write_latest_checkpoint_filerr4rrr save_dirname)rFr)rrbrcZfull_checkpoint_pathr+r<r<r=rs  $z.DistributedCheckpointer.save_state_dict_workerrc Cs^|jtjkr |jdd|jdur|j||d|d}t|t|||||dd}| D]}t j |j d|dd|} ||| f||<q9|jdur`|jj||d |jtjkrm|||n1t} z|||W|jdur|jj|t| d n|jdur|jj|t| d ww|jdur|jjd|d dSdS) aSave network weights, optimizer parameters, scheduler parameters to a checkpoint. Args: model (ImaginaireModel): The PyTorch model. optimizer (torch.optim.Optimizer): The model optimizer. scheduler (torch.optim.lr_scheduler.LRScheduler): The optimization scheduler. grad_scaler (torch.amp.GradScaler): The gradient scaler (for mixed precision training). iteration (int): Current iteration number. rrNiter_09rrrrr)rTr)rrrr"ron_save_checkpoint_startrSr(rrArrrr0on_save_checkpointrrrrron_save_checkpoint_end) rFrTrrrrrr)rbZoutput_dirnamerr<r<r=r,sF              zDistributedCheckpointer.savecsZt|jtjkr'|jr)|jr+|jt |j dd|j dSdSdSdS)N<r1) rfinalizerrrris_aliverrrr"rrrr<r=r8s   z DistributedCheckpointer.finalize)NF)NNN)r+N)r)+rNrOrPrrrrr CallBackGrouprr\r rr rrrtimerrrrr lr_scheduler LRScheduleramp GradScalerrrrrrrr"rr r(rr rrr,r8 __classcell__r<r<rr=rsb%,    &P  & 9r)T)Qrenumrmultiprocessingrrrr collectionsrrtypingrrrrrr r rtorch.distributedtorch.distributed.checkpointr checkpointrr r r ,torch.distributed.checkpoint.default_plannerr'torch.distributed.checkpoint.state_dictrrrrr%torch.distributed.checkpoint.statefulrZ1cosmos_predict2._src.imaginaire.checkpointer.baser:cosmos_predict2._src.imaginaire.checkpointer.s3_filesystemrr&cosmos_predict2._src.imaginaire.configrr%cosmos_predict2._src.imaginaire.modelr%cosmos_predict2._src.imaginaire.utilsrrr-cosmos_predict2._src.imaginaire.utils.easy_iorr Z_DefaultLoadPlannerr!r"r#r$r%%torch.distributed.checkpoint.metadatar&r'dictrrr>r4rrrQrSrEnumrrrrrrr<r<r<r=sn#  $       $.( \  R