o wi^R@sdZddlZddlZddlmZddlmZmZddlZddl m Z ddl m Z ddgZ Gd d d ZGd d d eZGd ddeZGdddeZGdddZGdddeZe GdddeZdS)z Dataloaders.N)chain)OptionalTuple)logging) experimentalMegatronPretrainingBatchSampler%MegatronPretrainingRandomBatchSamplerc@sneZdZdZ    ddedededed ed ed eed eed eeddfddZddZ e j ddZ dS)BaseMegatronSampler TNF total_samplesconsumed_samplesmicro_batch_sizedata_parallel_rankdata_parallel_size drop_lastglobal_batch_sizerampup_batch_size pad_samples_to_global_batch_sizereturnc Cs|dkr td||dkrtd||dkr!td|||kr-td|||durK|durK|||dkrKtd|d|d|d | rU|durUtd ||_||_||_||_||_|j||_||_||_ | |_ t d |d |dS) Nrno sample to consume: {}2micro_batch_size size must be greater than 0, but /data parallel size must be greater than 0, but Adata_parallel_rank should be smaller than data size, but {} >= {}`global_batch_size` ()) is not divisible by `micro_batch_size () x data_parallel_size ()`zi`pad_samples_to_global_batch_size` can be `True` only when `global_batch_size` is set to an integer valuez=Instantiating MegatronPretrainingSampler with total_samples: z and consumed_samples: ) RuntimeErrorformatr r r rr$micro_batch_times_data_parallel_sizerrrrinfo) selfr r r rrrrrrr"g/home/ubuntu/sommelier/.venv/lib/python3.10/site-packages/nemo/collections/common/data/data_samplers.py__init__$sN   zBaseMegatronSampler.__init__cCs^|j|j}|jdur&|jr||j}n ||jd|j}||j|jS|d|jdS)N)r r rrr)r!num_available_samplesnum_global_batchesr"r"r#__len__Zs   zBaseMegatronSampler.__len__cCdSNr"r!r"r"r#__iter__gzBaseMegatronSampler.__iter__)TNNF) __name__ __module__ __qualname____doc__intboolrlistr$r(abcabstractmethodr,r"r"r"r#r !s<    6 r c@s(eZdZdZddZddZddZdS) MegatronPretrainingSamplerr cC|j|j}||j}||fSr )rr r! start_idxend_idxr"r"r#get_start_end_idxn  z,MegatronPretrainingSampler.get_start_end_idxcCstd| ddS)r r%)ranger!pad_samples_numr"r"r#_get_padding_indicestsz/MegatronPretrainingSampler._get_padding_indicesccsg}t|j|j}|js"|jr"t| |j}||}t||}|D]}| |t||j krA| \}}|||Vg}q$t|dkra|jsc|jrRJd| \}}|||VdSdSdS)NrzDwith pad_samples_to_global_batch_size all batches should be complete) r@r r rrlenrrCrappendrr=)r!batchindicesrB pad_indicesidxr;r<r"r"r#r,xs,      z#MegatronPretrainingSampler.__iter__N)r.r/r0r1r=rCr,r"r"r"r#r7ks  r7c@seZdZdZddZdS)MegatronCorePretrainingSamplerr cCs dg|S)r Nr"rAr"r"r#rCs z3MegatronCorePretrainingSampler._get_padding_indicesN)r.r/r0r1rCr"r"r"r#rJs rJcsleZdZdZ    ddededed ed ed ed eed eededdffdd ZddZddZ Z S) MegatronPretrainingRandomSamplerr TNFrr r r rrrrrseedrc sVtj||||||||d|rJd|s|jdkrtd|j|j|_| |_dS)Nr r r rrrrrzB`MegatronPretrainingRandomSampler` does not support sample paddingr%a`MegatronPretrainingRandomSampler` does not support drop_last=False when micro_batch_size * data_parallel_size > 1. Please reduce your MBS and data parallelism to 1 if you want to use drop_last=False, or switch to drop_last=True to avoid this error)superr$rrr last_batch_sizerL) r!r r r rrrrrrL __class__r"r#r$(   z)MegatronPretrainingRandomSampler.__init__cCs|j|jr|jnd}||j|}|jdur2|jr ||j}n ||jd|j}||j|jS|jr:||jS|d|jS)Nrr%)r rrOr rr)r!active_total_samplesr&r'r"r"r#r(s   z(MegatronPretrainingRandomSampler.__len__c #s|j|j}|j||_|j|}||jdksJ|j|j|j}||j}|j|t }| |j |jtj ||d }fdd||dD}g}|D]} || t||jkrp|j|j7_|Vg}qUt|dkr|js|VdSdSdS)Nr generatorcg|]}|qSr"r".0xr;r"r# z=MegatronPretrainingRandomSampler.__iter__..)r rOr epochrr rrtorch Generator manual_seedrLrandpermtolistrErDr r!rScurrent_epoch_samples bucket_size bucket_offsetg random_idx idx_rangerFrIr"rZr#r,s.       z)MegatronPretrainingRandomSampler.__iter__)TNFr) r.r/r0r1r2r3rr$r(r, __classcell__r"r"rPr#rKs:    "rKc@seZdZUdZeed<eed<eed< ddededed ed ed ed ed dfddZded dfddZe d efddZ e j ded dfddZ d efddZ e jddZdS)BaseMegatronBatchSampleraHMegatron style BatchSampler. Let mbs, gbs, tp, pp, and dp stand for "micro batch size", "global batch size", "tensor model parallel world size", "pipeline model parallel world size", and "data parallel world size", the number of micro batches (hereafter, nmb) is defined as :math:`nmb = gbs \div (mbs \times dp)`. See `apex/transformer/microbatches.py#L91-L98 `_ for the initial settings of the number of micro batches and `apex/transformer/microbatches.py#L160-L177 _`. for warming up of global batch size. e.g.) `(mbs, gbs, tp, pp, dp) = (1, 16, 1, 1, 2)`, then the number of micro batches is :math:`gbs \div (mbs \times dp) = 16 \div (1 \times 2) = 8`. In this case, an instance of Megatron Batch Sampler on each data parallel rank is expected returns :math:`nmb \times mbs = 8` indices. _global_batch_size_num_micro_batches-_global_batch_size_on_this_data_parallel_rankFr r r rrrrrNc Cs|dkr td||dkrtd||dkr!td|||kr-td||||_||_||_||_||_||_||_|j|j|_ | |dS)aMConstructor of Megatron-LM style Batch Sampler. Args: total_samples: The size of dataset. consumed_samples: The number of samples that have been used. micro_batch_size: The size of each micro batch. global_batch_size: The size of global batch. data_parallel_rank: The value you can obtain via `parallel_state.get_data_parallel_rank()` of megatron.core. data_parallel_size: The value you can obtain via `parallel_state.get_data_parallel_world_size()` of megatron.core. rrrrrN) rrr r r rrrrrupdate_global_batch_size) r!r r r rrrrrr"r"r#r$s*z!BaseMegatronBatchSampler.__init__new_global_batch_sizecCsX||_|j|jdkrtd|jd|jd|jd|j|j|_|j|j|_dS)zUpdate the global batch size.rrrrrN)rlrrr rrmrnr!rpr"r"r#ro6s z1BaseMegatronBatchSampler.update_global_batch_sizecCs|jSr9)rlr+r"r"r#rBsz*BaseMegatronBatchSampler.global_batch_sizecCstd|j|ddS)r zF`self.update_global_batch_size(new_global_batch_size)` is recommended.)rpN)warningswarnrorqr"r"r#rGs cCs6|j|j|j}|jr||jS||jd|jS)zLength of Batch Sampler. ..note:: When `rampup_batch_size` is enabled, the return value can be not exactly precise. r%)r r rr)r!r&r"r"r#r(Ms z BaseMegatronBatchSampler.__len__cCr)r*r"r+r"r"r#r,Zr-z!BaseMegatronBatchSampler.__iter__)F)r.r/r0r1r2__annotations__r3r$ropropertyrsetterr(r5r6r,r"r"r"r#rks>   0  rkc@s.eZdZdZdeeeffddZddZdS)rr rcCr8r9)rrnr:r"r"r#r=ar>z1MegatronPretrainingBatchSampler.get_start_end_idxc#sgt|j|j|jD],}|t|jkr9fddt|j|j|jD}t||jks4J|Vgq tdkrm|j sofddt|jt|jD}|j rh|j|jt|}|dg|}|VdSdSdS)Ncg|]}|qSr"r"rXirFr"r#r[nsz.rcrwr"r"rxrzr"r#r[~r\r?) r@r r rErDrlrrrnrr)r!rIrGnum_padr"rzr#r,gs.  " z(MegatronPretrainingBatchSampler.__iter__N)r.r/r0r1rr2r=r,r"r"r"r#r^s csfeZdZdZ  ddededededed ed ed ed ed dffdd Zd efddZddZZ S)rr Frr r r rrrrrrLrNc sVtj||||||||d|rJd|s|jdkrtd|j|j|_| |_dS)NrMzG`MegatronPretrainingRandomBatchSampler` does not support sample paddingr%a!`MegatronPretrainingRandomBatchSampler` does not support drop_last=False when micro_batch_size * data_parallel_size > 1. Please reduce your MBS and data parallelism to 1 if you want to use drop_last=False, or switch to drop_last=True to avoid this error)rNr$rrr rlrOrL) r!r r r rrrrrrLrPr"r#r$rRz.MegatronPretrainingRandomBatchSampler.__init__cCsH|j|jr|jnd}||j|}|jr||jS||jd|jS)zLength of Random Batch Sampler. ..note:: When `rampup_batch_size` is enabled, the return value can be not exactly precise. rr%)r rrOr r)r!rSr&r"r"r#r(s  z-MegatronPretrainingRandomBatchSampler.__len__c #s|j|j}|j||_|j|}||jdksJ|j|j|j}||j}|j|t }| |j |jtj ||d }fdd||dD}g}|D]} || t||jkrp|j|j7_|Vg}qUt|dkr|js|VdSdSdS)NrrTcrVr"r"rWrZr"r#r[r\zBMegatronPretrainingRandomBatchSampler.__iter__..)r rOr r]rr rrr^r_r`rLrarbrErDrnrlrrcr"rZr#r,s.       z.MegatronPretrainingRandomBatchSampler.__iter__)Fr) r.r/r0r1r2r3r$r(r,rjr"r"rPr#rs6   ")r1r5rr itertoolsrtypingrrr^ nemo.utilsrnemo.utils.decoratorsr__all__r r7rJrKrkrrr"r"r"r#s&   J&Tq'