o wZh @s ddlZddlZddlZddlZddlZddlZddlmZmZddlm Z ddl m Z ddl m Z mZddlmZddlmZmZmZmZmZmZddlmZddlZddlmZddlmZddl mm!Z"dd l#m$Z$dd l%m&Z&dd l'm(Z(m)Z)m*Z+dd l,m-Z-dd l.m/Z/m0Z0ddl1m2Z2ddl3m4Z4m5Z5m6Z6ddl7m8Z8ddl9m:Z:m;Z;mZ>m?Z?m@Z@ddlAmBZBmCZCmDZDmEZEddlmFZFmGZGddlHmIZJddlKmLZLmMZMmNZNmOZOddlPmQZQmRZRmSZSmTZTmUZUddlVmWZWdZXeSrdZYdZZej[\ZXneTrdZYdZZneUrdZYdZZej]\ZXnd ZYd!ZZd"ZXGd#d$d$eZ^Gd%d&d&eZ_Gd'd(d(ej`eZad)ej`d*ejbd+efd,d-Zcd.d/Zd 0 1dd)ej`d2eefd3d4Zfdd5d6Zgd7d8Zhd9d:Zidd)ej`d;eefdeefd?d@Zkd)ej`dAeefdBdCZlGdDdEdEZmGdFdGdGeaZnGdHdIdIeaZoGdJdKdKeoZpGdLdMdMeoZqGdNdOdOeaZrGdPdQdQerZsGdRdSdSej`ZtGdTdUdUeoZuGdVdWdWej`ZvGdXdYdYejwZxGdZd[d[ej`Zyejzd\efd]d^Z{ejzd_efd`daZ|ejzdbefdcddZ}eejzdeefdfdgZ~eejzdhefdidjZeejzdkefdldmZeejzdnefdodpZdqed+edredsefdtduZ vddwej`dxej`dyeedzffd{d|ZGd}d~d~eMZGdddeLZddeefddZGdddej`ZGdddej`ZGdddej`ZdS)N)ABCabstractmethod) nullcontext)deepcopy)autoEnumwraps)AnyCallablecast no_type_checkOptionalUnion)mock) checkpoint) DeviceMesh) CPUOffload fully_shardFullyShardedDataParallel) TrainingState)FSDPParamGroupRegisterPostBackwardFunction)#NO_RESHARD_AFTER_FORWARD_STRATEGIES)BackwardPrefetchMixedPrecisionShardingStrategy)ShardedGradScaler)always_wrap_policyModuleWrapPolicywrap)distribute_tensorDTensorShard)ColwiseParallelparallelize_moduleRowwiseParallelSequenceParallel)TransformerDecoderLayerTransformerEncoderLayer)DistributedDataParallel)MultiProcessTestCaseMultiThreadedTestCase run_subtests TEST_SKIPS) FILE_SCHEMAget_cycles_per_ms TEST_CUDATEST_HPUTEST_XPU) has_tritoncudaZncclzhpu:0ZhcclxpuZxcclcpuZglooc@eZdZeZeZdS) FSDPInitModeN)__name__ __module__ __qualname__rNO_FSDP RECURSIVErArAR/var/www/auris/lib/python3.10/site-packages/torch/testing/_internal/common_fsdp.pyr;Vs r;c@seZdZeZeZeZdS)DEVICEInitModeN)r<r=r>r DEVICE_BEFORE DEVICE_AFTERZ DEVICE_NEVERrArArArBrC_s rCc@sneZdZdZedeejdffddZedejfddZ edd d Z e ed e d e de jfd dZdS) FSDPTestModelzZThis defines the interface expected from all models used commonly for FSDP unit tests.return.cCdS)z+Returns an input for the model as as tuple.NrAselfdevicerArArB get_inputlzFSDPTestModel.get_inputcCrH)z,Returns the loss given the input and output.NrA)rJinputoutputrArArBget_lossqrMzFSDPTestModel.get_lossNcCrH)z__doc__rtupletorchTensorrLrPrS staticmethodr nnModulerWrArArArBrFhs  rFmodel process_group assert_fnc Csdd|D}|dd|D7}t|}ddt|D}tj|||d|d}|dus4J|ddD]}|dusBJt||D] \\}} \}} || | qGq:dS) a All-gathers module states across ranks and calls ``assert_fn`` on each pair of corresponding states from rank 0 and a nonzero rank. For example, if ``assert_fn`` is ``self.assertEqual()``, then this checks that all module states are equal across ranks. cS g|] \}}||fqSrAdetachr8).0 param_nameparamrArArB z)_assert_module_states..cSrcrArd)rfZ buffer_namebufferrArArBrirjcSsg|]}dqSNrA)rf_rArArBrisgrouprNr9)named_parametersZ named_buffersdistZget_world_sizerangeZall_gather_objectzip) r`rarbZnamed_module_states world_sizeolistZ rank0_statesstatermp1p2rArArB_assert_module_statess"     rycCs ttSrl)r[rK DEVICE_TYPErArArArB get_devtype r{FT zero_buffersc Cs|rt|nt}|Q|D]}t |Wdn1s&wYq|rM|D]"}t |Wdn1sGwYq2WddSWddS1s`wYdS)zBZeros the parameters and optionally buffers of ``model`` in place.N)FSDPsummon_full_paramsr parametersr[no_gradZzero_buffers)r`r}Z summon_fullctxrhrkrArArB _zero_models"      "rcCs"|s|t}|r ||Srl)torzhalf state_dict)r` cpu_offloadrrArArB_get_state_dicts  rcsdfdd|DS)Nrmcs$g|]}|durt|ndqS)Nnone)str)rfstest_name_mappingrArBris$z subtest_name..)join)rrTrArrB subtest_namesrcCs|D]\}}|jtdkr|||<q|dkr|ndg}t|ttttj f|d}| D] }|| t ||<q5|S)Nr8r) itemsrKr[r8rqZbroadcast_object_listr dictrr\keysrrz)rankrrgrhrurArArB_broadcast_state_dicts   rrecursecCsBtj||dtt|WdS1swYdS)a[ Returns the full unsharded parameters of ``model``. Any FSDP-managed parameters offloaded to CPU are moved to GPU in the returned list. Args: recurse (bool): If ``False``, only unshards the parameters immediate to ``model``; if ``True``, recurses through the module hierarchy rooted at ``model``. )rN)r~rrlistr)r`rrArArBget_full_paramss $rmove_to_devicecCs|r|tS|Srl)rrz)r`rrArArB_move_to_devicer wrap_fsdpcOs|s|St|g|Ri|Srlr~)r`rrTrUrArArB_maybe_wrap_fsdpsrc@sBeZdZdedefddZdefddZdefdd Zd d Zd S) DummyProcessGrouprsizecCs||_||_dSrl)_rank_size)rJrrrArArB__init__s zDummyProcessGroup.__init__rGcC|jSrl)rrJrArArBrzDummyProcessGroup.rankcCrrl)rrrArArBrrzDummyProcessGroup.sizecOst}dd}||_|S)NcSstj}|d|S)Nr9)r[futuresFuture set_result)futurerArArB get_futures  z/DummyProcessGroup.allreduce..get_future)rZMockr)rJrTrUZ dist_waitrrArArB allreduceszDummyProcessGroup.allreduceN)r<r=r>intrrrrrArArArBrs  rcseZdZdejdededeffdd ZddZd d Z d d Z d dZ e   ddejde dedeeeefdededeejeffddZddZZS)TransformerWithSharedParamsrodevice_init_modeadd_bn deterministiccst||_||_|rtdd}d}t|||_ tj |ddddd|_ t |||_ |j j|j _|d|j j|f|d tj|jtjd d|_|r^tj|jntj|_|tjkrn|t}|rv|dSdS) Nrg?)d_modelZnum_encoder_layersZnum_decoder_layersZdim_feedforwardZdropout vocab_bias long_buffer)dtype)superrrrrtr[ manual_seedr^Z Embedding embed_tokensZ Transformer transformerLinear output_projweightregister_bufferZnew_onesZ zeros_likerlongbsZ BatchNorm1dZIdentitybnrCrDrrzeval)rJrorrrZd_vocabr __class__rArBrs>        z$TransformerWithSharedParams.__init__cCsNtd|jtjd|dd|j}tj|jd|dd|j}||fS)Nr9 rKr5)r[rrZarangeviewr)rJrKsrctgtrArArBrL.sz%TransformerWithSharedParams.get_inputcCsJ||}||j|j|}||}||}|||}||Srl)rrrZtype_asrrr)rJZsrc_idsZtgt_idsrrxrArArBforward4s     z#TransformerWithSharedParams.forwardcCs.|\}}tjj|d|d|dddS)Nsum)Z reduction)r^ functionalZ cross_entropyrr)rJrNrOrmrrArArBrP<sz$TransformerWithSharedParams.get_losscC |dSrlbackwardrQrArArBrSB z(TransformerWithSharedParams.run_backwardNFTfsdp_init_mode fsdp_kwargsrGc Cs|duri}|tjkrt|tr|d}n|}t||||S|tjkrud|vr.ttth}n| d}d|vrI|dt j t j hvrIt|tsId}n|}t|trU|d} n|} t| |||} t | |fd|i|} |tjkrs| t} | Std|)au Initializes a :class:`TransformerWithSharedParams` instance. Args: fsdp_init_mode (FSDPInitMode): If ``NO_FSDP``, then does not wrap any modules with FSDP. If ``RECURSIVE``, then wraps with top-level FSDP. By default, the top-level FSDP uses the ``ModuleWrapPolicy`` for encoder and decoder layers, but a different auto wrap policy may be specified via ``fsdp_kwargs``. device_init_mode (DEVICEInitMode): Determines model movement to DEVICE. fsdp_kwargs (Optional[Dict[str, Any]]): Optional keyword arguments forwarded to the FSDP constructor. deterministic (bool): Whether to make the model deterministic across constructions. add_bn (bool): Whether to include batch norm in the model. Nrauto_wrap_policysharding_strategyUnsupported FSDP init mode: )r;r? isinstancerZrr@rr)r(poprZ HYBRID_SHARDZ_HYBRID_SHARD_ZERO2r~rCrErrz ValueError) rorrrrrZpgrZfsdp_pgZ tformer_pgm fsdp_modelrArArBrWEsV          z TransformerWithSharedParams.initcCs|jgSrl)rrrArArBget_ignored_modulessz/TransformerWithSharedParams.get_ignored_modules)NFT)r<r=r>rq ProcessGrouprCboolrrLrrPrSr]r;rrrr rr^r_r~rWr __classcell__rArArrBrsD*  MrcseZdZdejdededeffdd ZddZd d Z d d Z d dZ e  ddejde dedeeeefdedejf ddZZS)NestedWrappedModulerorrrc st|_|_|tjk}fdd}|r#tdt t t dd||t |t t dd|t t dd||t t dd|t t dd||_ dS)Ncr t|fiS|SrlrlayerrrorrArB _maybe_wrapz1NestedWrappedModule.__init__.._maybe_wraprrr5r)rrrrrtrCrDr[rr^ SequentialrrmodulerJrorrrrrrrrrBrs$      zNestedWrappedModule.__init__cCs"td|jtjdd|dfS)Nr9r5rr)r[rrZrandrIrArArBrLszNestedWrappedModule.get_inputcC ||SrlrrJrrArArBrr|zNestedWrappedModule.forwardcCs |}|Srl)rrJrNrOrRrArArBrPszNestedWrappedModule.get_losscCrrlrrQrArArBrSrz NestedWrappedModule.run_backwardNFrrrGcCsn|duri}|tjkrt|d||dS|tjkr0t|fd||d|}|tjkr.|t}|Std|)a Initializes a :class:`NestedWrappedModule` instance. Args: fsdp_init_mode (FSDPInitMode): If ``NO_FSDP``, then does not wrap any modules with FSDP. If ``RECURSIVE``, then wraps some nested modules with FSDP but not the top-level module. The model may later be wrapped with a top-level FSDP external to this method if desired. device_init_mode (DEVICEInitMode): Determines model movement to DEVICE. fsdp_kwargs (Optional[Dict[str, Any]]): Optional keyword arguments forwarded to the FSDP constructor. deterministic (bool): Whether to make the model deterministic across constructions. NFrrrTr) r;r?rr@rCrErrzr)rorrrrrrArArBrWs.    zNestedWrappedModule.initNF)r<r=r>rqrrrCrrLrrPrSr]r;rrrr r^r_rWrrArArrBrs< rc sJeZdZe  d dejdededee e e fde f fdd Z ZS) AlwaysWrapNestedWrappedModuleNFrorrrrcsltttj|tj|||d}|tjkr|S|tjkr4|pi}t|fdti|}|tj kr2| t }|SdS)z Initializes a :class:`NestedWrappedModule` instance, but unlike :meth:`NestedWrappedModule.init`, for the ``RECURSIVE`` init mode, this wraps with top-level FSDP and the ``always_wrap_policy()`` auto wrap policy. )rorrrrrN) rrrWr;r?r@r~rrCrErrz)rorrrrr`rrrArBrWs&    z"AlwaysWrapNestedWrappedModule.initr)r<r=r>r]rqrr;rCrrrr rrWrrArArrBrsrc steZdZdejdededeffdd Zedd d Z e  ddejd e ded e e e efdef ddZZS)NonUniformReqGradNWMrorrrc stt||_|_|tjk}fdd}|r%t dt t t dd||t |t t dd|t t dd||t t t dd|t t dd||_dS)NcrrlrrrrArBr+rz2NonUniformReqGradNWM.__init__.._maybe_wraprrr5r)rrrrrrtrCrDr[rr^rrrrrrrrBrs,     zNonUniformReqGradNWM.__init__rGNcCs,|D]\}}t||s|dqdSr)rprematchZrequires_grad_)r`Z req_grad_masknprArArB_set_nonuniform_req_gradBs   z-NonUniformReqGradNWM._set_nonuniform_req_gradFrrcCstd}|tjkrt|d||d}t|||S|tjkrC|dur%i}t|fd||d|}|tjkr;| t }t|||St d|)a Initializes a :class:`NestedWrappedModule` instance, but unlike :meth:`NestedWrappedModule.init`, it wraps a second :class:`torch.nn.Sequential` container to enable the desired non-uniform ``requires_grad`` ``use_orig_params=True`` tests. For both ``RECURSIVE`` and ``NO_FSDP`` init modes, freezes all parameters except the last two to validate ``ShardedGradScaler`` support for ranks with no (non-zero sized) local shards in FSDP ``use_orig_params=True`` mode. zmodule\.2.*\.1.*FrNTr) rcompiler;r?rrr@rCrErrzr)rorrrrZreq_grad_patternZ ddp_modelrrArArBrWHs6       zNonUniformReqGradNWM.initrXr)r<r=r>rqrrrCrr]rr;rrrr rWrrArArrBrs4* rc sveZdZdZdejdedeffdd ZddZd d Z d d Z d dZ e de ededededef ddZZS)ModuleWithDelayzThis class wraps a :class:`FSDPTestModel` to optionally add a delay after computing the loss and/or before the gradient reduction.rdelay_after_loss_msdelay_before_reduction_mscs t||_||_||_dSrl)rrrrr)rJrrrrrArBr{s  zModuleWithDelay.__init__cCs |j|Srl)rrLrIrArArBrLrzModuleWithDelay.get_inputcCrrlrrrArArBrr|zModuleWithDelay.forwardcCsT|j||}|jdkr(tstrt|jd|Str(tj t |jt |SNr) rrPrr2r3timesleepr1r[r6_sleeprr0rrArArBrPs zModuleWithDelay.get_losscsTtjjfdd}td|j|WddS1s#wYdS)NcsNjdkr trtjtjtn tstr t jd|i|Sr) rr1r[r6rrr0r2r3rrrVorig_reduce_scatterrJrArB_delayed_reduce_scatters z=ModuleWithDelay.run_backward.._delayed_reduce_scatterz'torch.distributed.reduce_scatter_tensor)r[ distributedreduce_scatter_tensorrpatchrrS)rJrRrrArrBrSs "zModuleWithDelay.run_backward module_class model_args model_kwargscOst|j|i|||S)aA Args: module_class (Type[FSDPTestModel]): Wrapped module class to which to add delays. model_args: Positional arguments forwarded to the ``module_class`` ``init()``. delay_after_loss_ms (int): Delay after computing the loss/before the optimizer step (in ms). delay_before_reduction_ms (int): Delay before reduce-scattering gradients (in ms). model_kwargs: Keyword arguments forwarded to the ``module_class`` ``init()``. )rrW)r rrr r rArArBrWs zModuleWithDelay.init)r<r=r>rYr^r_rrrLrrPrSr]typerFr rWrrArArrBrws2  rc@sReZdZeejddddfdejdedede e e e fde d ed efd d ZdS) NestedWrappedModuleWithDelayNFrrorrrrrrc Cstjt|||||||dS)Nrorrrrrr)rrWrrrArArBrWs z!NestedWrappedModuleWithDelay.init)r<r=r>r]rCrErqrr;rrrr rrrWrArArArBrs,rc$eZdZfddZddZZS)DummyDDPcst||_dSrl)rrr)rJrrrArBrs  zDummyDDP.__init__cOs|j|i|SrlrrJrTrUrArArBrszDummyDDP.forwardr<r=r>rrrrArArrBrs rcseZdZdejdedededef fdd Zdd Z d d Z e ddejde dede eeefdedef ddZZS)MixtureOfExpertsrorrdelay_before_free_msrc s$tj||||d||_||_||_|tjk|_|r#t d|j d}d}d} t t |||j} tdd| D|_| D]} d| _qC|rPt d t t |||j} |rwtj| g} t| | fi|} t| |fi|} t t t | ||j| | t t || |j|_dS) N)rorrr*rrrcs|]}|VqdSrl)Znumel)rfrrArArB z,MixtureOfExperts.__init__..Tr)rrrorrrCrDrr[rrrr^rrrZnum_expert_paramsexpertrZ new_groupr~rr)rJrorrrrrZd_expertZd_sharedZd_inputrrZsharedZ expert_grouprrArBrsD     zMixtureOfExperts.__init__csxjdkr7jd}t|tr7tjjjjfdd}t d| |WdS1s2wY|S)NrrcsDtrtjtjtn tstrt jd|i|S)Nr) r1r[r6rrrr0r2r3rrrV orig_reshardrJrArB_delayed_reshardsz2MixtureOfExperts.forward.._delayed_reshardz.torch.distributed.fsdp._runtime_utils._reshard) rrrr~r[rfsdpZ_runtime_utilsZ_reshardrr )rJrrrrArrBrs      zMixtureOfExperts.forwardcCs||jsAt,|D]}t|drq|jdur.|j|jtj j |j|j dqWddS1s:wYdSdS)Nrrn) rrr[rrhasattrgradZdiv_rtr all_reducero)rJrRrrArArBrS1s    "zMixtureOfExperts.run_backwardNFrrrcCsr|duri}|tjkrt|d|||dS|tjkr2t|fd|||d|}|tjkr0|t}|Std|)a Initializes a :class:`MixtureOfExperts` instance. Args: fsdp_init_mode (FSDPInitMode): If ``NO_FSDP``, then does not wrap any modules with FSDP. If ``RECURSIVE``, then wraps some nested modules with FSDP, including the expert and shared layers, but not the top-level module. The model may later be wrapped with a top-level FSDP external to this method if desired. device_init_mode (DEVICEInitMode): Determines model movement to DEVICE. fsdp_kwargs (Optional[Dict[str, Any]]): Optional keyword arguments forwarded to the FSDP constructor. deterministic (bool): Whether to make the model deterministic across constructions. delay_before_free_ms (int): Delay before resharding expert parameters in the forward pass (in ms). NF)rrrrTr) r;r?rr@rCrErrzr)rorrrrrrrArArBrW=s2    zMixtureOfExperts.init)NFr)r<r=r>rqrrrCrrrrSr]r;rrrr rWrrArArrBrs>4 rc sdeZdZ ddddddedeejded ed ef fd d Zd ej dej fddZ ddZ Z S)MLPNTFr5)bias with_bufferdim_multiplierdimrKr$r%r&csdttj|||||d|_tj|||||d|_|r-|dtj|f|ddSd|_ dS)N)rKr$rkr) rrr^rin_projout_projrr[Zrandnrk)rJr'rKr$r%r&rrArBrrs  z MLP.__init__rrGcCs@||}t|}||}t|}|jdur||j}|Srl)r(Frelur)rk)rJrzrArArBrs      z MLP.forwardcCs"|jdurtjj|jdSdSrl)rkr[r^rWZnormal_rrArArBreset_parameterss zMLP.reset_parametersrl) r<r=r>rrr[rKrrr\rr-rrArArrBr#qs& r#csFeZdZdddedeffddZdeded ed dfd d ZZS) MLPStackF)with_seq_parallelmlp_dimr/csLt|ddt|t|ddg}|r|tj|ddtj|||_dS)N)r&Fr$)r#appendr^ LayerNormrrr/)rJr0r/modulesrrArBrs    zMLPStack.__init__tp_meshdp_meshuse_activation_checkpointingrGcKstddtddtddtddtdd|jrttddntd}|jr-tdd|d<t|||d|D]}t|tjr?q6|rEt |t |fd |i|q6t |fd |i||S) NF)Zuse_local_outputr9)Zoutput_layouts)z 0.in_projz 0.out_projz 1.in_projz 1.out_projz 2.in_projz 2.out_proj)Z sequence_dim3) device_meshparallelize_planmesh) r$r&r/r#r'r%rr^r4rr)rJr6r7r8rr;rrArArB parallelizes(   zMLPStack.parallelize) r<r=r>rrrrr=rrArArrBr.s r.csVeZdZdZd dedeffdd Zdejde e ejejfejffd d Z Z S) DoubleLinearz This can be used for returning multiple outputs from a module (``use_second_linear=True``) or for having an unused module (``False``). Tr'use_second_linearcs:tt|||_t|||_t|_||_dSrl) rrr^rlin1lin2ZReLUr+r?)rJr'r?rrArBrs   zDoubleLinear.__init__rrGcCs6|jr||||||fS|||Srl)r?r+r@rArrArArBrs zDoubleLinear.forwardT) r<r=r>rYrrrr[r\rrZrrrArArrBr>sr>new_all_gather_into_tensorccBtj}t|t_z dVWt|t_dSt|t_wrl)rqZall_gather_into_tensorbarrier)rCZorig_all_gatherrArArBpatch_all_gather rFnew_reduce_scatter_tensorccrDrl)rqr rE)rHrrArArBpatch_reduce_scatterrGrInew_all_reduceccrDrl)rqr"rE)rJZorig_all_reducerArArBpatch_all_reducerGrK new_unshardccBtj}t|t_z dVWt|t_dSt|t_wrl)rZunshardrqrE)rLZ orig_unshardrArArB patch_unshard rN new_reshardccrMrl)rZreshardrqrE)rPrrArArB patch_reshard rOrQnew_post_backwardccrMrl)rZ post_backwardrqrE)rRZorig_post_backwardrArArBpatch_post_backwardrOrS new_backwardccrMrl)rrrqrE)rTZ orig_backwardrArArB*patch_register_post_backward_hook_backward$rOrUrrTrUcOsRt|dkr |d}nd|vr|d}n td|d|||||i|S)NrrOz,Cannot get reduce-scatter output from args: z kwargs: )lenAssertionError)clsrrbrTrUrOrArArBreduce_scatter_with_assert1s   rYrAreplicated_modulesharded_moduleprefixes_to_ignore.cCst||D]\\}}\}}|}|D]} || d}q|||||tt|ts1J|j|j} } t | t dt dfkrHt dt || | } || | |jdurd||jq ||jt |j| | } ||jtt|jtsJ||j | q dS)NrzmFSDP's (Shard(0), Shard(0)) layout differs from distribute_tensor(), so we cannot check for equality using it)rsrpreplace assertEqualZassertIsInstancer"rr: placementsrZr#rWr!Zto_localr!Z assertIsNoneZassertIsNotNone)rXrZr[r\Zreplicated_nameZreplicated_paramZ sharded_nameZ sharded_paramZclean_sharded_nameprefixr<r`Zsharded_ref_paramZsharded_ref_gradrArArBcheck_sharded_parityDs2       rbcs@eZdZeddZfddZddZddZd d ZZ S) FSDPTestMultiThreadcCtSrl DEVICE_COUNTrrArArBrtfzFSDPTestMultiThread.world_sizecst|dSrl)rsetUpZ_spawn_threadsrrrArBrhjs  zFSDPTestMultiThread.setUpcOt|g|Ri|Srlr-rrArArBr-nz FSDPTestMultiThread.run_subtestscCtjdSrlr[_dynamoresetrrArArBperThreadSetUpqz"FSDPTestMultiThread.perThreadSetUpcCrlrlrmrrArArBperThreadTearDowntrqz%FSDPTestMultiThread.perThreadTearDown) r<r=r>propertyrtrhr-rprrrrArArrBrces  rcc$sjeZdZfddZeddZeddZedefdd Zed d Z d d Z ddZ ddZ ddZ eddZ       d4dejdedededeededeed ed!ed"eeeeffd#d$Zdd%d&edddddddddf d'eed(ed)ed*eed+eded,ed-ee d.ee!deed/ed0ed ed!ed1eeeefd"eeeeff d2d3Z"Z#S)5FSDPTestcs tdtjd<|dS)N0ZTORCH_NCCL_DESYNC_DEBUG)rrhosenvironZ_spawn_processesrrrArBrhys   zFSDPTest.setUpcCrdrlrerrArArBrtrgzFSDPTest.world_sizecCs tjSrl)rqZdistributed_c10dZ_get_default_grouprrArArBras zFSDPTest.process_grouprGcCrHrrArrArArBdestroy_pg_upon_exitrMzFSDPTest.destroy_pg_upon_exitcCst|jSrl)r/ file_namerrArArB init_methodszFSDPTest.init_methodcC|||jdSrl)r_r)rJrrrArArB_check_cpu_offloadrzFSDPTest._check_cpu_offloadcCr{rl)r_backward_prefetch)rJrr}rArArB_check_backward_prefetchrz!FSDPTest._check_backward_prefetchcCr{rl)r_forward_prefetch)rJrrrArArB_check_forward_prefetchrz FSDPTest._check_forward_prefetchcOrirlrjrrArArBr-rkzFSDPTest.run_subtestsc Ks(||}||_||_|dd}td|jd|jz%|r2tjjjj }t j d|j||dnt j |j tt|j|jdWnty]} zd| jd vrXttd jd} ~ wwd} |jt} tsitrotj| | g} t j| d tj|||tjt j| d t dS) Nfake_pgFz dist init r=z, world=Zfake)backendrtrstore)rzrrtrZ recompilerZbackend_unavailable) device_ids) rrygetprintrtr[testing _internalrrZ FakeStorerqZinit_process_grouprzDISTRIBUTED_BACKENDr RuntimeErrorrTsysexitr. exit_coderfr1r3Z acceleratorZset_device_indexrErnroZrun_testZdestroy_process_group) rXrZ test_namerypiperUrJrrer device_idrArArB_runsL         z FSDPTest._run{Gz?NFr` num_stepsautocastlrfsdp_cpu_offload save_modelmixed_precisionenable_sharded_grad_scaler use_pure_fp16sharded_grad_scaler_kwargsc  CsF|o|j} t|j} | duri} td d|i| } tjj||dd}t|D]}| tj j t |dY|j tt }| sK|r_t|ts_t|tjrV|}n tdd|D}||}| rt|tr|jtvr|D] }||jtdqs|j ||| }Wdn1swY| |}|s| s|jtjksJdn'| r||jtjnt|tr|dusJ||j|jn||jtj|j || rt|tr|D] }||jtdq| || |rd d |!"D}t#||$|q*t|tr|%t&j'|(S) Nenabledg?)rZmomentum)rcsrrl)r)rfrrArArBrrz4FSDPTest._train_for_several_steps..r8zeloss data type should be float32, as the original parameter data type is float32.cSsi|] \}}||qSrA)clone)rfkvrArArB sz5FSDPTest._train_for_several_steps..rA))offload_paramsnextrrKrr[optimZSGDrrZ zero_gradamprrzrrLrr~r\rrZrrr_rPrscalerZfloat32Zfloat16Z param_dtyperSstepupdaterrrZload_state_dictZ _assert_staterZIDLEre)rJr`rrrrrrrrrZcpu_offload_paramsZ model_deviceZsharded_grad_scalerrrmrNrOrrRrrArArB_train_for_several_stepssn              z!FSDPTest._train_for_several_stepsrT model_classrr ref_init_fn num_itersrr}rruse_orig_params init_kwargsc" Ks|tjks Jd|duri}d}|j}|j|jtjtjfddi|}|dur>tr5t|t gt d}n t||g|d}n||}|rH| }|j ||| du||| | ||d }t | }|||| | | | dz|j|j|||fddi|}Wnty}z td |d t||d}~wwt|tst||jfi|}|r| }|tjkr|t }|duo|j}|o|tjk}|o|tjk}|rtd }| D] }||j|q|r|td t nt}||j ||d |||| | ||d } Wdn 1s wY|rdS|r/td }| D] }||j|q| t } t|}!tjj|| d d| durO|sQ|j||!ddddSdSdS)a Tests FSDP training against a reference, which defaults to DDP but may be customized with ``ref_init_fn``. Args: model_class (Type[FSDPTestModel]): A model class that inherits from ``FSDPTestModel``, which defines the expected interface. fsdp_init_mode (FSDPInitMode): The mode to initialize the FSDP-wrapped model. This should not be ``NO_FSDP``. ref_init_fn (Optional[Callable]): A callable to invoke that wraps a non-wrapped model to construct the reference model, where this wrapper should provide data parallel semantics. If ``None``, then the callable defaults to the DDP constructor. z.Expects an FSDP init mode that wraps with FSDPNrrT)rZ output_device)rrrrrrr)rr}rrrrz Initializing z raised error r8zOAn FSDP-managed module with parameter CPU offloading enabled has parameters on F)rrrrrrrr)Z check_dtypezFSDP did not match DDP)Z exact_devicemsg) r;r?rarrWrCrDr2DDPrzrrrrr Exceptionrrrr~rErrr[rKr_assertRaisesRegexrrrrZ assert_close)"rJrrrrrrrr}rrrrrrrrrrrr`Z ref_modelZref_lossZ ddp_paramsrrrZexpects_device_errorZexpects_cpu_deviceZ cpu_devicerhcontextZ fsdp_lossZfsdp_unsharded_paramsrArArB_test_fsdp_parity)s #                zFSDPTest._test_fsdp_parity)rNFNFFN)$r<r=r>rhrsrtrarrxrzr|r~rr- classmethodrr^r_rfloatrrrrrr rrrFr;rCr rrrrrArArrBrtxs     8   \     rtcompile_compute_on_modulecs.fddGdddtfdd}|S)Ncs>tjjj|i|dust|dr|ddSdS)Nr)r[rrrrrrV)rrArB!fully_shard_with_compiled_computes  z=compiled_fsdp_test..fully_shard_with_compiled_computec@r:)z*compiled_fsdp_test..FullyShardModeN)r<r=r>rEAGERCOMPILED_COMPUTErArArArBFullyShardModes rcstfdd}|S)Ncstjjj}D]b}|jkrtstdqtjj j }tj j j }tj |jkr.|}n|jkr@dtjj _ dtj j _ }ntd||j|j<|i|tj |j|j<|tjj _ |tj j _ qdS)Nz0Inductor on GPU needs Triton and recent GPU archTr9z!Need to implement FullyShardMode=)r[rrrrr4warningswarnrnconfigZskip_fsdp_hooksZ _inductorZcompile_threadsrErNotImplementedError __globals__r<)rTrUZoriginal_fully_shardmodeZoriginal_skip_fsdp_hooksZoriginal_compile_threadsZfully_shard_patch)rrfuncrArBwrappers0              z6compiled_fsdp_test..decorator..wrapperr)rr)rr)rrB decorators z%compiled_fsdp_test..decorator)r)rrrA)rrrrBcompiled_fsdp_tests $rcs&eZdZdfdd ZddZZS) SkipModulerGNcs ttjdddd|_dSN Fr2)rrr^rlinrrrArBrs zSkipModule.__init__cCrrl)rrrArArBr r|zSkipModule.forwardrXrrArArrBrsrcr) NestedLinearcsJt|rttjddddt|_dStjddddt|_dSr)rrr r^rrrz nested_linear)rJ fsdp_wraprrArBrs  zNestedLinear.__init__cCrrl)rrrArArBrr|zNestedLinear.forwardrrArArrBrs rcr) SkipModelcsHttjddddt|_tt|_t t |dtd|_ dS)NrFr2)r)r) rrr^rrrzlinearr linear_skipr rr)rJZ double_nestrrArBrs   zSkipModel.__init__cCs"||}||}||}|Srl)rrrrrArArBr$s   zSkipModel.forwardrrArArrBrs r)FT)FFrB)rArl) contextlibrvrrrrabcrrrcopyrenumrr functoolsr typingr r r r rrZunittestrr[Ztorch.distributedrrqZtorch.nnr^Ztorch.nn.functionalrr*Ztorch.distributed._composablerZtorch.distributed.device_meshrZtorch.distributed.fsdprrrr~Z$torch.distributed.fsdp._common_utilsrZ5torch.distributed.fsdp._fully_shard._fsdp_param_grouprrZ"torch.distributed.fsdp._init_utilsrZ2torch.distributed.fsdp.fully_sharded_data_parallelrrrZ*torch.distributed.fsdp.sharded_grad_scalerrZtorch.distributed.fsdp.wraprrr Ztorch.distributed.tensorr!r"r#Z!torch.distributed.tensor.parallelr$r%r&r'r(r)Ztorch.nn.parallel.distributedr*rZ*torch.testing._internal.common_distributedr+r,r-r.Z$torch.testing._internal.common_utilsr/r0r1r2r3Ztorch.utils._tritonr4rfrzrr6Z device_countr7r;rCr_rFrryr{rrrrrrrrrrrrrrrrrr#rr.r>contextmanagerrFrIrKrNrQrSrUrYrZrrbrcrtrrrrrrArArArBs                     ^"`M  -          !]3