o viJ@sddlZddlZddlZddlZddlZddlmZddlZddl m Z ddl m Z ddl mZmZz ddlmZdZWneyKdZedYnwdd lmZmZdd lmZdd lmZmZmZmZmZdd lm Z dd l!m"Z"GdddZ#dS)N)INTERNAL)distributed_init)maybe_enable_memory_snapshotmaybe_enable_profiling)parallel_stateTFzMegatron-core is not installed.) LazyConfig instantiate)ImaginaireModel)callback distributedemalogmisc) Checkpointer)StragglerDetectorV2cseZdZdZfddZdedejjj dejjj ddfd d Z dd ej j e jBd ejjdejjjdejjdeeejfdededeeeejfejeffddZeddedejjj deddfddZZS)ImaginaireTraineraThe base trainer class of Imaginaire. All trainers in Imaginaire should inherit ImaginaireTrainer. It contains the basic functionality for model training (particularly suited for large-scale training), including data parallel (DDP/FSDP), model weight average (EMA), mixed-precision training (fp16/bf16). Attributes: checkpointer (Checkpointer): checkpointer object to save/load model weights and optimizer states. training_timer (misc.Timer): Timer object to time code blocks and functions. cst||_t]tt|jdr+|jj dkr t dt d|jj |j_ t radttjjvrFtj|jj|jj|jj ddntj|jj|jj|jj d|jjt_tjrad tjd <Wd n1skwYtrtj|jjd d t||jjdt||jjdt t!rt "|jjdtrt #d|j$d dt%&gdnt%&ddgt%j'|j(j)d d|j(j*j+t,j-j*_+|j(j*j.t,j-j*_.d t,j-j*_/t,j-j0j1_/t2j3||d|_4|j5j6d urt7|j5|j|j4d|_8nt9|j5j6|j5|j|j4d|_8t%:|_;t<|jj(j=j>|jj(j=j?|jj(j=j@|jj(j=jA|jj(j=jBd|_C|jCDtEEtEjFtGHt%jI|j(jJd S)zConstructor of the trainer. Args: config (Config): The config object for the Imaginaire codebase. context_parallel_sizezBoth config.model.context_parallel_size and config.model_parallel.context_parallel_size are set. config.model.context_parallel_size is deprecated. Please only set config.model_parallel.context_parallel_size.ztUsing deprecated config.model.context_parallel_size. Please use config.model_parallel.context_parallel_size instead.create_gloo_process_groupsF)pipeline_model_parallel_sizetensor_model_parallel_sizerr)rrr1CUDA_DEVICE_MAX_CONNECTIONSNT)exist_okz /config.pklz /config.yamlz /stdout.logzConfig: ) use_color) TORCH_HOMEIMAGINAIRE_OUTPUT_ROOTZENABLE_ONELOGGERHF_HOMEr)seedby_rank)configtrainer) callbacks)enabled report_freq profile_freqmax_diff raise_error)Ksuper__init__r rr inithasattrmodelmodel_parallelr ValueErrorr critical USE_MEGATRONinspect signaturerinitialize_model_parallel parametersrrsequence_parallelosenvironis_rank0makedirsjob path_localrsave_pkl save_yamldistbarrierrinit_loguru_fileinfo pretty_printrprint_environ_variablesset_random_seedr!rcudnn deterministictorchbackends benchmark allow_tf32cudamatmulr CallBackGroupr" checkpointtyper checkpointerr TrainingTimertraining_timerrstraggler_detectionr#r$r%r&r'straggler_detector initializesignalSIGALRM functoolspartialtimeout_handlertimeout_period)selfr  __class__K/data/cameron/vidgen/cosmos-policy/cosmos_policy/_src/imaginaire/trainer.pyr)9s       &         zImaginaireTrainer.__init__r,dataloader_traindataloader_valreturnNc Cs.|jd|jjjd}||jjj|j||jj|jj \}}t j j di|jjj }|j|j||||}d}td|jjj|jjjdkr[t|jjj|} n|jjjdkre|} n td|jjjtd|jj||d |jjjr|dkr|jjjr|j|||d d } t|j|d @} t|j|d '} t|} |j|zQz8| d )|j!j"d|jjj#j$d d t%| }Wdn1swYWdn1swYWnt&yYW|j'|nwW|j'|n|j'|w||jjj(krd } nt)j|dd}|jj*|||d |jj+|||d |j,s7| -| j,s?Jd|j,sGJd|j.| ||||||d\}}}|jj/|||||d |dkrgq|d7}||jj0j1dkr|jj2|||||d |jj3|||||d |jjjr||jjj4dkr|j|||d t56|jjj7|j!8|| r| 9| r| 9q| rnqWdn 1swYWdn 1swYt:d||jj0j1dkr|jj2|||||d |jj;||d |j<t=|j>dS)aThe training function. Args: model (ImaginaireModel): The PyTorch model. dataloader_train (torch.utils.data.DataLoader): The training data loader. dataloader_val (torch.utils.data.DataLoader): The validation data loader. rK) memory_formatrzDistributed parallelism mode: ddpfsdpz&Unknown distributed parallelism mode: zStarting training... iterationF) global_stepTra dataloading) profile_cudaNdevicez"model_ddp is not in training mode.zmodel is not in training mode.)rhgrad_accum_iterrzDone with training.)rK)?tor r!rdon_train_startr"on_optimizer_init_startZinit_optimizer_scheduler optimizer schedulerrGamp GradScalergrad_scaler_argson_optimizer_init_endrPloadr r/distributed_parallelismr parallel_model_wrapperrer.rArun_validationrun_validation_on_startvalidaterriteron_before_dataloadingrRrTprofile_sectionrSanalyze_dataloadingnext StopIterationon_after_dataloadingmax_iterron_training_step_starton_training_step_batch_starttrainingtrain training_stepon_training_step_batch_endrN save_itersaveon_training_step_endvalidation_iterrValarmr[generate_reportstepsuccess on_train_endfinalizer? on_app_end)r\r,rarbrrrs grad_scalerrhrn model_ddpZ _end_trainingtorch_profilerZmemory_profilerZdataloader_train_iter data_batch output_batchlossr_r_r`rs              <? D zImaginaireTrainer.trainrrrrrsrdatarhrnc Cst|||jjjdk|jj|d|d+|j d|jjj j | ||\}} Wdn1s9wYWdn1sHwY|jj |d|jj|| |d|dB|j d|jjj j'|| |jjj} | |jjjdkr|jn|Wdn1swYWdn1swY|jj||dWdn1swY|d7}||jjjkrE|d f|j d |jjj jJ|jj|||||d|||||jj||||d|jjjdkr|jj|||dn|j|||d|jd d Wdn 1s.wYWdn 1s>wYd }|| |fS)aThe training step. Args: model_ddp (torch.nn.Module | distributed.DistributedDataParallel): The model with a DDP wrapper or, the bare module, depending on whether distributed training is enabled or not. 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). data (dict[str, torch.Tensor]): Data batch (dictionary of tensors). iteration (int): Current iteration number. grad_accum_iter (int): Number of gradient accumulation iterations. Returns: output (dict[str, torch.Tensor]): The model output from the training data batch (dictionary of tensors). loss (torch.Tensor): The total loss of the training data batch. rrgforwardfwdNbackwardbwdreoptimizer_stepoptT) set_to_noner)r ddp_sync_gradr r!rnr"on_before_forwardrRrTrrSanalyze_forwardron_after_forwardon_before_backwardanalyze_backwardscalerrymoduleon_after_backwardanalyze_optimizeron_before_optimizer_steprupdateon_before_zero_grad zero_grad) r\rrrrsrrrhrnrrZ loss_scaledr_r_r`r sb           zImaginaireTrainer.training_stepc Cs|jj|||d|tj||jjjdDt|D]7\}}|jjj dur0||jjj kr0n$t j |dd}|jj |||d| ||\}}|jj|||||dqWdn1s^wY|jj||ddS)aValidate on the full validation dataset. Args: model (ImaginaireModel): The PyTorch model. dataloader_val (torch.utils.data.DataLoader): The validation data loader. iteration (int): Current iteration number. rg)r#NrKrl)r"on_validation_startevalr Z ema_scoper r# enumerater! max_val_iterrroon_validation_step_startZvalidation_stepon_validation_step_endon_validation_end)r\r,rbrhZval_iterrrrr_r_r`r}Ms zImaginaireTrainer.validate)rr)r)__name__ __module__ __qualname____doc__r)r rGutilsr DataLoaderrnnModuler DistributedDataParalleloptim Optimizer lr_scheduler LRSchedulerrtrudictstrTensorinttuplerno_gradr} __classcell__r_r_r]r`r-sB \ {   D,r)$rXr1r6rVrGtorch.distributedr r>torch.utils.data#cosmos_policy._src.imaginaire.flagsrZ4cosmos_policy._src.imaginaire.utils.context_managersrZ-cosmos_policy._src.imaginaire.utils.profilingrr megatron.corerr0 ImportErrorprint)cosmos_policy._src.imaginaire.lazy_configrr#cosmos_policy._src.imaginaire.modelr #cosmos_policy._src.imaginaire.utilsr r r rZ0cosmos_policy._src.imaginaire.utils.checkpointerr(cosmos_policy._src.imaginaire.utils.miscrrr_r_r_r`s.