Pib lUddlZddlZddlmZddlmZmZmZmZm Z m Z m Z ddl Z ddl mZddl mZddlmZmZddlmZmZddlmZmZddlmZmZmZmZmZdd l m!Z!dd l"m#Z#dd l$m%Z%dd l&m'Z'dd l(m)Z)m*Z*ddl+m,Z,ddl-m.Z.ddl/m0Z0m1Z1ddl2m3Z3ddl4m5Z5m6Z6ddl7m8Z8e6Z9ej:e;d<e j<Z=dZ>de?de#fdZ@deAfdZBdCde jCdeDde jCfdZEdDde?deAde?fdZFe8d !d"ee?efdeAfd#ZGdEd$ZHe8d%!de eDeDffd&ZId'ejJddfd(ZK dFd'd)d*ee?efd+e jLd,eAd-eAde%f d.ZMd/ejNdejNfd0ZO dGd'd)d1eAd+e e jLd2eAdee?eff d3ZP dHd'd)d4e'd1eAd+e e jLdee?eff d5ZQd'd)d4e'd*ee?efd+e jLddf d6ZR dHd7e?d8ejJd9e e e?deAfd:ZSd;dd<d'e,d=e ee?ejJgeAfd-eAd>eAd?e e!ddf d@ZTd'ejJdAe!dejJfdBZUdS)IN)chain)AnyCallablecastDictListOptionalTuple)nn)CPUOffloadPolicy fully_shard)distribute_tensorDTensor) DTensorSpec TensorMeta)_init_optim_stateget_optimizer_state_dictset_model_state_dictset_optimizer_state_dictStateDictOptions) DeviceMesh)ShardingStrategy)_IncompatibleKeys) Optimizer) NF4Tensorto_nf4)TransformerDecoder)MultiHeadAttention)DeepFusionModelEarlyFusionModel)get_adapter_state_dict) get_device get_logger) deprecated_logFstrategyreturnc,tt|S)zNHelper function to convert sharding strategy strings to ShardingStrategy enum.)getattrr)r&s s/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/torchtune/training/_distributed.py_get_sharding_strategyr+2s #X . ..ctjdd}tjdd}ttjdd}ttjdd}t j}t |o|o |dko|dko|S) a"Check if all environment variables required to initialize torch.distributed are set and distributed is properly installed. This indicates a distributed run. https://pytorch.org/docs/stable/distributed.html#environment-variable-initialization Checks the following conditions: * torch.distributed is available * master port and master address environment variables are set * world size is >1 * rank environment variable is set Returns: bool: True if all of the above conditions hold, False otherwise. MASTER_PORT MASTER_ADDR WORLD_SIZERANKr)osenvirongetintdist is_availablebool)portaddrsizerankavlbs r*is_distributedrA7s :>>- , ,D :>>- , ,D rz~~lA.. / /D rz~~fb)) * *D    D BB$!)B Bd C CCr,tensorsrcc*tjrtjrl|j}tjdkr"|t d}tj||d||S|S)zBroadcasts a tensor from a source to all other processes. Args: tensor (torch.Tensor): torch.Tensor to broadcast. src (int, optional): Source rank. Defaults to 0. Returns: torch.Tensor: Broadcasted tensor. ncclcudaN)rCgroup)r9r:is_initializeddevice get_backendtor" broadcast)rBrCrIs r*_broadcast_tensorrMNs t244     ' 'YYz&1122F v3d3333yy    r, device_typeoffload_ops_to_cpucXtjj}d}||vr||}|r|d|d}|S)aGets the PyTorch Distributed backend based on device type. Args: device_type (str): Device type to get backend for. offload_ops_to_cpu (bool, optional): Flag to check if any operations should be offloaded to CPU. Examples of these kinds of operations are CPU offload for FSDP and asynchronous save for distributed checkpointing. Defaults to False. Example: >>> get_distributed_backend("cuda") 'nccl' >>> get_distributed_backend("cpu") 'gloo' >>> get_distributed_backend("cuda", offload_ops_to_cpu=True) 'cuda:nccl,cpu:gloo' Returns: str: Distributed backend for use in ``torch.distributed.init_process_group``. rE:z ,cpu:gloo)r9Backenddefault_device_backend_map)rNrOrSbackends r*get_distributed_backendrUbsM("&!HG000,[96 557555 Nr,z^The functionality of `init_distributed` is covered by `torch.distributed.init_process_group`. )msgkwargsc tr5tjrtdtjdi|dSdS)aKInitialize process group required for ``torch.distributed``. Args: **kwargs (Dict[str, Any]): Additional arguments to pass to torch.distributed.init_process_group. Returns: bool: True if torch.distributed is initialized. Raises: RuntimeError: If torch.distributed is already initialized. z&torch.distributed already initialized.TFN)rAr9rH RuntimeErrorinit_process_group)rWs r*init_distributedr\sT    IGHH H ))&)))tur,c tjtjrtjndz}tj|td|dS)a Sets the number of threads used by torch to utilize all physical CPU cores for intra-op parallelism. Currently, this function sets num_threads to be the number of physical CPU cores divided by the number of GPUs as we use one process per GPU, and this avoids CPU oversubscription. Note that this is currently a rough approximation, and doesn't take into account environments where things like CPU affinity is set. r2z+Set intra op parallelism no. of threads to N) r5 cpu_counttorchrFr: device_countset_num_threadsr%info) num_threadss r*set_torch_num_threadsrdsq,..%*Z%<%<%>%>E !!!AK +&&&IIIKIIJJJJJr,z`get_world_size_and_rank` will move to `torchtune.utils._device` in future releases. Please use `torchtune.utils.get_world_size_and_rank` instead.ctjrOtjrz2load_from_full_model_state_dict..sR  ''VJu7JI,V,Vr,T)full_state_dictbroadcast_from_rank0rtru)rjmodel_state_dictoptionsrx) block_sizescaler_block_sizer2z'only support 1D FSDP but got mesh.ndim=r)dimN)shapedtypestride)mesh placements tensor_meta) local_tensorspec requires_grad device_mesh)rtassign)+any parameters state_dict(_DISTRIBUTED_STATE_DICT_API_IS_AVAILABLEkeysr7rKrrritemsryrzrxrrrrrndimNotImplementedErrorr>rr_rf ProcessGroup get_groupr?listchunk new_zeroscopy_rrrrrrrcpur Parameterload_state_dict)rjrsrIrtruhas_nf4meta_sharded_sd param_namesharded_meta_paramr sharded_sd full_tensorrrrshard_mesh_dimshard_world_size shard_rankr sharded_paramsharded_tensors r*load_from_full_model_state_dictrs>%%''G&&((O0LMLM!,,.. S SJ!0!4!4Z!@!@ ")*"5"8"89K9Q"R"RGJ  " !%#     $'7     '.}}; B; B #J !0!4!4Z!@!@ %..);)ABBEEfMMK)?;;5  "0)AA5 0=H $6$D$V!$)&7 *59q==-D DD"##'99^#<#< !%2DNN>4R4R$&&U[6FANNNOO!, 5 5ejjll C C o 1 o.44U;;;")!.$/;#5#@$."4"9"9";";"4":#5#<#<#>#>%%%#5"B " " "/?? !,!2&2&1""  6!/!3!3!5!5%'\.%A%AJz " "$$Zt$LLLr,rc|j}|j}||\}}g}|D]}|j^}}||zg|R} t j| |j|j } tj | ||d| | | |||j \} } | S)zR Manually gather NF4Tensor parameter since it does not support all_gather )rIrF)async_op)rrxfsdp_pre_all_gatherrrr>r_emptyrIrr9all_gather_into_tensorappendfsdp_post_all_gather) rr nf4_tensor quant_paramsmetadatafull_quant_params quant_paramd0dnrfull_quant_param full_param_s r*_gather_nf4_tensorr>s  $D,J';;DAAL(# 3 3 #Rdnn&&++---333 ; +,K4E    # k4>>+;+;e       !12222338Z%5MJ r, is_rank_zeroadapter_weights_onlyc,i}|}|D]\}}|jr||}t |dr>t |jtrt|}n| }t |tr||j }|r| ||<tj |rt|d}|S)aJ Converting sharded state dict into a full state dict on CPU Returning non-empty result only on rank0 to avoid peaking CPU memory Currenltly we can used distributed state dict API to process model without NF4Tensor. Otherwise, we need to manually gather any NF4 tensors until all-gather is supported in the NF4Tensor subclass TODO: add support for NF4Tensor at distributed state dict API Args: model (FSDPModule): Model to generate fully qualified names for cpu_state_dict is_rank_zero (bool): flag to check if the process is on rank 0 device (Optional[torch.device]): device to use for sharded tensors. Default: None adapter_weights_only (bool): flag to check if only trainable parameters should be returned. Default: False Returns: Dict[str, Any]: State dict on CPU rxN)rI)rris_cpurKryrzrxrrrrrr_rfbarrierr!)rjrrIrcpu_state_dictrrr|s r*gather_cpu_state_dictrVs.N!!##J'--//$$ E < %HHV$$E 5/ * * ,%-y99 ,*511))++ eY ' ' *HHU[))E  5).N: & !!####M/tLLL r,optcVtddd}t|||}|r|SiS)z Converting optimizer state from sharded to full For example, "exp_avg" in AdamW is `DTensor`, "exp_avg.full_tensor()" converts it to plain tensor on rank 0 Returning non-empty cpu state dict on rank 0 Tr~rru)rj optimizersr)rr)rjrrrIrr~s r*get_full_optimizer_state_dictrsQ4TG/WO r,ctr;tdd|tjdu}t ||||d Sd}t ||d}|d}|d}|d} t||D]\} } | D]\} } | |kr | | | <t| || |D]u\}}||vr ||}| |}|D]F\}}||}t|trt||j |j ||<A|||<Gv|||dd S) za Converting full optimizer state to sharded state dict and loading it into optimizer Trr)rjroptim_state_dictrparams param_groupsstate)rrN)rrr_rIrrrziprrzrrrrr)rjrrsrIrPARAMSrrfull_param_groups full_state param_groupfull_param_groupkeyvaluepidfull_pid param_statefull_param_stateattrrrs r*#load_from_full_optimizer_state_dictrs0- " !%%,u"5"55    !C'7      #~~''7   )#N3W% -0?P-Q-Q 8 8 )K).4466 ) ) U&==#( C  !$[%8:J6:R!S!S 8 8 Xe###Cj #-h#7 )9)?)?)A)A 8 8%D+%0%6N!.':: 8,='*6*5-- D))-8 D)) 8  8"  ,       r,namemodulenames_to_matchc|r||vrdS|d}t|dkr,|ddkot|dSdS)a Returs True for layers named {}.layers.i or layers that exactly match names_to_match, otherwise, returns False. This is a helper function for sharding a model with FSDP. In :func:`~torchtune.training.shard_model`, we iterate over the model's named modules and apply fully_shard using this condition. As part of our sharding strategy, we want each layer to be sharded separately, as this is generally efficient. We may also want to shard certain modules that are not layers, such as the embedding module. #TODO: a more robust way would be to shard on the module type, not the name. Args: name (str): Name of the module. module (nn.Module): Module to be sharded. names_to_match (Optional[List[str]]): List of names to match, if any. *args: Variable length argument list to be passed to the Embedding module. **kwargs: Arbitrary keyword arguments to be passed to the Embedding module. Returns: bool: True if the module name matches the condition, False otherwise. Examples: >>> names_to_match = ["embedding"] >>> layer_names = ["layers.0", "decoder.layers.1", "encoder.layers.2.attention", "my_wrapper.layer.1.something", "embedding"] >>> matches = [] >>> for name in layer_names: >>> if shard_condition_is_layer_or_match(name, None): matches.append(name) >>> print(matches) >>> ["layers.0", "decoder.layers.1", "embedding"] T.layersr4F)splitlenstrisdigit)rrrargsrW name_lists r*get_shard_conditionsrseN$.00t 3I 9~~}(GS[[2-G-GG 5r,T)reshard_after_forwarddp_meshshard_conditionsrrcF||d}|rt|d<d}tt|D]3\t fd|Drt fi||dz }4|dkrt dt |fi|dS)a{ Utility to shard a model with FSDP using the PyTorch Distributed fully_shard API. This method will over the model's named modules from the bottom-up and apply shard modules based on whether they meet any of the criteria from shard_conditions. Args: model (TransformerDecoder): Model to shard with FSDP. shard_conditions (List[Callable[[str, nn.Module], bool]]): A list of functions to determine which modules to shard with FSDP. Each function should take module name (relative to root) and the module itself, returning True if FSDP should shard the module and False otherwise. If any of shard_conditions return True for a given module, it will be sharded by FSDP. cpu_offload (bool): If set to True, FSDP will offload parameters, gradients, and optimizer states to CPU. reshard_after_forward (bool): Whether to reshard parameters and buffers after the forward pass. Setting this to True corresponds to the FULL_SHARD sharding strategy from FSDP1, while setting it to False corresponds to the SHARD_GRAD_OP sharding strategy. dp_mesh (Optional[DeviceMesh]): Device mesh to use for FSDP sharding under mutliple parallelism. Default to None. Raises: ValueError: If no layer modules were sharded, indicating that no shard_condition was triggered. )rroffload_policyrc(g|]}|SrYrY)r{shard_conditionmros r* zshard_model...s%NNN/1%%NNNr,r2zXNo layer modules were sharded. Please check if shard conditions are working as expected.N)r reversedr named_modulesrr ValueError) rjrrurr fsdp_kwargsnum_layers_shardedrros @@r* shard_modelrs>-B7SSK;(8(:(: $%e11334455$$1 NNNNN=MNNN O O $  ) )[ ) ) ) ! # Q f   %%%%%%%r,tp_meshcPt|ttf}|r|jn|}|}t |D]}t|tr|j|zdkrtd|jd|d|j |zdkrtd|j d|d|j |zdkrtd|j d|d|j|z|_|j |z|_ |j |z|_ |r||_|S)aS Utility to scale MultiHeadAttention parameters(num_heads, num_kv_heads, embed_dim) across tensor parallel devices. Each device will handle a portion of the attention computations. Args: model (nn.Module): Model whose attention parameters will be scaled by TP size. tp_mesh (DeviceMesh): Tensor parallel device mesh. Returns: nn.Module: The model with scaled MultiHeadAttention parameters. Raises: ValueError: If attention heads, kv heads, or embed dimension is not divisible by TP size. Examples: >>> from torchtune.modules import TransformerDecoder >>> from torch.distributed.device_mesh import DeviceMesh >>> model = TransformerDecoder( num_heads=32, num_kv_heads=32, embed_dim=4096, ) >>> tp_mesh = DeviceMesh("cuda", torch.arange(2)) # 2 GPUs >>> model = prepare_mha_for_tp(model, tp_mesh) >>> # Now each GPU has: >>> # num_heads = 16 (32/2) >>> # num_kv_heads = 16 (32/2) >>> # embed_dim = 2048 (4096/2) rzNumber of attention heads (z-) must be divisible by tensor parallel size (z).zNumber of KV heads (zEmbedding dimension () rzrr decoderr>rmodulesr num_headsr num_kv_heads embed_dim)rjris_fusion_modelrtp_sizers r*prepare_mha_for_tpr;sD!:J(KLLO.9emmEGllnnG '//## $ $11 a+ , , 1{W$)) 9!+99-4999~'1,, 91>99-4999{W$)) 9AK99-4999+0AK^w6AN+0AK  Lr,)r)F)r'N)FF)NF)N)Vloggingr5 itertoolsrtypingrrrrrr r r_torch.distributedrfr9r "torch.distributed._composable.fsdpr r torch.distributed._tensorrr)torch.distributed._tensor.placement_typesrr'torch.distributed.checkpoint.state_dictrrrrrtorch.distributed.device_meshrtorch.distributed.fsdprtorch.nn.modules.moduler torch.optimrtorchao.dtypes.nf4tensorrrtorchtune.modulesrtorchtune.modules.attentionrtorchtune.modules.model_fusionrr torchtune.modules.peftr!torchtune.utilsr"r#torchtune.utils._loggingr$r%Logger__annotations__ __version__ torch_versionrrr+r;rATensorr8rMrUr\rdriModulerqrIrrrrrrrrrrYr,r*rsI CCCCCCCCCCCCCCCCCC LLLLLLLL@@@@@@@@MMMMMMMM544444333333555555!!!!!!66666666000000::::::LLLLLLLL99999922222222//////!z||gn###! ,1(/S/-=//// DDDDD.elU\($SV: htCH~$* K K K K   D sCx     RRYR4RRRR* rMrM rM #s(^rM LrM  rM  rM  rMrMrMrMjblr|6&*!& ++ ++ U\ "+ +  #s(^ ++++d&*    U\ "   #s(^ 07 7 7 #s(^7  L 7  7 7 7 7 z+/.. . I.T#Y'.  ....l#'$( 1&1&1& 1&8S")$4d$:;<1& 1&  1& j ! 1& 1&1&1&1&h= 9= =Y======r,