o }oi\@sddlmZmZddlmZddlmZmZmZm Z m Z m Z m Z m Z mZmZmZddlZddlmZddlmZddlmZddlmZdd lmZdd lmZdd lmZdd l m!Z!dd l"m#Z#m$Z$ddl%m&Z&ddl'm(Z(ddl)m*Z*ddl+m,Z,ddl-m.Z.ddl/m0Z0ddlm1Z1m2Z2ddl3m4Z4ddl5m6Z6ddl7m8Z8ddl9m:Z:ddl;mZ>ddl?m@Z@ddlAmBZBddlCmDZDmEZEddlFmGZGerddlHmIZIed ZJGd!d"d"eZKGd#d$d$e(ZLGd%d&d&e ZMe@NeGd'eGd(eKfd)d*ZOdS)+) ExitStackcontextmanager) timedelta) TYPE_CHECKINGAnyCallableContextManagerDict GeneratorIteratorListLiteralOptionalUnionN)CPUAccelerator) Accelerator)default_pg_timeout)ClusterEnvironment) CheckpointIO) Precision) DDPStrategy)!_validate_keys_for_strict_loading)_PATH _Stateful)LightningDataModule) _DataFetcher)_WrappingCheckpointIO)CombinedLoader)DistributedDataParallelConfig)OptimizerConfig)Tensornn) noop_hook)Module) Optimizer) DataLoader)override) _strategy_lib) to_fabric)MegatronCheckpointIO)CallbackConnectorMegatronParallel)MegatronStrategy) DataSampler)megatronpytorchcAseZdZdZdddddddddddddddddddddeddddddddfd ed ed eed ed eededededededeedeededdee dee e j dee deedeedeedeeefdeedeed ed!d"ed#ed$ee jd%ed&ed'ed(eed)ed*df@fd+d, Zed[fd-d. Zd/ed*efd0d1Zed2ed*efd3d4Z   5d\d6e d7e!d8ee"d9ee"d:e#d*e$f d;d<Z%ed=e$d*e$fd>d?Z&ed@e'd*e ffdAdB Z(d]dCeed*e)fdDdEZ*d@e+j'd*dfdFdGZ,e  d^dHe-dIe.eee'e$effdJeedKee.ee"eegeffd*df dLdMZ/  d_dHe-dIeee'e$e.eee'e$efffdNed*e.eeffdOdPZ0e d`d@e'dQe.eeee1ffdNed*dfdRdSZ2e3d*e4dTfdUdVZ5e6ed*efdWdXZ7e6dYdZZ8Z9S)aFabricMegatronStrategyz' Fabric strategy for Megatron. NFrr.popenTtensor_model_parallel_sizepipeline_model_parallel_size$virtual_pipeline_model_parallel_size$pipeline_model_parallel_comm_backend"microbatch_group_size_per_vp_stagecontext_parallel_sizesequence_parallelexpert_model_parallel_sizemoe_extended_tpexpert_tensor_parallel_size"encoder_tensor_model_parallel_size$encoder_pipeline_model_parallel_size data_samplerr- acceleratorparallel_devicescluster_environment checkpoint_io precisionmegatron_callbacksddpprocess_group_backendtimeout start_method)r2spawnfork forkserverno_ddp_communication_hookoutput_data_idxpipeline_dtypeinit_model_paralleluse_tp_pp_dp_mapping#num_distributed_optimizer_instancesnccl_communicator_config_pathkwargsreturnc s*tjd||||||||d|t|_| |_||_||_||_|dur(|n||_||_ ||_ | |_ | |_ ||_ ||_| |_| |_||_||_||_||_||_||_t|_|ra|j|||_| |_t||_|dkrxt|_dSt|tr||_dS|dkrd|_d|_dStd|)N)r@rArBrCrDrGrHrIr.r/FzInvalid DDP type: ) super__init__r*rEr?r3r4r6r7r8r:r<r;r5r9r=r>rO_init_model_parallelrQrRrSrMaddrNr'enable_nvidia_optimizations_ddpr ddp_config isinstance ValueError) selfr3r4r5r6r7r8r9r:r;r<r=r>r?r@rArBrCrDrErFrGrHrIrMrNrOrPrQrRrSrT __class__rVT/home/ubuntu/.local/lib/python3.10/site-packages/nemo/lightning/fabric/strategies.pyrXHsd"       zFabricMegatronStrategy.__init__csd||jdus Jtj|j|j|j|jdt t j |jt dS)N) world_size global_rank local_rankparallel_config)_set_world_ranksrBr'init_parallel_ranksrdrerf parallelismrW_setup_distributedtorchcuda set_devicerPr`rarVrcrks  z)FabricMegatronStrategy._setup_distributed datamodulecCs<||jst|dr|j|_|jr|j|j|S)z) Process the datamodule. r?)setupr?hasattrrBre)r`rprVrVrcprocess_datamodules z)FabricMegatronStrategy.process_datamodule dataloadercCs:|jr |j|}t|jd}|t|dt||S)z> Process the dataloader. Returns an iterator. rNmax_size_cycle)r?transform_dataloader"_MegatronDataLoaderIterDataFetcherrNrqriter)r`rtoutputrVrVrcprocess_dataloaders   z)FabricMegatronStrategy.process_dataloader?modeloptimizer_configno_weight_decay_cond scale_lr_condlr_multcCs>t|jdr |j|}|jdusJdtj|||||dS)z/ Setup the Megatron optimizer. convert_configNz-Learning rate must be set in optimizer config)rrr)rrrDrlrr'setup_megatron_optimizer)r`r}r~rrrrVrVrcrs z/FabricMegatronStrategy.setup_megatron_optimizer optimizercCs(t|jdr |j|}|j||S)zJPass the optimizer to the precision-plugin if needed & add it as callback.setup_optimizer)rr _precisionrrErZ)r`rrVrVrcrs   z&FabricMegatronStrategy.setup_optimizermodulec s ddlm}t||jd}t|jdr|jj}t|jdr1|j|||j r1|j|j t|dr:| t ||j|j t |jt|j |d}|jrR||jr\|j|j|j sddlm}dd lm}|}|jdurz||jd <t|}|jrd |_| dt!|S|S) zL Setup the torch module. Returns a MegatronParallel object. r)get_model_configNconvert_modulerconfigure_model)precision_pluginvp_sizecpur]convert_module_fn)mpu)AppState process_groupF)"megatron.core.utilsrr'set_model_parallel_attributesrjrrrDrrr]rr+r5r^r@rrYrPr? callbacksrZ megatron.corer nemo.utilsrmodel_parallel_sizeget_data_parallel_group _ddp_kwargsrW setup_modulerMrequire_backward_grad_syncregister_comm_hookr") r`rrrmegatron_parallelrr app_statedist_data_parallelrarVrcrsF           z#FabricMegatronStrategy.setup_module empty_initcCsD|j}|}t}|r|td|||||S)zK Get the context manager used for initializing the module. meta)rDmodule_init_contextmegatron_contextr enter_contextrldevice)r`rprecision_init_ctxmodule_sharded_ctxstackrVrVrcr3s   z*FabricMegatronStrategy.module_init_contextcCsdS)z0 Move the module to the device. NrV)r`rrVrVrcmodule_to_deviceDsz'FabricMegatronStrategy.module_to_devicepathstatestorage_options filter_dictcCs(|j||pid}|jj|||ddS)aSave model, optimizer, and other state as a checkpoint file. Args: path: A path to where the file(s) should be saved state: A dictionary with contents to be saved. If the dict contains modules or optimizers, their state-dict will be retrieved and converted automatically. storage_options: Additional options for the ``CheckpointIO`` plugin filter: An optional dictionary containing filter callables that return a boolean indicating whether the given item should be saved (``True``) or filtered out (``False``). Each filter key should match a state key, where its filter will be applied to the ``state_dict`` generated. )filter) checkpointrrN)"_convert_stateful_objects_in_staterCsave_checkpoint)r`rrrrrVrVrcrJsz&FabricMegatronStrategy.save_checkpointstrictcCsht|tr tdtji}t|tr||d<n>|r9|d|d<d|vr8tj |d|ddd|d<n!| D]}t|trK||d<q=t|trYtj |dd|d<q=|j j ||d}t|trq|j |||diSt|||d| D]/\}}||vrqt|trt|tr|j ||||dq|||q||||<q|S) z& Load the checkpoint. zIOptimizer loading is not supported, pass it as a dict including the model state_dictrT) is_loading)sharded_state_dict)rrrr)r^r$NotImplementedErrorrlrm empty_cacher#rr'optimizer_sharded_state_dictitemsrCload_checkpointload_module_state_dictrkeyscopyrpopload_state_dict)r`rrrrobjrnamerVrVrcrasB          z&FabricMegatronStrategy.load_checkpointrcCstj|||ddS)z- Load the module state dict. rN)r'load_model_state_dict)r`rrrrVrVrcrsz-FabricMegatronStrategy.load_module_state_dict)NNNc#sfddlm}|jfdd}||_|jj}|jj}d|j_d|j_dV|_||j_||j_dS)z/ Context manager for Megatron. r)transformer_enginecs|}d|d<|S)zForces device to metarrrV)crToriginalrVrc_get_extra_te_kwargs_metaszJFabricMegatronStrategy.megatron_context.._get_extra_te_kwargs_metaFTN)megatron.core.extensionsr_get_extra_te_kwargsrjperform_initializationuse_cpu_initialization)r`_ter_orig_perform_initialization_orig_use_cpu_initializationrVrrcrs   z'FabricMegatronStrategy.megatron_contextcCs4|jdur t|_|jSt|jtrt|j_|jS)z( Get the checkpoint IO. N)_checkpoint_ior)r^rrCrorVrVrcrCs  z$FabricMegatronStrategy.checkpoint_iocCsddlm}|did|jd|jd|jd|jd|jd|jd |jd |j d |j d |j d |j d|j d|jd|jd|jd|jS)z- Get the parallelism config. r)ParallelismConfigr3r4r6r5r7r8r9r:r<r;r=r>rOrQrRrSNrV)3nemo.lightning.pytorch.strategies.megatron_strategyrr3r4r6r5r7r8r9r:r<r;r=r>rOrQrRrS)r`rrVrVrcrjsD      z"FabricMegatronStrategy.parallelismrUN)NNr|N)NN)NT)T):__name__ __module__ __qualname____doc__rintrstrboolrr rlrrrrr*r DDPLiteralrrr dtyperrXr&rkrrsr%r r{r+rrfloatr$rrr#rrrr!rrr rrr rrr rpropertyrCrj __classcell__rVrVrarcr0CsN        !Y  ; "  3  r0c sbeZdZdddedededdffdd Zdfd d Zded fd dZdfdd Z Z S)rxFruargsrNrTrUNcs.tj|i|||_d|_d|_d|_dS)Nr)rWrXrN_batch _batch_idx_dataloader_idx)r`rNrrTrarVrcrXs  z+_MegatronDataLoaderIterDataFetcher.__init__cs"ttt||jd|_|S)Nru)rW__iter__ry_DataFetcherWrapperrNiterator_wrapperrorarVrcrs z+_MegatronDataLoaderIterDataFetcher.__iter__rcCs|jrt|jSr)done StopIterationrrorVrVrc__next__sz+_MegatronDataLoaderIterDataFetcher.__next__cs td|_d|_d|_dS)z) Reset the data fetcher. Nr)rWresetrrrrorarVrcrs  z(_MegatronDataLoaderIterDataFetcher.reset)rUrxr) rrrrrrXrr rrrrVrVrarcrxs $rxcs|eZdZ ddededdfddZedefdd Zedefd d Z ede efd d Z eddZ fddZ ZS)rF data_fetcherrNrUNcCs||_||_dSr)rrN)r`rrNrVrVrcrXs z_DataFetcherWrapper.__init__cC|jjS)z4 Check if the data fetcher is done. )rrrorVrVrcrz_DataFetcherWrapper.donecCr)z7 Check if the data fetcher is fetched. )rfetchedrorVrVrcrrz_DataFetcherWrapper.fetchedcCr)z5 Get the length of the data fetcher. )rlengthrorVrVrcrrz_DataFetcherWrapper.lengthcCr)z& Get the data config. )r data_configrorVrVrcrrz_DataFetcherWrapper.data_configcsJ|j}|jrttt|\}}}||_||_||_|j s |S|||fSr) rrrrWrxrrrrrN)r`fetcherbatch batch_idxdataloader_idxrarVrcr$s z_DataFetcherWrapper.__next__)F)rrrrxrrXrrrrrrrrrrVrVrarcrs$  rstrategyrUcCstdid|jd|jd|jd|jd|jd|jd|jd|jd |j d |j d |j d |j d |j d|jd|jd|jd|jd|jS)z? Convert the Megatron strategy to the Fabric strategy. r3r4r6r5r7r8r9r:r<r;r=r>rOrQrFrGrHrINrV)r0r3r4r6r5r7r8r9r:r<r;r=r>rOrQr\rG_timeout _start_method)rrVrVrcconvert_megatron_strategy4sJ     r)P contextlibrrdatetimertypingrrrrr r r r r rrrllightning.fabric.acceleratorsr)lightning.fabric.accelerators.acceleratorr5lightning.fabric.plugins.collectives.torch_collectiver9lightning.fabric.plugins.environments.cluster_environmentr)lightning.fabric.plugins.io.checkpoint_ior"lightning.fabric.plugins.precisionrlightning.fabric.strategiesr$lightning.fabric.strategies.strategyr lightning.fabric.utilities.typesrrlightning.pytorchr lightning.pytorch.loops.fetchersr$lightning.pytorch.plugins.io.wrapperr+lightning.pytorch.utilities.combined_loaderrmegatron.core.distributedrmegatron.core.optimizerrr r!;torch.distributed.algorithms.ddp_comm_hooks.debugging_hooksr"torch.nnr# torch.optimr$torch.utils.datar%typing_extensionsr&nemo.lightningr' nemo.lightning.fabric.conversionr(nemo.lightning.io.plr) nemo.lightning.megatron_parallelr*r+!nemo.lightning.pytorch.strategiesr,+nemo.lightning.pytorch.plugins.data_samplerr-rr0rxrregisterrrVrVrVrcsR 4                        #5