o }oi@spddlZddlZddlZddlZddlZddlZddlZddlmZddl m Z m Z ddl m Z ddlmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZddl Z ddl!Z ddl"m#Z#ddl$m%Z%ddl&m'Z'dd l(m)Z*dd l(m+Z+dd l,m-Z-dd l.m/Z/dd l m0Z0m1Z1ddl2m3Z3z ddl4m5Z5dZ6Wn e7ydZ6Ynwede0ee8e0fee0Z9ede1j:dZ;edZm?Z@eGdddee9ZAdee9de9fddZBde1j:de j0fddZCdd ZDGd!d"d"e1jEee;ZFGd#d$d$ZGd%edefd&d'ZHGd(d)d)e*ZIGd*d+d+ZJe Gd,d-d-ee;e9fZKGd.d/d/ZLed0ZMGd1d2d2e1j:ee9eMfZNeGd3d4d4eZOeGd5d6d6eZPdee1j:deQfd7d8ZRdee1j:deQfd9d:ZSdeTfd;d<ZUd=edeefd>d?ZVGd@dAdAeNZWGdBdCdCeWZXdDe0dEe0fdFdGZYe dHdIZZe [dMdKdLZ\dS)NN) defaultdict)contextmanager nullcontext) dataclass) TYPE_CHECKINGAnyCallableDictGenericIterableIteratorListMappingOptionalProtocolSequenceTupleTypeVarUnioncastruntime_checkable) TrainerFn)move_data_to_deviceparallel_state)DistributedDataParallel)DistributedDataParallelConfig)OptimizerConfig)TransformerConfig)Tensornn)override)FullyShardedDataParallelTFDataTModelT)boundTc@s4eZdZdedefddZdejdejfddZdS) PrecisionPluginProtocoldatareturncCdSNselfr(r,r,T/home/ubuntu/.local/lib/python3.10/site-packages/nemo/lightning/megatron_parallel.py convert_inputNz%PrecisionPluginProtocol.convert_inputoutputcCr*r+r,)r.r2r,r,r/convert_outputPr1z&PrecisionPluginProtocol.convert_outputN)__name__ __module__ __qualname__r#r0torchrr3r,r,r,r/r'Lsr'dataloader_iterr)cCsJtdkr tdt|}t|trt|dkr|d}t|tj S)ai Moves the data to a device. In this case we unpack the dataloader iterator. There may be a wrapper on the dataloader iter from here: https://github.com/NVIDIA/NeMo/blob/main/nemo/lightning/fabric/strategies.py#L441. This will not subset the data for your with context parallel so please override this function if you want to use context parallel. Examples: If the dataloader_iter returns: [Tuple[, , ]] -> move to device If the dataloader_iter returns: [, ] -> move to device Returns: DataT: The data moved to the device. zDefault data step is being used in a context parallel environment.Please define your own data step that appropriately slices the data for context parallel.r) rget_context_parallel_world_size ValueErrornext isinstancetuplelenrr7cudacurrent_device)r8batchr,r,r/default_data_stepSs rDmodelcOs||g|Ri|Sr+r,)rErCargskwargsr,r,r/default_forward_stepssrHcCstd\}}t|ddr6dd|D}t|dkr|dn|}t|ddr6d d|D}t|dkr4|dn|}||fS) N)NNoverlap_grad_reduceFcSg|]}|jqSr,)no_sync.0 model_chunkr,r,r/ {z%extract_ddp_funcs..r9ralign_grad_reducecSrJr,)start_grad_syncrLr,r,r/rO~rP)getattrr@) ddp_configpipeline no_sync_funcgrad_sync_funcr,r,r/extract_ddp_funcsws  rXcseZdZdZ          dLdeeeefdeededdee e e ge fd ee ee ge fd ee egd fd ee d eedeededee egejfddffdd Z         dMdee e e ee e fdedee e e ge fd ee ee ge fd eddee dee dee dee dedej fddZ      dNde dee e e ge fd ee ee ge fd eddee dee dee defdd Z       dOde dee e e ge fd ee ee ge fd eddee dee dee dee defd!d"Z       dOde dee e e ge fd ee ee ge fd eddee dee dee dee defd#d$Z       dOde dee e e ge fd ee ee ge fd eddee dee dee dee defd%d&Z        dPd'ede dee e e ge fd ee ee ge fd eddee dee dee dedee defd(d)Zde eje geej d*fffd+d,Zd-d.Zd/d0Z d1d2Z!d3d4Z"dQd5d6Z#dej fd7d8Z$dRd:ede%ee&ffd;d<Z'de%ee&ffd=d>Z(d?d@Z)dAdBZ*dCdDZ+e,deeeeffdEdFZ-e,defdGdHZ.e/dIe&de&ffdJdK Z0Z1S)SMegatronParallelas Implements distributed model parallelism that is based on Megatron-LM. This supports various forms of parallelism: - tensor-parallelism - pipeline-parallelism - virtual pipeline parallelism - expert parallelism - sequence parallelism Attributes ---------- pipeline (Union[nn.Module, Iterable[nn.Module]]): The sequence of modules that constitute the pipeline. precision_plugin (Optional[PrecisionPluginProtocol]): An optional plugin for managing precision-specific operations. callbacks (CallbackConnector): A connector for managing and invoking callbacks. data_step (Callable[[Iterator[DataT]], DataT]): A function that takes an iterator over the data and returns the next batch. forward_step (Callable[[nn.Module, DataT], Tensor]): A function that defines the forward pass of a model. loss_reduction (Optional[Callable[[nn.Module], MegatronLossReduction]]): An optional function that defines how the loss is reduced. vp_size (Optional[int]): Virtual pipeline parallel size. ddp_config (Optional[DistributedDataParallelConfig]): An instance of Megatron core's DistributedDataParallelConfig which controls the Megatron DDP configuration. fsdp (Optional[str]): Whether model should run Torch FSDP2 instead of DDP, select from ["megatron", "torch"]. Defaults to None. cpu (bool): Whether model should reside on CPU. convert_module_fn (Optional[Callable[[ModelT], nn.Module]]): An optional function to apply to the model parameters after initialization. Examples -------- >>> from torch import nn >>> from megatron_ext.megatron_parallel import MegatronParallel >>> model = nn.Sequential(nn.Linear(10, 10), nn.ReLU(), nn.Linear(10, 5)) >>> megatron_model = MegatronParallel(model) >>> print(megatron_model) MegatronParallel( (0): Linear(in_features=10, out_features=10, bias=True) (1): ReLU() (2): Linear(in_features=10, out_features=5, bias=True) ) References ---------- Shoeybi, M., Patwary, M., Puri, R., LeGresley, P., Casper, J., & Catanzaro, B. (2019). Efficient Large-Scale Language Model Training on GPU Clusters Using Megatron-LM. arXiv preprint arXiv:1909.08053. NFrUprecision_plugin callbacksCallbackConnector data_step forward_steploss_reductionMegatronLossReductionvp_sizerTfsdpcpuconvert_module_fnr)c sddlm} t|tjrt|} n t|tjr|g} n|} |durRt| dkrR| dkrRddl m }t d|D]}| | d}t |drL|j|d| |q8t| ||_| |_|pbt|_|pgt|_|plt|_||_||_| |_| |_||_dS)Nrrr9)ioconfigure_model)vp_stage) megatron.corerr>r ModuleListlistModuler@&get_pipeline_model_parallel_world_sizenemo.lightningrerangereinithasattrrfappendsuper__init__rZ_cpur\r[rDr]rHr^r_rTrbrdra)r.rUrZr[r]r^r_rarTrbrcrdr _pipelinerei_model __class__r,r/rss2             zMegatronParallel.__init__Tr( forward_onlyz!MegatronLossReduction[DataT, Any] seq_lengthmicro_batch_sizenum_microbatchesstep_iwrap_forward_stepc  Cs|p|j} |p |j} i} | r|p|j}|j| || | d}n| }tj|||||||| d}|| d<|jd|}|jjd|d|}|jjd||d|rq|jjd ||dt | t r`| |j } | |}|jjd || ||d n t jd t jd }|jjd|||d|S)aThe method performs the forward pass of the model. This method is responsible for executing the forward pass of the model. If `forward_only` is set to False, During the execution, it invokes various callbacks at different stages of the operation. For more info about that see [CallbackConnector]. Args: data (Union[DataT, Iterator[DataT], List[Iterator[DataT]]]): The input data for the model. forward_only (bool, optional): If True, only perform the forward pass. Defaults to True. data_step (Optional[Callable[[Iterator[DataT]], DataT]], optional): Function to process the data. Defaults to None. forward_step (Optional[Callable[[nn.Module, DataT], Tensor]], optional): Function to perform the forward pass. Defaults to None. loss_reduction (Optional[MegatronLossReduction[DataT, Any]], optional): Function to reduce the loss. Defaults to None. seq_length (Optional[int], optional): Sequence length for the model. Defaults to None. micro_batch_size (Optional[int], optional): Size of the micro batch. Defaults to None. num_microbatches (Optional[int], optional): Number of microbatches. Defaults to None. wrap_forward_step (bool, optional): If True, wrap the forward step function. Defaults to True. Returns ------- torch.Tensor: The output tensor from the forward pass. )r^r]r_context)rzr|r}r{r~stepon_megatron_step_starton_megatron_microbatches_start)ron_megatron_microbatches_end)rmicrobatch_outputs%on_megatron_reduce_microbatches_start#on_megatron_reduce_microbatches_end)rr_rreduceddeviceon_megatron_step_end)rrr)r^r_r]wrapped_forward_step MegatronStepinferr[transform_eventeventr>_ModuleStepFunctionmodulereducer7tensorrArB)r.r(rzr]r^r_r{r|r}r~r _forward_step_loss_reduction_forward_context _data_stepforward_step_funcrrrr,r,r/forwardsX &       zMegatronParallel.forwardc Ks$|jd|f||||||dd|S)NtrainingF)r]r^r_r{r|r}rz_step) r.r(r]r^r_r{r|r}rGr,r,r/ training_stepDs  zMegatronParallel.training_stepc K&|jd|f|||||||dd| S)N validationTr]r^r_r{r|r}r~rzr r.r(r]r^r_r{r|r}r~rGr,r,r/validation_step\  z MegatronParallel.validation_stepc Kr)NtestTrrrr,r,r/ test_stepvrzMegatronParallel.test_stepc Kr)NpredictTrrrr,r,r/ predict_steprzMegatronParallel.predict_step step_typec  Ks|t|j|dstd|d|pt|j|} |p"t|j|} |p+t|j|}|jd|| | ||||| | d | S)Nrzself.module must have a `z _step` method) r(r]r^r_r{r|r}rzr~r,)rprAttributeErrorrfrom_data_stepfrom_forward_stepfrom_loss_reductionr)r.rr(r]r^r_r{r|r}rzr~rGrrrr,r,r/rs$ zMegatronParallel._stepMegatronCallbackProtocolcs0ddlmtfdd}|S)a@The method wraps the forward step function and returns a callable. The output is a forward_step function in the form of: https://github.com/NVIDIA/Megatron-LM/blob/main/pretrain_gpt.py#L129 Args: forward_step (Callable): The forward step function to be wrapped. loss_reduction (Callable): The loss reduction function. context (Dict): The context dictionary. data_step (Callable): The data step function. Returns ------- Callable: The wrapped forward step function. rrcsttr |}n}||}d}ttr|}n}ttr*|}n}jjd|||djrKjdt|jdddrKj|}|||}j ||||djroj dt|jdddroj |}jjd||||d ||fS) Nron_megatron_microbatch_start)rrCforward_callbackFrgignore_virtualrg)rCrEforward_moduleron_megatron_microbatch_end)rrCr2r) r>rr[rrZis_pipeline_first_stagerSrr0 _setup_moduleis_pipeline_last_stager3)r8rErrCrrr output_tensorrr]r^r_rr.r,r/wrapped_forward_step_funcsR           zHMegatronParallel.wrapped_forward_step..wrapped_forward_step_func)rhr functoolswraps)r.r^r_r]rrr,rr/rs 9z%MegatronParallel.wrapped_forward_stepc Cszddlm}ddlm}|D]y}|js"tr|jdkr"|tj | D]}||q&t |drDt |dsDt |j drC|j j |_ n |r|dkrddlm}tt|}tt|}d|d |d |}||||kr|d |d ||d dq|jr|z|jjjtjks|WdSWdSt y} zdt!| vr| WYd} ~ dSd} ~ ww)Nrr)8set_defaults_if_not_set_tensor_model_parallel_attributesmegatronrfset_input_tensorloggingzC > number of parameters on (tensor, pipeline) model parallel rank (z ,z): z# > number of trainable parameters: z (z.2%z of total)zis not attached to a `Trainer`)"rhr$megatron.core.tensor_parallel.layersrrtHAVE_CUSTOM_FSDPrbrAr7rB parametersrprrmodel_parallel_is_initializedget_data_parallel_rank nemo.utilsr_calc_number_of_paramsrj _calc_number_of_trainable_paramsget_tensor_model_parallel_rank get_pipeline_model_parallel_rankinfordapply_convert_module_fntrainerstatefnrTESTINGinit_ddp RuntimeErrorstr) r.rr model_moduleparamr num_paramsnum_trainable_paramsmsger,r,r/init_model_parallelsX             z$MegatronParallel.init_model_parallelcCs(tt|D] }|||||<qdSr+)rnr@rd)r.rvr,r,r/rPsz(MegatronParallel.apply_convert_module_fnc Cs,t|jtsdSddlm}ddlm}ddlm}t |D]\}}|j }t dd| Dr2t ntj}d}t|drHt|jjtrH|jjj}|dkpM|} |z|||} tr|jd krt| tsdd lm} t|jd ds|t|jd d | d t|jdd rt|jdd| d|jjsJd|jjsJdt|j|j|| d} nt| tst|j|j||jd d| | d} n| } Wdn1swY| |_ | j!|_!t"#|j$j%|j$j&|j$j'|j$j(|j$j)} | |_*|j*j+,|j$j+t-|j._$qt/|j|\}}|D] }||j_0||j_1qdS)Nrr) Float16Module) unwrap_modelcss|]}|jVqdSr+) requires_gradrMxr,r,r/ bsz,MegatronParallel.init_ddp..Foptimrruse_custom_fsdpTz=Setting module.config.use_custom_fsdp to True for MCore FSDP.gradient_accumulation_fusionzKSetting module.config.gradient_accumulation_fusion to False for MCore FSDP.z,Custom FSDP is not enabled in module.config.z)Custom FSDP is not enabled in ddp_config.)disable_bucketingwith_context_parallel)data_parallel_groupexpert_data_parallel_groupr)2r>rTrrhr megatron.core.transformer.modulernemo.utils.model_utilsr enumeraterallrrr7 enable_gradrprconfigr(overlap_param_gather_with_optimizer_steprrbr"rrrSsetattrwarningrDDPget_data_parallel_group%get_data_modulo_expert_parallel_groupbufferstypes FunctionType __getattr____code__ __globals__r4 __defaults__ __closure__original_getattr__dict__update getattr_proxyryrXrVrW)r.rrrmodel_chunk_idxrNrinit_ddp_contextrrunwrapped_moduler dist_modulerrVrWr,r,r/rTs               &  zMegatronParallel.init_ddpcCs"|D] }t|dr|j|j_qdS)Nr)rprryr)r.rNr,r,r/ teardown_ddps   zMegatronParallel.teardown_ddpc sFt|dr!t|jjfdd|D}|jdi|dSdS)Nsetupci|] \}}|vr||qSr,r,rMkv setup_argsr,r/ z2MegatronParallel._setup_module..r,)rpinspectgetfullargspecrrFitems)r.functionrG setup_kwargsr,r r/rs zMegatronParallel._setup_modulecsF|j|fi|t|jfdd|D}||i|}|S)Ncr r,r,r  call_argsr,r/rrz1MegatronParallel._call_module..)rrrrFr)r.rrFrG call_kwargsrr,rr/ _call_modules  zMegatronParallel._call_moduleprefixcCsRi}t|D] \}}|jdur||}||d|<q||}||q|S)aI Creates the sharded state dict which is used by dist_checkpoint to save the sharded tensors to disk. When given the sharded_stated_dict, dist_checkpoint.load will load the tensors corresponding to self.state_dict(). The sharded tensor mapping is defined in the GPTModel class from mcore. Nmodel_)rra_module_sharded_state_dictr)r.rsharded_state_dictindexrmodule_sharded_state_dictr,r,r/rs    z#MegatronParallel.sharded_state_dictcOs`t|dr |j|i|St|dr,d|dddg}|j|jg|Rd|i|Std)Nrrfrrzmodule.z!Could not find sharded state dict)rprjoinpoprrr<)r.rrFrGrr,r,r/rs  z+MegatronParallel._module_sharded_state_dictcC4|D]}|j}t|tst|tsJ|qdSr+)rr>rr"enable_forward_pre_hookr.rErNr,r,r/r$  z(MegatronParallel.enable_forward_pre_hookcCr#r+)rr>rr"disable_forward_pre_hookr%r,r,r/r'r&z)MegatronParallel.disable_forward_pre_hookcCs8|D]}|j}t|tst|tsJ|jddqdS)NT) force_sync)rr>rr"start_param_syncr%r,r,r/force_param_syncs z!MegatronParallel.force_param_synccCst|dkr |dSt|S)Nr9r)r@rjr.r,r,r/rUs zMegatronParallel.pipelinecCs|dS)Nrr,r+r,r,r/rzMegatronParallel.moduleitemc szt|WStyDt|dkr!td|jjd|dzt|j|d|WYStyCtd|jjd|dww)Nr'' object has no attribute 'z' and contains no modules) rrrrr@ryr4rS_modules_get_abs_string_indexr.r-rxr,r/rs   zMegatronParallel.__getattr__) NNNNNNNNFN) TNNNNNNNT)NNNNNN)NNNNNNN)NNNNNNTNr)N)r)2r4r5r6__doc__rr$r rr'rr r#rintrrboolr rkrsr r7r STEP_OUTPUTrrrrrrrrrrrrrr rrrr$r'r*propertyrUrr!r __classcell__r,r,rxr/rYs6     1    a           " P4X    rYc@seZdZdZddededefddZedd d ed edfd d Z edd d ed edfddZ edd d ed edfddZ de j fddZdS)rz This class acts as a bridge between Megatron core's lower-level functional API and PTL's object-oriented API, making it possible to use PTL-compatible functions in Megatron core. Fname is_property includes_selfcCs||_||_||_dSr+)r:r;r<)r.r:r;r<r,r,r/rss z_ModuleStepFunction.__init__rpl.LightningModulerr)cCs.|ddfD] }t||rt|SqdS)Nrr]rprclsrrfn_namer,r,r/rs   z"_ModuleStepFunction.from_data_stepcCsddlm}|jdt|dddr*t||ds!td|dt|dd d S|d d fD]}t||r@t|d d Sq1dS) NrrFrgrrzLightningModule does not have z _step methodT)r<rr^)rhrrrSrpr<r)r@rrrrAr,r,r/rs  z%_ModuleStepFunction.from_forward_stepcCs2|ddfD]}t||rt|ddSqdS)Nrr_T)r;r>r?r,r,r/r.s  z'_ModuleStepFunction.from_loss_reductioncsLt||j|jrttt||jtrSS|jr$fdd}|SS)Ncs|Sr+r,)r.rFattrr,r/wrappedBz-_ModuleStepFunction.__call__..wrapped)rSr:r;r>typer8r<)r.rrDr,rBr/__call__6s  z_ModuleStepFunction.__call__N)FF)r4r5r6r4rr6rs classmethodrrrrr rkrGr,r,r,r/r srr-c Cs~z t|j||WSty>}z(|dkr|z t|j|WWYd}~Sty9td|jjd|dwd}~ww)Nrr.r/)rrryrrrSrr4)r.r-rr,r,r/rJs rc sReZdZ ddededejjdeffdd Z dd d Z d e d e fd dZ Z S)rFrrTrrc sDttjjfdd|D}tjd||||d|dS)Ncr r,r,r init_parametersr,r/rcrz DDP.__init__..)rrTrrr,)r signatureMcoreDDPrsrrrr)r.rrTrrrGfiltered_kwargsrxrIr/rsWs z DDP.__init__rcKs|jjd||d|dS)N)r keep_varsr,)r state_dict)r.rrNrGr,r,r/rOlszDDP.state_dictr-r)cCs t||Sr+)rr2r,r,r/ro zDDP.__getattr__)F)rF)r4r5r6rrr7r rkr6rsrOrrr9r,r,rxr/rVs rc@speZdZdZddddZdddZdeddfd d Zded edefd d Z dddZ dddZ de fddZ dS)r\a A connector for managing and invoking callbacks. The CallbackConnector class in the MegatronParallel module is used to manage and invoke callbacks during the execution of the model. Callbacks are functions that are called at specific stages of the model execution, allowing you to hook into the model's operation for logging, debugging, or other purposes. The CallbackMethods class defines the names of the callback methods that can be used. These methods are: - `on_megatron_step_start` - `on_megatron_microbatch_start` - `on_megatron_microbatch_callback` - `on_megatron_microbatch_end` - `on_megatron_reduce_microbatches_start` - `on_megatron_reduce_microbatches_end` - `on_megatron_log_step_end` - `on_megatron_step_end` Each of these methods corresponds to a specific stage in the model's operation. You can define these methods in your callback functions to perform specific actions at these stages. There is no need for the class to be a subclass of a specific parent class. As long as the class contains the methods outlined above, it can be used as a callback. Nr)cCs tt|_|r|j|dSdSr+)rrjr[add)r.r[r,r,r/rss zCallbackConnector.__init__csdz ddlm}|jWn tyYnwfddttD}|D]3}t|tr>|j D] \}}|j| |q0q$|D]}t ||rVt t ||rV|j||q@q$|S)a4 Adds callback functions to the connector. Parameters ---------- *callbacks : CallbackT One or more callback functions to add. Returns ------- CallbackConnector The CallbackConnector instance to allow method chaining. Nrcs$h|]}|drt|s|qS)on) startswithrp)rMm _pl_callbackr,r/ s$z(CallbackConnector.add..)lightning.pytorchpytorchCallback ImportErrordirCallbackMethodsr>r\r[rextendrpcallablerSrq)r.r[plmegatron_methodscallback event_nameevent_callbacksmethodr,rUr/rQs&    zCallbackConnector.addr:c s|j|gD]y}t||d}t|rt|j}tdd|D}tdd|D}|r;|r;||i|q|rPfdd| D} ||i| q|rdddt ||D} || i|qd dt ||D} fd d| D} || i| qdS) aM Triggers an event and calls all associated callbacks. Parameters ---------- name : str The name of the event to trigger. *args : Any Positional arguments to pass to the callbacks. **kwargs : Any Keyword arguments to pass to the callbacks. Ncs|] }|j|jkVqdSr+)kindVAR_POSITIONALrMpr,r,r/rz*CallbackConnector.event..csrfr+)rg VAR_KEYWORDrir,r,r/rrkc i|] \}}|jvr||qSr,rr sigr,r/r z+CallbackConnector.event..cS&g|]\}}|j|j|jfvr|qSr,rgPOSITIONAL_ONLYPOSITIONAL_OR_KEYWORDrMargrr,r,r/rO z+CallbackConnector.event..cSrrr,rsrvr,r,r/rOrxcrmr,rnr ror,r/rrq) r[getrSr_rrKrvaluesanyrzip) r.r:rFrGrbcallback_methodparamsaccepts_var_argsaccepts_var_kwargsrM filtered_argsr,ror/rs0   zCallbackConnector.eventobjcKsV|j|gD]!}t||d}t|r(||fi|}|dur(t|t|r(|}q|S)a Triggers an event that allows callbacks to transform and return an object. This method applies a series of potential transformations to the input object by calling registered callbacks. Each callback has the opportunity to modify and return a new version of the object. Parameters ---------- name : str The name of the event to trigger. obj : T The object to be potentially transformed by callbacks. **kwargs : Any Additional keyword arguments to pass to the callbacks. Returns ------- T The potentially transformed object. N)r[ryrSr_r>rF)r.r:rrGrbr}resultr,r,r/rs z!CallbackConnector.transform_eventcCs4t|ts tdt}tti|j|j|_|S)a Adds another CallbackConnector's callbacks to this one. Parameters ---------- other : CallbackConnector Another CallbackConnector instance to add. Returns ------- CallbackConnector A new CallbackConnector instance with combined callbacks. Raises ------ ValueError If `other` is not an instance of CallbackConnector. (Can only add CallbackConnector instances)r>r\r<rrjr[)r.other new_connectorr,r,r/__add__ s zCallbackConnector.__add__cCs:t|ts td|jD] \}}|j||q|S)a In-place addition of another CallbackConnector's callbacks. Parameters ---------- other : CallbackConnector Another CallbackConnector instance to add. Returns ------- CallbackConnector The same CallbackConnector instance with combined callbacks. Raises ------ ValueError If `other` is not an instance of CallbackConnector. r)r>r\r<r[rr^)r.rrcrdr,r,r/__iadd__"s zCallbackConnector.__iadd__csRddttD}tfdd|D}|sdS|jD] }|vr&dSqdS)a Check if the given callback object is registered in the CallbackConnector. If the object has none of the methods of CallbackMethods, it returns True. If it has at least one of the methods, it checks if it's inside the CallbackConnector object. Args: callback_object: The object to check for callback methods. Returns ------- bool: True if the callback object is registered, False otherwise. cSs(g|]}ttt|r|ds|qS)__)r_rSr]rS)rMfuncr,r,r/rOIs z2CallbackConnector.__contains__..c3s|]}t|VqdSr+)rp)rMrecallback_objectr,r/rPz1CallbackConnector.__contains__..TF)r\r]r{r[rz)r.rcallback_methodshas_any_callback_methodrdr,rr/ __contains__;szCallbackConnector.__contains__r+r3)r)r\)r4r5r6r4rsrQrrr&rrrr6rr,r,r,r/r\ss  $1 ! r\c@seZdZUdZeeed<eee ee e efed<e ed<e ed<dZ eeed<dZeeed<dZeeed <dZeeed <dZeeed <e    d+deedede de deedeed eed eed d fddZd e efddZdeee ee e efd e e efddZeded eefddZeded eefddZeded eefddZed eee effddZed,ddZed-d d!Ze j!d.d#d$Z"ed ee dffd%d&Z#d e$e e eeeffd'd(Z%e j!d e fd)d*Z&dS)/ra Represents a single step in the Megatron model's training or inference process. This class encapsulates all the necessary information and logic for executing a single step (forward pass, and optionally backward pass) in the Megatron model. It handles data preparation, model execution, and provides utilities for inferring batch sizes and sequence lengths. Attributes: pipeline (MegatronParallel[ModelT]): The Megatron parallel model pipeline. data (Union[DataT, Iterator[DataT], List[Iterator[DataT]]]): Input data for the step. forward_step_func (Callable): Function to perform the forward step. forward_only (bool): If True, only perform forward pass (no backward pass). micro_batch_size (Optional[int]): Size of each micro-batch. seq_length (Optional[int]): Sequence length for the current step. num_microbatches (Optional[int]): Number of micro-batches in this step. decoder_seq_length (Optional[int]): Sequence length of decoder (used only in encoder-decoder style models) for the current step. Type Parameters: ModelT: The type of the model being used. DataT: The type of the input data. rUr(rrzNr|r{r}r~decoder_seq_lengthr)zMegatronStep[ModelT, DataT]c CsL|dur |jr |jj}||||||p|||p|||p"|||dS)a Creates a MegatronStep instance, inferring missing parameters if possible. This method attempts to infer the micro_batch_size, seq_length, and num_microbatches from the provided data if they are not explicitly specified. Args: pipeline (MegatronParallel[ModelT]): The Megatron parallel model pipeline. data (DataT): Input data for the step. forward_step_func (Callable): Function to perform the forward step. forward_only (bool): If True, only perform forward pass (no backward pass). micro_batch_size (Optional[int]): Size of each micro-batch. seq_length (Optional[int]): Sequence length for the current step. num_microbatches (Optional[int]): Number of micro-batches in this step. step_i (Optional[int]): Step index for the current step. Returns: MegatronStep[ModelT, DataT]: An instance of MegatronStep with inferred parameters. N)rUr(rrzr|r{r}r~)r global_stepinfer_micro_batch_sizeinfer_seq_lengthinfer_num_microbatches) r@rUr(rrzr|r{r}r~r,r,r/rs   zMegatronStep.inferc Csv|jdur td|jdurtd|jdurtd|\}}|p%|j}|j|j||j|j||j|j|j |j d S)a  Executes the Megatron step. This method performs the forward (and optionally backward) pass using the configured forward_backward_func. It ensures all necessary parameters are set before execution. Returns: List[Any]: The output of the forward_backward_func, typically containing loss values and other relevant information. Raises: ValueError: If any of num_microbatches, seq_length, or micro_batch_size is not set. Nznum_microbatches is not setzseq_length is not setzmicro_batch_size is not set) r data_iteratorrEr}r{r|rzradjust_tensor_shapes_fn) r}r<r{r| get_data_iterator_and_seq_lengthforward_backward_funcrrErzrr)r.rr{r,r,r/rGs&     zMegatronStep.__call__cCs^t|tr t|j|St|tr"tdd|Dr"tttt|Sttttt |ggS)a9 Converts the provided data into a list of iterators. This method is used to convert the input data into a list of iterators that can be used for data parallelism in the Megatron model. The input data can be a single data item, an iterator, or a list of iterators. Args: data (Union[DataT, Iterator[DataT], List[Iterator[DataT]]]): The input data to be converted into a list of iterators. Returns: List[Iterator[DataT]]: A list of iterators created from the input data. css|]}t|tVqdSr+)r>r rMr-r,r,r/rrz5MegatronStep.to_data_iterator_list..) r>r _make_data_iterator_listrErjrrr r#iterr-r,r,r/to_data_iterator_lists  z"MegatronStep.to_data_iterator_listcCsdt|tr |dSt|tr|tt|St|tt fr0t |dkr0|d}||SdS)a Infers the micro-batch size from the input data. This method attempts to determine the micro-batch size by examining the first dimension of the input data. It handles various data types including Tensors, dictionaries, lists, and tuples. Args: data (DataT): The input data from which to infer the micro-batch size. Returns: Optional[int]: The inferred micro-batch size, or None if it cannot be determined. rN) r>rsizedictrr=rrzrjr?r@r@r(_tensorr,r,r/rs    z#MegatronStep.infer_micro_batch_sizecCsdt|tr |dSt|tr|tt|St|tt fr0t |dkr0|d}||SdS)a Infers the sequence length from the input data. This method attempts to determine the sequence length by examining the second dimension of the input data. It handles various data types including Tensors, dictionaries, lists, and tuples. Args: data (DataT): The input data from which to infer the sequence length. Returns: Optional[int]: The inferred sequence length, or None if it cannot be determined. r9rN) r>rrrrr=rrzrjr?r@rr,r,r/rs    zMegatronStep.infer_seq_lengthcCst|ttttfr dSdS)a Infers the number of micro-batches from the input data. Currently, this method assumes a single micro-batch for common data types. It may need to be extended for more complex data structures or use cases. Args: data (DataT): The input data from which to infer the number of micro-batches. Returns: Optional[int]: The inferred number of micro-batches, or None if it cannot be determined. r9N)r>rr?rjr)r@r(r,r,r/r"sz#MegatronStep.infer_num_microbatchescCs|jjS)z Retrieves the model or list of models from the pipeline. Returns: Union[ModelT, List[ModelT]]: The model or list of models in the pipeline. )rUr+r,r,r/rE5zMegatronStep.modelr=cC|jjS)z Retrieves the PyTorch Lightning module from the pipeline. Returns: pl.LightningModule: The PyTorch Lightning module. )rUrr+r,r,r/ pl_module?rzMegatronStep.pl_module pl.TrainercCr)z Retrieves the PyTorch Lightning trainer from the pipeline. Returns: pl.Trainer: The PyTorch Lightning trainer. )rUrr+r,r,r/rIrzMegatronStep.trainerMegatronStepProtocolcCsddlm}|S)aI Retrieves the forward-backward function for the Megatron model. This property uses Megatron's scheduling to get the appropriate forward-backward function based on the current configuration. Returns: MegatronStepProtocol: The function to perform forward and backward passes. r)get_forward_backward_func))megatron.core.pipeline_parallel.schedulesr)r.rr,r,r/rSs z"MegatronStep.forward_backward_funccCs&ddlm}||j|j|j|j|jS)a| Retrieves the function to adjust send and receive tensor shapes in Megatron-Core's forward pass. Currently only used during non-interleaved pipelining for Distillation. Returns: Union[Callable, None]: The function which takes in tensor shapes and returns updated shapes, or None if not applicable. r),get_tensor_shapes_adjust_fn_for_distillation)+nemo.collections.llm.modelopt.distill.utilsrrEr{r|rrz)r.rr,r,r/rbs z$MegatronStep.adjust_tensor_shapes_fncsd}|jrSt|j}t|trt|dkr|\}d}n |ddd}d|vr2|ddnd}ddlm}|||j d}|rRfd d |D}t |}n|j}d}| |}||fS) aD Converts the provided data into a list of iterators. For finetuning, where sequence length is different for each step, this function also outputs the sequence length of the current batch. Returns: List[Iterator[DataT]]: A list of iterators created from the input data. Fr:TrNtokensr9)get_iterator_k_splitcsg|]}|fqSr,r,)rMd batch_idxdataloader_idxr,r/rOszAMegatronStep.get_data_iterator_and_seq_length..) has_global_batch_samplerr=r(r>r?r@r2nemo.collections.nlp.modules.common.megatron.utilsrr} itertoolschainr)r.has_dataloader_idx batch_datarCr{rr( packed_datar,rr/rws$      z-MegatronStep.get_data_iterator_and_seq_lengthcCs`t|jdddur|jjjjdk}|St|jdddur,ddlm}t|jjj |}|Sd}|S)N datamodulerCpredict_dataloadersr)MegatronPretrainingBatchSamplerF) rSrr data_samplerdataloader_typeLnemo.collections.nlp.data.language_modeling.megatron.megatron_batch_samplersrr>r batch_sampler)r.use_global_batch_samplerrr,r,r/rs  z%MegatronStep.has_global_batch_sampler)NNNN)r)r=)r)r)r)r)'r4r5r6r4rYr$__annotations__rr#r r rr6r|rr5r{r}r~rrHrrrGrrrrr8rErrrcached_propertyrrrrrr,r,r,r/r^sx    +'     "$rc @seZdZdZdedefddZdeddfddZded ed d ddfd d Zded ed d de ddf ddZ dede e ddfddZ dede e ddfddZ dede e dd deejeeejffddf ddZ ddede e deeejeeejffddfddZdS)r]a_ Defines callback methods for various stages of the Megatron model's execution. This class outlines the structure for callbacks that can be implemented to hook into different phases of the Megatron model's training or inference process. Each method represents a specific point in the execution where custom logic can be inserted. rr)cCdS)a Called at the beginning of each Megatron step. This method is invoked before any processing of the step begins. It allows for any necessary setup or initialization for the step. Args: step (MegatronStep): The MegatronStep object representing the current step. Returns: MegatronStep: The potentially modified MegatronStep object. Nr,r.rr,r,r/rs z&CallbackMethods.on_megatron_step_startNcCr)ab Called before processing of microbatches begins. This method is invoked just before the model starts processing the microbatches within a step. It can be used for any preparations needed before microbatch processing. Args: step (MegatronStep): The MegatronStep object representing the current step. Nr,rr,r,r/rs z.CallbackMethods.on_megatron_microbatches_startrCrr`cCr)a Called at the start of processing each microbatch. This method is invoked before the forward pass of each microbatch. It provides access to the current batch data and the loss reduction callback. Args: step (MegatronStep): The MegatronStep object representing the current step. batch (DataT): The current microbatch of data being processed. forward_callback (MegatronLossReduction): The callback for loss reduction. Nr,)r.rrCrr,r,r/rsz,CallbackMethods.on_megatron_microbatch_startr2cCr)aU Called at the end of processing each microbatch. This method is invoked after the forward pass of each microbatch. It provides access to the processed batch, the loss reduction callback, and the output of the forward pass. Args: step (MegatronStep): The MegatronStep object representing the current step. batch (DataT): The microbatch of data that was processed. forward_callback (MegatronLossReduction): The callback for loss reduction. output (Any): The output from the forward pass for this microbatch. Nr,)r.rrCrr2r,r,r/rsz*CallbackMethods.on_megatron_microbatch_endrcCr)a Called after all microbatches in a step have been processed. This method is invoked once all microbatches within a step have been processed. It provides access to the outputs from all microbatches. Args: step (MegatronStep): The MegatronStep object representing the current step. microbatch_outputs (List[Any]): A list of outputs from all processed microbatches. Nr,r.rrr,r,r/rs z,CallbackMethods.on_megatron_microbatches_endcCr)a Called before the reduction of microbatch outputs begins. This method is invoked just before the model starts reducing (e.g., averaging) the outputs from all microbatches. It can be used for any preparations needed before the reduction process. Args: step (MegatronStep): The MegatronStep object representing the current step. microbatch_outputs (List[Any]): A list of outputs from all processed microbatches. Nr,rr,r,r/r sz5CallbackMethods.on_megatron_reduce_microbatches_startr_rcCr)a Called after the reduction of microbatch outputs is complete. This method is invoked after the model has finished reducing the outputs from all microbatches. It provides access to the original microbatch outputs, the loss reduction object, and the final reduced output. Args: step (MegatronStep): The MegatronStep object representing the current step. microbatch_outputs (List[Any]): A list of outputs from all processed microbatches. loss_reduction (MegatronLossReduction): The object used for loss reduction. reduced (Union[torch.Tensor, Dict[str, torch.Tensor]]): The final reduced output. Nr,)r.rrr_rr,r,r/rsz3CallbackMethods.on_megatron_reduce_microbatches_endcCr)a Called at the end of each Megatron step. This method is invoked after all processing for a step is complete. It provides access to the outputs from all microbatches and the final reduced output (if available). Args: step (MegatronStep): The MegatronStep object representing the current step. microbatch_outputs (List[Any]): A list of outputs from all processed microbatches. reduced (Optional[Union[torch.Tensor, Dict[str, torch.Tensor]]]): The final reduced output, if available. This may be None for certain configurations or pipeline stages. Nr,)r.rrrr,r,r/r3sz$CallbackMethods.on_megatron_step_endr+)r4r5r6r4rrrr#rrrr rrrr7rr rrrrr,r,r,r/r]sj      r] ReductionTcspeZdZdfdd ZdddZddZd ed ejde eje ffd d Z e j d ee dejfddZZS)r`r)Ncs$tt|d|_||jdSr+)rrr`rsrCregister_forward_pre_hook_pre_forward_hookr+rxr,r/rsLszMegatronLossReduction.__init__cCs ||_dSr+rC)r.rCr,r,r/rQrPzMegatronLossReduction.setupcCs |jf|Sr+r)r.rrr,r,r/rTs z'MegatronLossReduction._pre_forward_hookrC forward_outcCtdNz Must be implemented by subclass.NotImplementedError)r.rCrr,r,r/rWrEzMegatronLossReduction.forwardlosses_reduced_per_micro_batchcCrrr)r.rr,r,r/rZr,zMegatronLossReduction.reducer3)r4r5r6rsrrr#r7rrrrabcabstractmethodrrr9r,r,rxr/r`Ks ""r`c@s4eZdZdejdeejeeejfffddZdS)rrr)cCr*r+r,)r.rr,r,r/rGar1z!MegatronCallbackProtocol.__call__N) r4r5r6r7rrr rrGr,r,r,r/r_s,rc@sheZdZdddddeeeefdeejjeejjfde de de d e e d e d e d e fd dZ dS)rNF)rrzcollect_non_loss_datarrEr}r{r|rrzrr)c Cr*r+r,) r.rrrEr}r{r|rrzrr,r,r/rGfs zMegatronStepProtocol.__call__)r4r5r6rr r r7r rkr5rr6rjrGr,r,r,r/rds.     rcC t|tsJtdd|DS)NcS"g|] }tdd|DqS)cSsg|]}|qSr,)nelementrir,r,r/rOxz5_calc_number_of_params...sumrrMrr,r,r/rOx"z*_calc_number_of_params..r>rjrrEr,r,r/rurcCr)NcSr)cSsg|] }|jr|qSr,)rnumelrir,r,r/rO~sz?_calc_number_of_trainable_params...rrr,r,r/rO~rz4_calc_number_of_trainable_params..rrr,r,r/r{rrcCs t|tsdStdd|DS)NFcss|] }t|tjjVqdSr+)r> collectionsrr rr,r,r/rsz'is_list_of_iterators..)r>rjr)varr,r,r/is_list_of_iteratorss rrcslt|tr t|dkr |SGfddd|g}t|t|kr4||dt|t|ks#|S)aConvert data iterator into form expected by Megatron. With interleaved pipeline parallelism, Megatron expects a list of one data iterator per model chunk. Each model chunk independently gets data from its data iterator, so we need to interact with the data iterator multiple times for each microbatch step. Instead of incorporating this logic into the data loader, we cache the iterator's output to the first model chunk and reuse it in the other model chunks. r9csHeZdZdZGdddZdefddZfddZd d Zd d Z d S)z1_make_data_iterator_list..CachingIteratorz$Iterator wrapper that caches values.c@s(eZdZdZddZddZddZdS) z7_make_data_iterator_list..CachingIterator.Proxyz{Returns values from caching iterator wrapper. Assumed to never advance past the caching iterator. cSst|_dSr+)queueQueuecacher+r,r,r/rssz@_make_data_iterator_list..CachingIterator.Proxy.__init__cS|Sr+r,r+r,r,r/__iter__z@_make_data_iterator_list..CachingIterator.Proxy.__iter__cSs |jSr+)r get_nowaitr+r,r,r/__next__rPz@_make_data_iterator_list..CachingIterator.Proxy.__next__N)r4r5r6r4rsrrr,r,r,r/Proxys  riteratorcSs||_g|_dSr+)rproxies)r.rr,r,r/rss z:_make_data_iterator_list..CachingIterator.__init__cs|j|jdS)N)rrqrr+CachingIteratorr,r/ make_proxys z<_make_data_iterator_list..CachingIterator.make_proxycSrr+r,r+r,r,r/rrz:_make_data_iterator_list..CachingIterator.__iter__cSs&t|j}|jD]}|j|q|Sr+)r=rrrput)r.valproxyr,r,r/rs  z:_make_data_iterator_list..CachingIterator.__next__N) r4r5r6r4rr rsrrrr,rr,r/rs  rr)r>rjr@rqr)rEritersr,rr/rs  $rc sveZdZddededdffdd Zd eeejfd ejde ejejeeejfffd d Z dejfd dZ Z S)MaskedTokenLossReductionFTr val_drop_lastr)Ncst||_||_dSr+)rrrsrr)r.rrrxr,r/rss  z!MaskedTokenLossReduction.__init__rCrcCst|tr |\}}||d<t||d\}}|jr-|js-|r-|dks(Jdt|}| tj }t | d| dg}||d|ifS)zTaken from: https://github.com/NVIDIA/NeMo/blob/main /nemo/collections/nlp/models/language_modeling/megatron_gpt_model.py#L951-L976 . loss_maskrz!Got NaN loss with non-empty inputr9loss_sum_and_ub_size)r>r?masked_token_lossrrisnanr7 zeros_likeclonedetachtor5catview)r.rCrrloss_sumnum_valid_tokensrr,r,r/rs  "z MaskedTokenLossReduction.forwardcCs|rVd|dvrdd|D}t|}|Sddlm}dd|D}t|dkr4t|jddn tjddgtj d }tj j ||j d d d |d|d }|Stjdtj d S)zTaken from: https://github.com/NVIDIA/NeMo/blob/main /nemo/collections/nlp/models/language_modeling/megatron_gpt_model.py#L535-L552 .avgrcSsg|]}|dqS)rr,rr,r,r/rOrz3MaskedTokenLossReduction.reduce..rcSs$g|]}|dddkr|dqS)rr9rr,rr,r,r/rOs)dimrrTr)groupr9)r7rmeanrhrr@vstackrrrArB distributed all_reducer)r.rrlossrrr,r,r/rs(    zMaskedTokenLossReduction.reduce)FT) r4r5r6r6rsr rr7rrrrr9r,r,rxr/rs  rc sTeZdZdeeejfdeejejfdeejeeejffffdd ZZ S)$MaskedTokenLossReductionWithLossMaskrCrr)cs|\}}||d<t||S)Nr)rrr)r.rCrrrxr,r/rsz,MaskedTokenLossReductionWithLossMask.forward) r4r5r6r rr7rrrr9r,r,rxr/r s r rmaskcCs:|d}|d}t||}|}||fS)zS The function takes as input per-token loss and masks non-required values. r)rfloatr7r)rr lossesrrrr,r,r/r s rccs4ddlm}m}|z dVW|dS|w)Nrclear_aux_losses_tracker&reduce_aux_losses_tracker_across_ranks)'megatron.core.transformer.moe.moe_utilsrrrr,r,r/moe_loss_tracker_ctxs r?cst8t}fdd|D}i}|D]\}}||vr%d||<|||7<q|WdS1s>wYdS)Ncs"i|] \}}||dqS)rz)r r  loss_scaler,r/r*rz,aggregate_moe_loss_stats..r)rr"get_moe_layer_wise_logging_trackerrrr-)rtracker aux_lossestotal_loss_dictr: loss_listr,rr/aggregate_moe_loss_stats&s$r)r)]rcollections.abcrrrrrrr contextlibrr dataclassesrtypingrrrr r r r r rrrrrrrrrr7torch.distributed lightning.pytorch.trainer.statesrlightning.pytorch.utilitiesrrhrmegatron.core.distributedrrLrmegatron.core.optimizerr,megatron.core.transformer.transformer_configrrr typing_extensionsr!%megatron.core.distributed.custom_fsdpr"rr[rr#rkr$r&r7rXrYr`r'rDrHrXrirYrrrr\rr]rr`rrr5rrr6rrrr rrno_gradrr,r,r,r/s  L              ? lU=: