o }oi;@sddlmZddlmZmZmZmZddlZddl m Z ddl Z ddlmZmZddlmZddlmZmZmZddlmZddlmZdd lmZmZmZdd lm Z dd l!m"Z"Gd d d e j#eZ$dS)deepcopy)AnyDictLiteralOptionalN)EVAL_DATALOADERSTRAIN_DATALOADERS)parallel_state) WorkerConfigget_savable_loaderget_train_dataset)Self) TaskEncoder)IOMixin serializationtrack_io)MegatronDataSampler)loggingcseZdZdZ           d)deded ed ed ed ed edBdedededBdeedeedeeddffdd Z de j e fddZ d*dedfddZdefddZdefd d!Zd+d"d#Zdeeeffd$d%Zd&eeefddfd'd(ZZS),EnergonDataModulea7 A PyTorch Lightning DataModule for handling Energon datasets. It provides a seamless interface to load training and validation data, saving, and sampling strategies. The module integrates with the Megatron-Energon framework for efficient data handling in large-scale distributed training. NTdpath train_encoder seq_lengthmicro_batch_sizeglobal_batch_size num_workersnum_val_workers pin_memoryshuffle_buffer_sizemax_samples_per_sequencedecoder_seq_lengthpacking_buffer_sizevalidation_encoderreturnc st||_||_| |_||_||_||_||_||_||_| |_ | |_ ||_ | r-| n||_ d|_ t|j|j|j|jd|_dd|j_d|_d|_| |_|pS|j|_||_|j j|_dS)a` Initialize the EnergonModule. Parameters: path (str): Path to the dataset (must be in webdataset format, and prepared using energon prepare). train_encoder (TaskEncoder): Encoder for training data. seq_length (int, optional): The maximum sequence length for tokenized text. Defaults to 2048. micro_batch_size (int, optional): The batch size for training and validation. Defaults to 1. global_batch_size (int, optional): The global batch size across all processes. Defaults to 1. num_workers (int, optional): Number of workers for data loading. Defaults to 1. num_val_workers (int, optional): Number of workers for validation data loading. Defaults to num_workers. pin_memory (bool, optional): Whether to pin memory in the DataLoader. Defaults to True. shuffle_buffer_size (int, optional): Size of the shuffle buffer. Defaults to 100. max_samples_per_sequence (int, optional): Maximum number of samples per sequence to load from memory. Defaults to None (loads the whole tar file at once). decoder_seq_length (int, optional): The max seq length for the decoder. Used in encoder-decoder models. packing_buffer_size (int, optional): Size of the packing buffer for batched samples. Defaults to None. validation_encoder (TaskEncoder, optional): Encoder for validation data. Defaults to None and will be the same as train_encoder. **kwargs: Additional keyword arguments passed to get_train_dataset() of Energon. r)seq_lendecoder_seq_lenrrcSs|S)N) dataloaderr)r)Y/home/ubuntu/.local/lib/python3.10/site-packages/nemo/collections/vlm/data/data_module.pyfsz,EnergonDataModule.__init__..N)super__init__rrr#rrrr r!r"rr%init_global_stepr data_samplertransform_dataloadertrain_dataloader_objectval_dataloader_objectr$rkwargs tokenizer)selfrrrrrrrr r!r"r#r$r%r4 __class__r)r+r.)s8 '  zEnergonDataModule.__init__cKsTdd|D}|D]}tt|stt|q tjt|fi|}|S)NcSs"i|] \}}|dvr|t|qS))r%rr).0kvr)r)r+ ss"z-EnergonDataModule.io_init..)itemsvaluesrfind_node_traversertyperfdlConfig)r6r4 cfg_kwargsvalcfgr)r)r+io_initqs  zEnergonDataModule.io_initrDsplit)trainrDc CsT|dvrtd|dkr|jn|j}t|jf|j|||j||j|jd|j }|S)a Provide the dataset for training or validation. This method retrieves the dataset for the specified split (either 'train' or 'val') and configures it according to the worker configuration. Parameters: worker_config: Configuration for the data loader workers. split (Literal['train', 'val'], optional): The data split to retrieve ('train' or 'val'). Defaults to 'val'. Returns: Dataset: The dataset configured for the specified split. >rDrHz=Invalid value for split. Allowed values are 'train' or 'val'.rD) batch_size task_encoder worker_configr$ split_partr!r") ValueErrorr%rr rrr$r!r"r4)r6rKrGrJ_datasetr)r)r+datasets_provider{s"  z#EnergonDataModule.datasets_providerc Cs|jr|jj|_|j|j_td|j|jr|jSts0td|j t |j }n&t }t }t}td|d|d|dt |||j |ddd }|j|d d }t||d }||_|jS) a Initialize and return the training DataLoader. This method initializes the DataLoader for the training dataset. It uses the global step from the trainer to configure the data sampler and ensures that the parallel state is initialized correctly for distributed training. Returns: TRAIN_DATALOADERS: The DataLoader for the training dataset. z5Ttrain dataloader initializing with init_global_step zZData loader parallel state is not initialized,using default worker config with no_workers z( Train dataloader initializing withrank world_size  data_parallel_group z ****** Nrrank world_sizerdata_parallel_groupworker_debug_pathworker_log_levelrHrGrK)trainer global_stepr/r0rinfor2r is_initializedrr default_worker_configget_data_parallel_rankget_data_parallel_world_sizeget_data_parallel_grouprOr )r6rKrSrTrU train_datasetenergon_dataloaderr)r)r+train_dataloadersJ   z"EnergonDataModule.train_dataloadercCs|jr|jStstd|jt|j}n%t }t }t }td|d|d|t|||j|ddd}|j |dd }t ||d }||_|jS) an Initialize and return the validation DataLoader. This method initializes the DataLoader for the validation dataset. It ensures that the parallel state is initialized correctly for distributed training and returns a configured DataLoader object. Returns: EVAL_DATALOADERS: The DataLoader for the validation dataset. zjData loader val data loader parallel state is not initialized,using default worker config with no_workers zrank rPrQNrrRrDrXrY)r3r r]rr\rr r^rr_r`rarOr )r6rKrSrTrU val_datasetenergon_loaderr)r)r+val_dataloaders2  z EnergonDataModule.val_dataloadercCstddS)z Return None as test dataset split does not exist. This method overrides the test_dataloader method and returns None since the test dataset split is not defined or used in this module. Returns: None z-Data loader test dataset split does not existN)rwarning)r6r)r)r+test_dataloaders z!EnergonDataModule.test_dataloadercCsz|jr6|jj}g}tjtkr|jdd}|j |jj |j }|dur)g}t d|||dSt diS)aT Save the state of the data module. This method is called when saving a checkpoint. It generates and saves the state of the data module, including the state of the dataloader and the number of consumed samples. Returns: Dict[str, Any]: A dictionary containing the state of the data module. r)global_dst_rankNz:Data loader saving dataloader state dict consumed samples )dataloader_stateconsumed_sampleszHtrainer object not connected to data module object returning empty state)rZrdtorch distributedget_rankr get_model_parallel_src_ranksave_state_globalr0compute_consumed_samplesr[r/rr\rh)r6dataloader_objstaterlr)r)r+ state_dicts     zEnergonDataModule.state_dictruc Csd|vrtd|dS|d}z |jr'|jj|tdn td|t dWnt yN}ztd|WYd}~nd}~wwzdd l m }Wnt tfymtd dd lm }Ynw|d }||j_||j_td |||d ddS)az Load the state of the data module from a checkpoint. This method is called when loading a checkpoint. It restores the state of the data module, including the state of the dataloader and the number of consumed samples. Parameters: state_dict (Dict[str, Any]): The state dictionary containing the saved state of the data module. rkzpData loader state cannot be resumed from state_dict, it does not have the required key dataloader_state. It has NzData loader state restoredz%Cannot restore state from state_dict zhCannot restore state from state_dict: Is the trainer object is initialized and attached to datamodule???zFailed to dataloader restore state due to [Please ensure you are using same version of energon while saving and loading, Continuing without restoring data loader] : r)update_num_microbatcheszCMegatron num_microbatches_calculator not found, using Apex version.rlz2Data loader load state dict with consumed_samples F)rlconsistency_check)rrhkeysrZ datamodulerdrestore_state_globalr\errorrM Exception)megatron.core.num_microbatches_calculatorrv ImportErrorModuleNotFoundError(apex.transformer.pipeline_parallel.utilsr0init_consumed_samplesprev_consumed_samples)r6rurtervrlr)r)r+load_state_dictsN    z!EnergonDataModule.load_state_dict) rrrrNTrNNNN)rD)r&N)__name__ __module__ __qualname____doc__strrintboolrr.rArBrrFrrOr rdrrgrirrrur __classcell__r)r)r7r+r sb      H ", & ""r)%copyrtypingrrrrfiddlerAlightning.pytorchpytorchpltorch.distributedrm!lightning.pytorch.utilities.typesrr megatron.corer megatron.energonr r r typing_extensionsr&nemo.collections.vlm.data.task_encoderrnemo.lightning.io.mixinrrrnemo.lightning.pytorch.pluginsr nemo.utilsrLightningDataModulerr)r)r)r+s