o }oiN@s>ddlZddlZddlZddlmZddlmZmZmZddl Z ddl m Z ddl m Z mZ d*dd d ed e fd d Z d+dedededeeed df ddZ      d,de dededeeedededd edededdeded e fd d!ZGd"d#d#e ZGd$d%d%ZGd&d'd'eZGd(d)d)eZdS)-N)chain)ListLiteralOptional)_IndexBatchSamplerWrapper) DataLoaderDatasetTFdatasetr drop_lastreturncKs&t|fd|ji|}||_||_|S)N collate_fn)rr _drop_last!_pad_samples_to_global_batch_size)r r pad_samples_to_global_batch_sizekwargsoutputrG/home/ubuntu/.local/lib/python3.10/site-packages/nemo/lightning/data.pycreate_dataloadersr global_rankmicro_batch_sizeglobal_batch_sizerampup_batch_sizec Csddlm}ddlm}zddlm}m}m}m} m } d} Wn(t t fyFt dddlm}ddlm}m}m} dd lm} d } Ynw|} tj|d d krY| j} n|} | rdd lm}|durt| | ||| j|d ddSt||r||ksJ||ksJ| ||| jksJdStddd lm}|dur| | ||| j|d ddSt||r||ksJ||ksJ| ||| jksJdStd)aD Initializes the data for distributed training by setting up the microbatch calculator based on the provided global rank and data configuration. This function checks if the microbatch calculator has already been initialized. If it has, the function validates that the current configuration matches the initialized settings. If the calculator has not been initialized, it sets up a new one with the provided configuration. Args: global_rank (int): The global rank of the current process. config (DataConfig): The data configuration object containing settings for global batch size, micro batch size, data parallel size, and optional ramp-up batch size. Raises ------ Exception: If the microbatch calculator has already been initialized with different settings. r).NEMO_MEGATRON_MODEL_PARALLEL_APPSTATE_OVERRIDE)AppState)!ConstantNumMicroBatchesCalculatorget_current_global_batch_sizeget_micro_batch_sizeget_num_microbatches init_num_microbatches_calculatorTzCMegatron num_microbatches_calculator not found, using Apex version.)ConstantNumMicroBatches)rrr)setup_microbatch_calculatorFfalsetrue)#_GLOBAL_NUM_MICROBATCHES_CALCULATORN)rankrrdata_parallel_sizerdecrease_batch_size_if_neededz*Microbatch calculator already initialized.)nemo.lightning._strategy_libr nemo.utilsr)megatron.core.num_microbatches_calculatorrrrrr ImportErrorModuleNotFoundErrorloggingwarningapex.transformer.microbatchesr (apex.transformer.pipeline_parallel.utilsr!osenvirongetlowerrr$r& isinstance Exception)rrrrrrrrrrrMCORE_MB_CALCULATOR app_stateinit_global_rankr$rrrr!&s^         r!singletrain dataloaderconsumed_samplesdataloader_type)r:cyclicbatchrdataloader_mode)r; validationtestpredictr% world_sizec Cs|dkrtt|j||||| | ||d } n4|dkr'tt|j||| | |d} n"|dkrBddlm} | t|j|||| | || d} nt|d |d vrQt| } t|j| |j |j |j |j d S) a This function takes an existing PyTorch `DataLoader` and configures it to use a Megatron sampler. The Megatron sampler is responsible for splitting the data into batches during training with Megatron. Args: dataloader (DataLoader): The original PyTorch DataLoader to wrap. micro_batch_size (int): The size of each micro-batch. global_batch_size (int): The effective size of the training batch across all data parallel devices. rampup_batch_size (Optional[List[int]]): A list of target batch sizes for a gradual rampup schedule during training (optional). consumed_samples (int, optional): The number of samples consumed before starting this iteration (defaults to 0). dataloader_type (Literal["single", "cyclic", "batch"], optional): The type of Megatron sampler to use. Valid options are: - "single": Uses `MegatronPretrainingSampler` for single pass data sampling. - "cyclic": Uses `MegatronPretrainingRandomSampler` for cyclic data sampling. - "batch": Uses `MegatronPretrainingBatchSampler` for batch sampling. This is the option to use for fine-tuning workloads, where sequence lengths are variable between samples. Sampling the entire global batch together ensures that sequences in a global batch are padded to the same lengths. Defaults to "single". drop_last (bool, optional): Whether to drop the last incomplete batch (defaults to True). pad_samples_to_global_batch_size (bool, optional): Whether to pad the last incomplete batch to the `global_batch_size` (defaults to False, only applies when `drop_last` is False). dataloader_mode (Literal["train", "validation", "test", "predict"]): The mode of dataloader. Returns: DataLoader: A new DataLoader instance with the configured Megatron sampler. r:) total_samplesr>rrrdata_parallel_rankr&r rr@)rGr>rrHr&r rAr)MegatronPretrainingBatchSampler)rGr>rrrHr&r rz" dataloader type is not supported.)rDrE) batch_sampler num_workers pin_memorypersistent_workersr ) MegatronPretrainingSamplerlenr MegatronPretrainingRandomSamplerLnemo.collections.nlp.data.language_modeling.megatron.megatron_batch_samplersrIr6rrrKrLrMr ) r=rrrr>r?r rrBr%rFrJrIrrradd_megatron_samplersX.   rRcs"eZdZdZdfdd ZZS)WrappedDataLoaderz@Wrapper around torch DataLoader which stores the dataloader moder;c stjdi|||_dS)Nr)super__init__mode)selfrVdataloader_kwargs __class__rrrUs zWrappedDataLoader.__init__)r;)__name__ __module__ __qualname____doc__rU __classcell__rrrYrrSsrSc@sjeZdZ    ddededededed ed eed eed eed dfddZddZe j ddZ dS)BaseMegatronSamplerTNFrGr>rrHr&r rrrr c Cs|dkr td||dkrtd||dkr!td|||kr/td|d||durM|durM|||dkrMtd|d|d |d | rW|durWtd ||_||_||_||_|j||_||_||_| |_t d |d |dS)Nrzno sample to consume: z2micro_batch_size size must be greater than 0, but z/data parallel size must be greater than 0, but z9data_parallel_rank should be smaller than data size, but z >= z`global_batch_size` (z)) is not divisible by `micro_batch_size (z) x data_parallel_size (z)`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: ) RuntimeErrorrGr>rrH$micro_batch_times_data_parallel_sizer rrr-info rWrGr>rrHr&r rrrrrrrUsH   zBaseMegatronSampler.__init__cCsX|jdur"|jr|j|j}n |j|jd|j}||j|jS|jd|jdS)Nr<)rr rGrb)rWnum_global_batchesrrr__len__(s zBaseMegatronSampler.__len__cCsdSNr)rWrrr__iter__4szBaseMegatronSampler.__iter__TNNF) r[r\r]intboolrlistrUrfabcabstractmethodrhrrrrr`s:   3 r`csheZdZ    ddededededed ed eed eed eeffd d ZddZddZ Z S)rNTNFrGr>rrHr&r rrrc s>tj||||||||| d ||krtd|d|dS)N) rGr>rrHr&r rrrzno samples left to consume: z, )rTrUrardrYrrrU9s  z#MegatronPretrainingSampler.__init__cCs|j|j}||j}||fSrg)rHr)rW start_idxend_idxrrrget_start_end_idxSs  z,MegatronPretrainingSampler.get_start_end_idxccsg}t|j|j}|js&|jr&t| |j}td| dd}t||}|D]}||t||j krE| \}}|||Vg}q(t|dkre|jsg|jrVJd| \}}|||VdSdSdS)Nr<rzDwith pad_samples_to_global_batch_size all batches should be complete) ranger>rGr rrOrrappendrbrq)rWrAindicespad_samples_num pad_indicesidxrorprrrrhXs,     z#MegatronPretrainingSampler.__iter__ri) r[r\r]rjrkrrlrUrqrhr_rrrYrrN8s4  rNcsheZdZ    ddedededed ed ed eed eed eddffdd ZddZddZZ S)rPTNFrrGr>rrHr&r rrseedr c sVtj||||||||d|rJd|s|jdkrtd|j|j|_| |_dS)N)rGr>rrHr&r rrzB`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)rTrUrbrarGlast_batch_sizery) rWrGr>rrHr&r rrryrYrrrUrs(   z)MegatronPretrainingRandomSampler.__init__cCs|j|jr|jnd}||j|}|jdur2|jr ||j}n ||jd|j}||j|jS|jr:||jS|d|jS)Nrr<)rGr rzr>rrb)rWactive_total_samplesnum_available_samplesrerrrrfs   z(MegatronPretrainingRandomSampler.__len__c #s|j|j}|j||_|j|}||jdksJ|j|j}|j|j|j}||}|j|t}| |j |jtj ||d }fdd||dD}g} |D]} | | t| |jkru|j|j7_| Vg} qZt| dkr|js| VdSdSdS)Nr) generatorcsg|]}|qSrr).0xrorr sz=MegatronPretrainingRandomSampler.__iter__..)rGrzr>epochrbrrHtorch Generator manual_seedryrandpermtolistrtrOr ) rWr{current_epoch_samplesr& bucket_size bucket_offsetg random_idx idx_rangerArxrrrrhs0       z)MegatronPretrainingRandomSampler.__iter__)TNFr) r[r\r]rjrkrrUrfrhr_rrrYrrPqs8   !rP)TFrg)Nrr:TFr;rr<)rmr-r1 itertoolsrtypingrrrr'lightning.pytorch.overrides.distributedrtorch.utils.datarrrkrrjr!rRrSr`rNrPrrrrs     f     c D9