o
    4ѹg                     @   s  d dl Z d dlZd dlZd dlZd dlZd dlmZ d dlmZ d dlmZ	 d dl
mZmZ d dlmZmZ d dlmZ d dlmZ d dlmZ d d	lmZ d d
lmZ d dlmZ d dlmZ d dl m!  m"Z# zd dl$Z$W n   dZ$Y edddZ%G dd dZ&dS )    N)tqdm)datetime)autocast
GradScaler)nullcontextcontextmanager)Path)DistributedDataParallel)FullyShardedDataParallel)	to_device)recursive_average)average_checkpoints)ShardedGradScalerFc                 c   sn    |r#t jjjd| dd d V  W d    d S 1 sw   Y  d S | t jks-| t jkr2d V  d S d V  d S )NTF)enableddtypeZcache_enabled)torchcudaampr   float16bfloat16r   use_deepspeed r   a/Users/admin/.pyenv/versions/3.10.0/lib/python3.10/site-packages/funasr/train_utils/trainer_ds.pymaybe_autocast   s   "

r   c                   @   s   e Zd ZdZ									d'dededed	ed
edefddZ						d(ddZ				d)ddZ							d*ddZ	i fddZ
i fddZd+ddZ				d)ddZ		d,defdd Zd+d!d"Zd#d$ Zd%d& ZdS )-Trainera   
    A simple trainer class for training a PyTorch model, saving checkpoints at the end of each epoch,
    and optionally resuming from a saved checkpoint.

    Attributes:
        max_epoch (int): Maximum number of epochs for training.
        model (torch.nn.Module): The model to be trained.
        optim (torch.optim.Optimizer): The optimizer to use for training.
        scheduler (torch.optim.lr_scheduler._LRScheduler): The learning rate scheduler.
        dataloader_train (torch.utils.data.DataLoader): DataLoader for the training dataset.
        dataloader_val (torch.utils.data.DataLoader): DataLoader for the validation dataset.
        output_dir (str): Directory where model checkpoints will be saved.
        resume (str, optional): Path to a checkpoint to resume training from.
    r      F./use_ddpuse_fsdpuse_fp16use_bf16r   
output_dirc
              	   K   s  || _ || _|| _|| _|| _|
dd| _|	| _tj	
| js(tj| jdd |
dd| _d| _|
dd| _|
d	d
| _d| _tj| _|| _|| _| jrTtj| _| jr[tj| _|
dd| _|
dd| _|
dd| _|
dd| _|
dd| _|
dd| _|
dd| _|
dd| _d| _ d| _!d| _"d| _#d| _$i | _%d| _&d| _'i | _(i | _)|
dd| _*d| _+d| _,d| _-|
dd| _.| j.rt/j0|
d d! t/j1|
|
d"d#|
d$d%|
d&d'|	d(dd) tj	2|	d*}tj|dd zdd+l3m4} ||| _5W n   d,| _5Y || _6|
d-d| _7|
d.d,}|d,ur/t8|t9r/|:d/}|| _;|
d0d,}|d,urHt8|t9rH|:d/}|| _<d,S )1a  
        Initializes the Trainer class with the model, optimizer, scheduler, dataloaders, and other settings.

        Args:
            model (torch.nn.Module): The model to be trained.
            optim (torch.optim.Optimizer): The optimizer to use for training.
            scheduler (torch.optim.lr_scheduler._LRScheduler): The learning rate scheduler.
            dataloader_train (torch.utils.data.DataLoader): The DataLoader for the training dataset.
            dataloader_val (torch.utils.data.DataLoader): The DataLoader for the validation dataset.
            **kwargs: Additional keyword arguments:
                      max_epoch (int): The maximum number of epochs for training.
                      output_dir (str): The directory where model checkpoints will be saved. Default is './'.
                      resume (str, optional): The file path to a checkpoint to resume training from.
        devicer   Texist_okresumer   	max_epochd   log_interval2   save_checkpoint_intervali  validate_intervalkeep_nbest_modelsi  avg_keep_nbest_models_typeaccavg_nbest_model
   
accum_gradr   	grad_clipg      $@grad_clip_typeg       @         reset_gpu_cacheF	use_wandbZwandb_tokenkeyZwandb_projectZ
my_projectZ
wandb_teamZmy_teamZwandb_exp_nameZmy_expZtraining)configprojectentitynamedirZjob_typereinitZtensorboard)SummaryWriterNdeepspeed_configexcludes,effective_save_name_excludes)=rank
local_rank
world_sizer   r   getr#   r"   ospathexistsmakedirsr&   start_epochr'   r)   batch_totalr   float32r   r    r!   r   r   r+   r,   r-   r.   r0   r2   r3   r4   train_acc_avgtrain_loss_avgval_acc_avgval_loss_avgZbest_acc_idxsaved_ckptsZstep_or_epochbest_step_or_epochval_acc_step_or_epochval_loss_step_or_epochr8   start_data_split_i
start_stepstep_in_epochr9   wandblogininitjoinZtensorboardXrB   writerr   rC   
isinstancestrsplitrD   rF   )selfrG   rH   rI   r   r   r    r!   r   r"   kwargsZtensorboard_dirrB   rD   rF   r   r   r   __init__8   s   


	




zTrainer.__init__Nc                 K   sN  | j s| jr
t  |du rdn|}| jrtd| d| j d || j| j	| j
| j| j|||dd|dd| j|d	d|d
dd}	|}t|drW|j |	d< |r_| |	d< tj| jdd |du rqd| }
nd| d| }
tj| j|
}t  |j| j|
|	d W d   n1 sw   Y  td| d ttj| jd}t  |j| jd|	d W d   n1 sw   Y  | jdkr|
| _| jdkrC| j	|
 | j	| j kr$|
| _ttj| jd}t  |j| jd|	d W d   n	1 sw   Y  td| j	| j dd|  ntd| j	|
 dd| j	| j ddtj| j| j  nq| jdkr| j
|
 | j
| j kr|
| _ttj| jd}t  |j| jd|	d W d   n	1 s{w   Y  td| j
| j dd|  n#td | j
|
 dd!| j
| j ddtj| j| j  ntd" t| d#| j d$|
 | j|
< | jdkrt| j| jkr| jdkrt| j| jjd%}n	t | j| jjd%}|| jv r| j|= tj| j|}td&|  tj!|rt"#| n| jrn| j$dkrtd| d| j$ d'| j d || | | j| j	| j
| j| j|||dd|dd| j|d	d|d
dd(}	|}t|drj|j }n| }| j%durtd)| j%  i }|& D ]+}| j%D ]}|'d*d}|(|rtd+| d,| d-  qq|| ||< q||	d< n||	d< |r| |	d< tj| jdd |du rd| }
nd| d| }
tj| j|
}t)|	| td| d ttj| jd}t)|	| | jdkr|
| _| jdkra| j	|
 | j	| j krB|
| _ttj| jd}t)|	| td| j	| j dd|  nxtd| j	|
 dd| j	| j ddtj| j| j  nY| jdkr| j
|
 | j
| j kr|
| _ttj| jd}t)|	| td| j
| j dd|  n#td | j
|
 dd!| j
| j ddtj| j| j  ntd" t| d#| j d$|
 | j|
< | jdkrt| j| jkr| jdkrt| j| jjd%}n	t | j| jjd%}|| jv r| j|= tj| j|}td&|  tj!|rt"#| | j s| jr%t  dS dS ).a`  
        Saves a checkpoint containing the model's state, the optimizer's state,
        and the scheduler's state at the end of the given epoch. This method is
        intended to be called at the end of each epoch to save the training progress.

        Args:
            epoch (int): The epoch number at which the checkpoint is being saved.
        NzSave checkpoint: , rank: 
data_split_ir   data_split_numr   rS   rR   )epochrV   rX   rY   rW   r.   stepr\   rj   rk   rP   rS   rR   module
state_dictscaler_stateTr$   model.pt.ep.)save_dirtagZclient_statez
Checkpoint saved to model.ptr7   r/   zmodel.pt.bestzUpdate best acc: z.4f, zNo improvement in acc: z < losszUpdate best loss: zNo improvement in loss: z > ZUndoZval_Z_step_or_epochr:   zDelete: z, local_rank: )rl   	optimizer	schedulerrV   rX   rY   rW   r.   rm   r\   rj   rk   rP   rS   rR   zeffective_save_name_excludes: module.key:  matching: z, not save it)*r   r   distbarrierr   logginginforH   rV   rX   rY   rW   r.   rJ   rP   hasattrrn   ro   rK   rN   r"   rL   r`   r   no_gradsave_checkpointr   printgetattrr-   lenminmaxrM   
misc_utilsZsmart_removerG   rF   keysreplace
startswithsave)re   rl   rm   modeloptimry   scalerr\   rf   state	ckpt_namefilenameZlatestZ	best_ckptr;   ro   Zdst_state_dictkk_exk_tmpr   r   r   r      s  









4

4









4
4

zTrainer.save_checkpointc              	   C   sr  | j r)| jrtj| jd}tj|r|| jd\}}|d | _|d | _	d|v r0|d ni | _
d|v r;|d ni | _d|v rF|d nd| _d|v rQ|d nd	| _d
|v r\|d
 nd	| _d|v rg|d nd	| _| jdu rqd	n| j| _d|v r}|d nd	| _| jdu rd	n| j| _t|d  d|v r|d nd	| _d|v r|d nd	| _|| j td| d nrtd| d nhtj| jd}tj|r!tj|dd}|d | _|d }| }	|	 D ]v}
d}| jdur| jD ]}|
dd}||rtd|
 d| d d} qq|rq|
ds-d|
 | v r-d|
 }n|
drDd|
 | vrD|
ddd}n|
}|| v rT|| |	|
< qtd|
 d|  q| |	 | |d   | |d!  |durd"|v r| |d"  |d | _	d|v r|d ni | _
d|v r|d ni | _d|v r|d nd| _d|v r|d nd	| _d
|v r|d
 nd	| _d|v r|d nd	| _| jdu rd	n| j| _d|v r|d nd	| _| jdu rd	n| j| _t|d  d|v r|d nd	| _d|v r|d nd	| _|| j td| d ntd| d | j!s1| j"r7t#$  dS dS )#z
        Resumes training from a checkpoint at the given file path.
        Loads the model's state, the optimizer's state, and the scheduler's state.

        Args:
            resume_path (str): The file path to the checkpoint to resume from.
        ru   rl   rV   rX   rY   rW   r7   rj   r   rP   rm   Nr\   rR   rS   z%Checkpoint loaded successfully from ''zNo checkpoint found at 'z', does not resume status!cpu)Zmap_locationro   Frz   r{   r|   z
, excludedTr   zMiss key in ckpt: model: z, ckpt: rx   ry   rp   )%r&   r   rK   rL   r`   r"   rM   Zload_checkpointrO   rV   rX   rY   rW   rZ   rP   r[   r\   r   rR   rS   tor#   isfiler   loadro   r   rD   r   r   r   r   Zload_state_dictr   r   r}   r~   )re   r   r   ry   r   Zckpt_
checkpointZ	src_stateZ	dst_stater   Zexcludes_flagr   r   Zk_ddpr   r   r   resume_checkpoint  s   









zTrainer.resume_checkpointc                 K   s  | j s	| js	| jrt  td| d| j d |  | j	}	|
  i }
td| j}|j| t }|}t|D ]E\}}|  jd7  _|  jd7  _i |||dd|dd||dd | j| jd	}t }|| d
|d d< t|| j}t}| j s| jr||	 dkr|jn|}| 8 t }| j|||d t }|| d
|d d< | j|||d t }|| d
|d d< W d   n1 sw   Y  | j|||||d t | d
}t }|| d
|d d< ||d d< | d |d< t||d< | j| |d   ! "  |d  | _d|d v r=| j#| |d d   ! "  |d  | _#| j$|dd | j| j% dkr\| j&|||| j'|d | jd | j| j( dkr| j)||||||d | j|dd|dd| j| j#d t }q@| j s| js| jrtj| jtj*d| j}tj| j#tj*d| j}tj+|tj,j-d tj+|tj,j-d |  ! " | j. | _|  ! " | j. | _#dS dS )z
        Defines the training process for a single epoch with gradient accumulation.
        Args:
            epoch (int): The current epoch number.
        zTrain epoch: rh   ri   r   r   rj   rk   r[   )speed_statsrl   	batch_idxrj   rk   log_steprP   r\   0.3fr   	data_load	loss_dictforward_timeZbackward_timeNZ
optim_time
total_timelrbatch_num_epochrw   r/   statstrainrt   )r   dataloader_valrl   ra   rm   r\   )
r   r   ry   r   rm   r\   rj   rk   rS   rR   r   op)/r   r   r   r}   r~   r   r   rG   r   r2   	zero_gradr   tensorr   r#   batch_sampler	set_epochtimeperf_counter	enumeraterP   r\   rJ   r   r   Zno_syncforward_stepbackward_stepupdate_stepZget_last_lrr   rS   detachr   itemrR   logr,   validate_epochra   r+   r   rQ   
all_reduceReduceOpSUMrI   )re   r   r   ry   r   Zdataloader_trainr   rl   rf   r2   r   Ziterator_stoptime_begtime5r   batchr   time1Z
my_contexttime2time3Ztime4r   rS   rR   r   r   r   train_epoch"  s   

 	


zTrainer.train_epochc                 C   sv   t | j| jd |di |}W d    n1 sw   Y  |\}}}dd | D }||d< ||d< ||d< d S )Nr   c                 S   s   i | ]\}}|d ur||qS Nr   .0r   vr   r   r   
<dictcomp>  s    z(Trainer.forward_step.<locals>.<dictcomp>rw   r   weightr   )r   r   r   items)re   r   r   r   retvalrw   r   r   r   r   r   r     s   
zTrainer.forward_stepc                 C   sP   |d }| j r||}d S || j }| js| jr"||  d S |  d S )Nrw   )r   Zbackwardr2   r    r!   Zscale)re   r   r   r   rw   Zscaled_lossr   r   r   r     s   
zTrainer.backward_stepc                 C   s   |d }| j r|  d S |d | j dkrg| jdkr=tjjj| | j| j	d}t
|s=td| d |  d S | jsC| jrGt  | jsM| jrW|| |  n|  |  |jdd d S d S )	Nr   r   r   )Zmax_normZ	norm_typezThe grad norm is z. Skipping updating the model.T)Zset_to_none)r   rm   r2   r3   r   nnutilsZclip_grad_norm_
parametersr4   isfiniter   warningr   r   r   r}   r~   r    r!   update)re   r   r   ry   r   r   r   Z	grad_normr   r   r   r     s2   




zTrainer.update_stepc                 K   s  | j s	| js	| jrt  td| d| j d |  t	
  i }t }|}|j| t|D ]\}	}
i ||	|dd|dd|	|dd |	d |	d d	d
	}t }|| d|d d< t|
| j}
t }| j||
|d t }|| d|d d< t | d}t }||d d< t||d< | j|dd t }| j|	 |d     |	d  | _d|d v r| j|	 |d d     |	d  | _q6| j s| js| jr(t	j| jt	jd| j}t	j| jt	jd| j}tj|tjj d tj|tjj d |   | j! | _|   | j! | _W d   n	1 s3w   Y  |dddu rGd| }nd| d|d }| j| j"|< | j| j#|< | j sj| jsj| jrnt  |$  dS )z
        Defines the validation process for a single epoch.
        Should be implemented with the actual model validation steps.

        Args:
            epoch (int): The current epoch number.
        zValidate epoch: rh   ri   rj   r   rk   r   r[   r5   )	r   rl   r   rj   rk   r   rP   r\   r   r   r   r   r   r   r   r   valr   rw   r/   r   r   r   Nr\   rq   rr   )%r   r   r   r}   r~   r   r   rG   evalr   r   r   r   r   r   r   rJ   r   r#   r   r   r   rU   r   r   r   rT   r   rQ   r   r   r   r   rI   rX   rY   r   )re   r   r   rl   ra   rf   r   r   r   r   r   r   r   r   r   r   rU   rT   r   r   r   r   r     sx   

;zTrainer.validate_epochr   r   c              	   K   sJ  |d     }|d }|d }|d }|d }|d }	|d }
|d }|d	 }|d
 }|d }|dd }|d | j dkr|d urH|n|}dtj d d d tj	 d d d tj
 d d d tj d d d }t| | d}t| | d}dg | d| j d| d| j d| d| d|d  d|	 d| d| d|dd|ddt|dd |dd!|
dd"d#d$ | D  d%| d%| }t| d&| j d'| |d&| j d(| |
i}| j}|d ur|d&| j d'| || |d&| j d(| |
| | D ](\}}|d)| j d*| d| | | | |d)| j d*| d| < q7| D ](\}}|d)| j d*| d| t|| t||d)| j d*| d| < qd| jrtd urtj||d+ d S d S d S d S ),Nrw   rl   r   r\   rP   r   r   r   r   rj   rk   r   r   r   zWGPU, memory: usage: {:.3f} GB, peak: {:.3f} GB, cache: {:.3f} GB, cache_peak: {:.3f} GBi   Z	_loss_avgZ_acc_avgr7   rh   z	, epoch: /z, data_slice: z, step_in_slice: z, step_in_epoch: z, total step: z, (loss_avg_rank: z.3fz), (loss_avg_slice: z), (ppl_avg_slice: z.3ez), (acc_avg_slice: z), (lr: z), c                 S   s*   g | ]\}}|t |   d fqS )   )roundr   r   r   r   r   r   r   
<listcomp>]  s   * zTrainer.log.<locals>.<listcomp>rv   rG   z_loss/z_lr/Z
stats_rankr   )Zsetp)r   r   r   rJ   r)   formatr   r   Zmemory_allocatedZmax_memory_allocatedZmemory_reservedZmax_memory_reservedr   r`   rG   r'   mathexpr   r   r   ra   Z
add_scalarr   r9   r]   r   )re   r   rt   rf   rw   rl   r   r\   rP   r   r   r   r   rj   rk   r   Zgpu_infoZloss_avg_epochZacc_avg_epochdescriptionZdescription_dictra   r;   varr   r   r   r   .  s   
	


&$&$
0zTrainer.logc                 C   sB   | j s| jr
t  |d ur|  | j s| jrtj  d S d S r   )r   r   r}   r~   closer   distributedZdestroy_process_group)re   ra   r   r   r   r   x  s   zTrainer.closec           
      K   s  | j rSddlm} ddlm} ddlm} ttj	
dd}ttj	
dd}ttj	
dddkrNtd	 ||||| d
 td ||||| d
 d }	 |S | jrvttj	
dd}	||	}t||	g|
di 
ddd}|S |j|
ddd}|S )Nr   ).estimate_zero2_model_states_mem_needs_all_live).estimate_zero3_model_states_mem_needs_all_live)*convert_zero_checkpoint_to_fp32_state_dictZLOCAL_WORLD_SIZEr   Z
WORLD_SIZEZRANKz/Estimating model states memory needs (zero2)...)Znum_gpus_per_nodeZ	num_nodesz/Estimating model states memory needs (zero3)...Z
LOCAL_RANKZ
train_conffind_unused_parametersF)Z
device_idsr   r#   r   )r#   )r   Z$deepspeed.runtime.zero.stage_1_and_2r   Zdeepspeed.runtime.zero.stage3r   Zdeepspeed.utils.zero_to_fp32r   intrK   environrJ   r   r   r   r   DDPr   )
re   r   rf   r   r   r   Zlocal_world_sizerI   r#   rH   r   r   r   
warp_model  sF   


zTrainer.warp_modelc                    s  ddl m} ddlm} ddlm}m} dd l}t	d  
dd}||v s)J |
|}	|	| fi  
d}t	d	  
d
d}
|
|v sLJ |
|
|fi  
d}
| jrdd l}|d| ji}t| jd}||}W d    n1 sw   Y  d|v r|d d rtj| _d|v r|d d rtj| _d|v rd }d
|v rd }
n fdd}
|j||||
| d\}}}}
|||
fS )Nr   )optim_classes)scheduler_classes)	OmegaConf
DictConfigzBuild optimr   ZadamZ
optim_confzBuild schedulerry   Zwarmuplrscheduler_confrC   rZbf16r   Zfp16rx   c                    s   | fi   dS )Nr   )rJ   )optrf   Zscheduler_classr   r   ry     s   z/Trainer.warp_optim_scheduler.<locals>.scheduler)argsr   rx   Zlr_schedulerZmodel_parameters)Zfunasr.optimizersr   Zfunasr.schedulersr   Z	omegaconfr   r   jsonr   r   rJ   r   r   	deepspeedcreaterC   openr   r   r   r   r   Z
initialize)re   r   rf   r   r   r   r   r   r   Zoptim_classry   r   r   ZfinZ
ds_configsrx   r   r   r   r   warp_optim_scheduler  sJ   




zTrainer.warp_optim_scheduler)	r   r   r   FFFFFr   )NNNNNN)NNNN)NNNNNNNr   )Nr   )__name__
__module____qualname____doc__boolrc   rg   r   r   r   r   r   r   r   dictr   r   r   r   r   r   r   r   r   (   s|    	

q
 s
 
z
"
]

J.r   )NF)'r   rK   r   r   r   r   r   Ztorch.distributedr   r}   Ztorch.cuda.ampr   r   
contextlibr   r   pathlibr   Ztorch.nn.parallelr	   r   Ztorch.distributed.fsdpr
   ZFSDPZfunasr.train_utils.device_funcsr   Zfunasr.train_utils.recursive_opr   Z'funasr.train_utils.average_nbest_modelsr   Z*torch.distributed.fsdp.sharded_grad_scalerr   Zfunasr.utils.miscr   miscr   r]   r   r   r   r   r   r   <module>   s2    