o vi( @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? d7de@eAe fde>deBdeBde9f ddZCGddde7Z6e2DdWn!eEyZFzddlm6Z6e2DeFd WYdZF[FndZF[Fwwe d!dd"gZGd#d$ZHGd%d&d&e%ZIGd'd(d(e%ZJGd)d*d*eAejKZLGd+d,d,ZMGd-d.d.ZNd/ejOd0ejOd1e,d2e-ddf d3d4ZPGd5d6d6e'ZQdS)8uu 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)MetadataTensorStorageMetadataTF state_dictmetadatastrictdcp_allow_mismatched_sizereturnc Csg} |D]v\}}|drq||jvr!|r td|dq|j|}|s`t|tr`t|dddur`|j|kr`|sPt d|jd|d|qt d|jd|d|t|t ru|j durt|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*r+requestsfqnobjmdr=R/data/cameron/vidgen/cosmos-policy/cosmos_policy/_src/predict2/checkpointer/dcp.pycreate_default_local_load_plan_s6   " r?c@s(eZdZdefddZdefddZdS)r r+cCs ||_dSN)r+)selfr+r=r=r>set_partial_channel_weights z-DefaultLoadPlanner.set_partial_channel_weightr,cCs|jdusJ|jr=t|j}t|jj}||}|r=dt_t|j\}}t|}||@r:|||_|_ dt_t |j|j|j t |ddS)N2_3r+F) r)r%setr(keysr1r$_derived_versionoriginal_state_dictmappingsr?allow_partial_loadr4)rA current_keys load_keys missing_keysold_state_dict old_mappingsold_keysr=r=r>create_local_plans$  z$DefaultLoadPlanner.create_local_planN)__name__ __module__ __qualname__boolrBr"rPr=r=r=r>r sr zIfor the back comptiable pytorch! New DefaultLoadPlanner class is created.z, using default plannerStateDictItemPath save_pathc Cstj|||dg}t|dru|jdurwt|jj}t|}t||}t||}|rMdd|D}dt tdd|D}t d||ryd d|D}d d|D}dt td d|D}t d |dSdSdSdS) Nstorage_readerplannerr)cS g|] }d|vs d|vr|qSzblocks.0zblocks.r=.0keyr=r=r>  z'dcp_load_state_dict.. cs(|]}d|dddVqdSr-N joinsplitr]kr=r=r> &z&dcp_load_state_dict..z"Missing keys in pretrained model: cSsg|]}d|vr|qS) _extra_stater=r\r=r=r>r_scSrZr[r=r\r=r=r>r_r`csrbrcrerhr=r=r>rjrkz%Unexpected keys in pretrained model: ) dcploadhasattrr)rDr1rElistrfsortedrr5) _state_dictrX load_plannerrLcheckpoint_keys model_keysunexpected_keysZmissing_keys_strZunexpected_keys_strr=r=r>dcp_load_state_dicts,    rwc@steZdZdZddeejeejfdefddZ ifde e e fde e e ffd d Zd e e e fdd fd dZd S) ModelWrapperz%Wrapper for model state dict handlingFmodelload_ema_to_regcsttjr gn|_||_|jrIg}ddlm}||ddlm }||ddl m }||t fdd|DsKJdt dSdS)Nr)DiffusionModel)Text2WorldModelRectifiedFlow)WANDiffusionModelc3s|]}t|VqdSr@)r3)r]clsryr=r>rjz(ModelWrapper.__init__..zPModelWrapper only supports DiffusionModel when load_ema_to_reg is True, but got )r3r ModuleryrzZ3cosmos_policy._src.predict2.models.text2world_modelr{appendZBcosmos_policy._src.predict2.models.text2world_model_rectified_flowr|Z;cosmos_policy._src.predict2.models.text2world_wan2pt1_modelr}anytype)rAryrzZsupported_model_typesZpredict2_DiffusionModelZ&predict2_DiffusionModel_rectified_flowZwan2pt1_DiffusionModelr=rr>__init__s       zModelWrapper.__init__ mapping_keysr,c Cs&ddtt|jD}|jr=|jdjjjrJdt|}t dd|Ds-Jd|D] }| ||| dd <q/t |jdjd r|jdjj r i|_|d d d g}|D]#}|}|D] \}}| ||}qg||kr|||f||j|<q_|D] \}}| |||<q|S) NcS$i|]}|D]\}}||qqSr=r/r]sdrivr=r=r> $z+ModelWrapper.state_dict..rAEMA is enabled, can not load EMA weights to regular model weightscs|]}|dVqdS)net.N startswithrhr=r=r>rjrz*ModelWrapper.state_dict..zAll keys must start with net.rnet_ema.use_lora)z base_layer.zbase_model.model.)maprryrzconfigemaenabledrprEallpopreplacerorcheckpoint_to_model_keyupdater/r) rArrrall_keysriZkeys_to_updatenew_keyZfrom_keyZto_keyr=r=r>r(s:     zModelWrapper.state_dictr(NcCst|jdjdr+|jdjjr+t|dr'|jD] \}}||||<qntd|jr]|jdjj j r:Jdt | }t dd|DsMJd|D] }||||d d <qOtjt|td d d }t t||jdS)Nrrrz2checkpoint_to_model_key is not set by `state_dict`rcsr)rNrrhr=r=r>rj1rz/ModelWrapper.load_state_dict..z!All keys must start with net_ema.rrF)r*)model_state_dictoptions)roryrrrr/rr6rzrrrprErr functoolspartialrrr)rAr(checkpoint_key model_keyrrifuncr=r=r>load_state_dict%s(   zModelWrapper.load_state_dict)F)rQrRrS__doc__r r rrrTrdictstrrrr(rr=r=r=r>rxs $&'rxc@sveZdZdeejeejfdeejj eejj fddfddZ de e e ffddZd e e e fddfd d ZdS) OptimizerWrapperryoptimr,NcCs<t|tjr |gn||_t|tjjr|g|_dS||_dSr@)r3r rrytorchr Optimizer)rAryrr=r=r>r>s$zOptimizerWrapper.__init__cCs.tjttddd}ddt||j|jDS)NTZflatten_optimizer_state_dict)rcSrr=rrr=r=r>rKrz/OptimizerWrapper.state_dict..)rrrrrryr)rArr=r=r>r(Fs zOptimizerWrapper.state_dictr(cCs.tjt|tddd}tt||j|jdS)NTr)optim_state_dictr)rrrrrprryr)rAr(rr=r=r>rMs z OptimizerWrapper.load_state_dict)rQrRrSr r rrrrrrrrrr(rr=r=r=r>r=s rc@seZdZdZdZdS) AsyncModedisabledZasync_with_pinned_memN)rQrRrSDISABLEDASYNC_WITH_PINNED_MEMr=r=r=r>rVsrc@s eZdZdS) TerminateN)rQrRrSr=r=r=r>r[src@s*eZdZdededefddZddZdS) SaveDone iteration elapsed_time succeededcCs||_||_||_dSr@rrr)rArrrr=r=r>r`s zSaveDone.__init__cCsd|jd|jd|jdS)NzSaveDone(iteration=z, elapsed_time=z , succeeded=)rrAr=r=r>__str__eszSaveDone.__str__N)rQrRrSintfloatrTrrr=r=r=r>r_s 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_devicerinitDistributedCheckpointerrdebuggetr3rinfoputclosedestroy_process_grouptupletime monotonicsave_state_dict_worker Exceptionerrorr) rrrrZcheckpoint_handlerZ received_datar(checkpoint_path start_timerrrer=r=r>save_checkpoint_in_backgroundisX         "   # 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)/rryr schedulerrNFconfig_checkpoint config_job callbacksrcst|||||_|jrtj|_ntj|_|rtj|_|jtjkrVtd}| |_ | |_ |j t |j |j ||fdd|_|jd|_d|_d|_tj|_dSdS)NspawnT)targetargsdaemonF)superrrdcp_async_mode_enabledrr async_moderrQueue mp_queue_send mp_queue_recvProcessrmpstartcpu_offload_state_dictstagingstaging_ckpt_filerrStreamstaging_stream)rArrrrctx __class__r=r>rs4     z DistributedCheckpointer.__init__r,csb}g}|durtjj|}|jnljrtj dsj}j rkdj j j d|}td|sk|}tj|d}tj|j dr^tj|j d}|d|}n td|d ||}jru|jn|d jr|d nd}tjd krjD]}|jvsJd |djqfdd|D}t||fS)Nz.ptzs3:///z/checkpoints/iter_\d{9}/?$z!checkpoints/latest_checkpoint.txt) backend_keyz /checkpoints/zLatest checkpoint file z not found, load from ryrrzInvalid key to resume: z not in csg|] }|jvr|qSr=)keys_not_to_resumer\rr=r>r_ szFDistributedCheckpointer.keys_to_resume_during_load..)_read_latest_checkpoint_filerpathrfZ load_dirnameextend KEYS_TO_SAVE load_pathrr0Zload_s3_backend_keyrload_from_object_storebucketresearchrexistsrnstriprwarningload_training_stateronly_load_scheduler_statelenrrD)rAlatest_checkpoint_file resume_keysrZ old_ckpt_pathZlatest_ckpt_pathcheckpoint_filer^r=rr>keys_to_resume_during_loadsF    " z2DistributedCheckpointer.keys_to_resume_during_loadzcheckpoint loadingry optimizerr grad_scalercCsL|jdur |j||\}}t|}td|d|d}|dur |||D]}tdd} t| drKtd|j j | |j j t j ||} td||| } tjtjd | d d |d krtd t|} | } t| | | | | q.|dkrtdt||}|} tj| | | d|| q.|dkrtd|} tj| | | d|| q.|dkrtd||d} tj| | | d|| d| d}q.td|d|jdur|jj|| dtd|d|ntdtj|jdur$|jj|||d|S)NzResuming ckpt z with keys: rT)rIrBzset_partial_channel_weight: zStart loading checkpoint from z starting F rank0_onlyryz- Loading the model...rz- Loading the optimizer...rWrz- 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_startrrqrr5_check_checkpoint_existsr rorr+rBrrrfget_storage_readerrrbarrierrrxr(rwrrrmrnr6on_load_checkpointr empty_cacheon_load_checkpoint_end)rAryrrrrrrr^rscur_key_ckpt_full_pathrX_model_wrapperrrZ_optim_wrapperr=r=r>rns                       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_utilsr!r" ImportErrorrrrrrrrstreamrrrmaybe_wait_for_staging)rArr(r!r"rr=r=r>_async_with_pinned_memory`s0  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)rAr,r=rr>r)ws    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)rrrrrremptyrr3rrrloggerrron_save_checkpoint_successrrqueueEmptyEOFErrorBrokenPipeError)rAr/rretZ save_donerr=r=r>get_previous_checkpoint_resultss6    z7DistributedCheckpointer.get_previous_checkpoint_resultsrcCs"|jr t|jjj|dSt|dS)NZcredential_pathr)r)save_to_object_storerr credentialsr rArr=r=r>get_storage_writers  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_writerrYzSaving last checkpoint file zSaved checkpoint to r)r/r@rmsaverris_rank0print_write_latest_checkpoint_filerr5rrrf save_dirname)rArArrirZfull_checkpoint_pathrCr=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. rr/Niter_09rrrrr2)ryr)rrrr;ron_save_checkpoint_startrxr(rrErrrfrHon_save_checkpointr*rrrr5on_save_checkpoint_end) rAryrrrrrrAriZoutput_dirnamerr=r=r>rDsF              zDistributedCheckpointer.savecsZt|jtjkr'|jr)|jr+|jt |j dd|j dSdSdSdS)N<rI) rfinalizerrrris_aliverrrr;rfrrr=r>rPs   z DistributedCheckpointer.finalize)NF)NNN)r,N)r)+rQrRrSrrrrr CallBackGrouprTrr rr rrrtimerrrrr lr_scheduler LRScheduleramp GradScalerrrnrrr*r)r;rr r@rr rrrDrP __classcell__r=r=rr>rsb%,    &P  & 9r)TF)Rrenumrmultiprocessingrr6rr collectionsrrtypingrrrrrr r rtorch.distributedtorch.distributed.checkpointr checkpointrmr r r ,torch.distributed.checkpoint.default_plannerr'torch.distributed.checkpoint.state_dictrrrrr%torch.distributed.checkpoint.statefulrZ/cosmos_policy._src.imaginaire.checkpointer.baserZ8cosmos_policy._src.imaginaire.checkpointer.s3_filesystemrr$cosmos_policy._src.imaginaire.configrr#cosmos_policy._src.imaginaire.modelr#cosmos_policy._src.imaginaire.utilsrrr+cosmos_policy._src.imaginaire.utils.easy_iorr Z_DefaultLoadPlannerr!r"r#r$r%%torch.distributed.checkpoint.metadatar&r'rrrTr?r5r'rrUrwrxrEnumrrrrrrr=r=r=r>s#  $         1, Y  R