o
    }oi                     @   sj   d dl Z d dlZd dlmZmZmZ d dlmZ d dl	m
Z
 d dlmZ G dd dZG dd deZdS )	    N)ListLiteralOptional)
DataLoader)MegatronStepc                   @   sF   e Zd ZdejfddZdeddfddZdd
ededefddZ	dS )DataSamplertrainerc                 C   s
   || _ d S N)r   )selfr    r   _/home/ubuntu/.local/lib/python3.10/site-packages/nemo/lightning/pytorch/plugins/data_sampler.pyconnect   s   
zDataSampler.connectglobal_rankreturnNc                 C      t  r	   NotImplementedError)r
   r   r   r   r   setup      zDataSampler.setupr   
dataloaderconsumed_samplesc                 C   r   r	   r   )r
   r   r   r   r   r   transform_dataloader    r   z DataSampler.transform_dataloaderr   )
__name__
__module____qualname__plTrainerr   intr   r   r   r   r   r   r   r      s    r   c                   @   s   e Zd Z								d(deded	ed
eee  ded dedededee fddZdeddfddZ	d)de
dede
fddZd)defddZdedefddZdeddfd d!Zdeddfd"d#Zedefd$d%Zedefd&d'ZdS )*MegatronDataSampler      Nsingler   Tseq_lenmicro_batch_sizeglobal_batch_sizerampup_batch_sizedataloader_type)r"   cyclicbatchinit_consumed_samplesinit_global_step
output_logdecoder_seq_lenc
           
      C   sN   || _ |	| _|| _|| _|| _|| _|| _|| _| j| _d| _	d | _
|| _d S )Nr   )r#   r-   r,   r$   r%   r&   r'   r*   prev_consumed_samplesif_first_stepprev_global_batch_sizer+   )
r
   r#   r$   r%   r&   r'   r*   r+   r,   r-   r   r   r   __init__%   s   
zMegatronDataSampler.__init__r   r   c                 C   s$   ddl m} ||| j| j| j d S )Nr   )setup_microbatch_calculator)nemo.lightning.datar2   r$   r%   r&   )r
   r   r2   r   r   r   r   >   s   zMegatronDataSampler.setupr   r   c                 C   sj   ddl m} ddlm} t|dd}| }| }||| j| j| j	|dkr)| j
nd| j|dv|||d
S )Nr   )parallel_state)add_megatron_samplermodetrain)testpredict)	r$   r%   r&   r   r'   	drop_lastdataloader_moderank
world_size)megatron.corer4   r3   r5   getattrget_data_parallel_rankget_data_parallel_world_sizer$   r%   r&   r*   r'   )r
   r   r   r4   r5   r6   data_parallel_rankdata_parallel_sizer   r   r   r   C   s"   z(MegatronDataSampler.transform_dataloaderc                 C   s   ddl m} ddlm} t| drt| jj|sdS | }| jd ur/| j	| j
| j  }t|S | j||j | j | j  }t|S )Nr   )MegatronStrategy)AppStater   )!nemo.lightning.pytorch.strategiesrD   
nemo.utilsrE   hasattr
isinstancer   strategyr&   r.   r/   current_global_batch_sizer*   rC   r$   num_microbatchesr   )r
   steps_since_resumerD   rE   	app_stater   r   r   r   compute_consumed_samplesY   s   
z,MegatronDataSampler.compute_consumed_samplesstepc                 C   s   t j|| j| j| j| jdS )N)
seq_lengthr$   rL   decoder_seq_length)dataclassesreplacer#   r$   rL   r-   r
   rP   r   r   r   on_megatron_step_startm   s   z*MegatronDataSampler.on_megatron_step_startc                 C   s>   |j sd S | jd ur| j| jkr| jrd|j _d S d S d S d S )NT)r   r&   r0   rK   should_stoprU   r   r   r   on_megatron_microbatches_startv   s   
z2MegatronDataSampler.on_megatron_microbatches_startc              	   C   s   |j }|j}zddlm} W n ttfy$   td ddlm} Y nw | j	| _
|jrT| |jd | j }| jrK|rKt|ddrK|jd|ddd	 || _||dd
 | jrc|rc|jd| j	ddd	 d| _d S )Nr   )update_num_microbatchesCMegatron num_microbatches_calculator not found, using Apex version.   trainingFr   T)prog_bar
batch_size)r   consistency_checkr%   )r   	pl_module)megatron.core.num_microbatches_calculatorrY   ImportErrorModuleNotFoundErrorloggingwarning(apex.transformer.pipeline_parallel.utilsrK   r0   step_irO   r+   r,   r?   logr.   r/   )r
   rP   r   r`   rY   r   r   r   r   on_megatron_step_end   s>   


z(MegatronDataSampler.on_megatron_step_endc              	   C   sF   z
ddl m} W | S  ttfy"   td ddlm} Y | S w )Nr   )get_num_microbatchesrZ   )ra   rj   rb   rc   rd   re   rf   )r
   rj   r   r   r   rL      s   
z$MegatronDataSampler.num_microbatchesc              	   C   sV   zddl m} W n ttfy   td ddlm} Y nw | r'| }|S d}|S )Nr   )get_current_global_batch_sizerZ   r[   )ra   rk   rb   rc   rd   re   rf   )r
   rk   rK   r   r   r   rK      s   
z-MegatronDataSampler.current_global_batch_size)r    r!   Nr"   r   r   TNr   )r   r   r   r   r   r   r   boolr1   r   r   r   rO   r   rV   rX   ri   propertyrL   rK   r   r   r   r   r   $   sL    
	

	(
r   )rS   rd   typingr   r   r   lightning.pytorchpytorchr   torch.utils.datar    nemo.lightning.megatron_parallelr   r   r   r   r   r   r   <module>   s   