U <c @sddlZddlZddlZddlmZmZddlmZddlm Z ddl m Z m Z ddl mZddlmZmZmZmZmZmZmZmZddlmZddlZddlmZddlmZdd l m!Z!dd l m"Z#dd l$m%Z%m&Z&m'Z'm(Z(dd l)m*Z*dd l+m,Z,m-Z-m.Z.ddlm/Z/m0Z0ddl1m2Z3ddl4m5Z5m6Z6ddl7m8Z8m9Z9Gddde Z:Gddde Z;GdddejedddZ?dKejd>eJZOGd?d@d@e6ZPGdAdBdBej<ZQGdCdDdDej<ZRGdEdFdFej<ZSdGdHZTdIdJZUdS)NN)ABCabstractmethod)suppress)deepcopy)Enumauto)inf)AnyCallableDictListOptionalTupleTypeUnion)mock) CPUOffload)FullyShardedDataParallel)BackwardPrefetchMixedPrecisionShardingStrategyTrainingState_)ShardedGradScaler)always_wrap_policytransformer_auto_wrap_policywrap)TransformerDecoderLayerTransformerEncoderLayer)DistributedDataParallel) TEST_SKIPSMultiProcessTestCase) FILE_SCHEMAget_cycles_per_msc@seZdZeZeZdS) FSDPInitModeN)__name__ __module__ __qualname__rNO_FSDP RECURSIVEr)r)G/tmp/pip-unpacked-wheel-gikjz4vx/torch/testing/_internal/common_fsdp.pyr#(sr#c@seZdZeZeZeZdS) CUDAInitModeN)r$r%r&r CUDA_BEFORE CUDA_AFTERZ CUDA_NEVERr)r)r)r*r+1sr+c @seZdZdZeeejdfdddZeejdddZ eddd d Z e edd d e j eeeeeeefeeejd ddZdS) FSDPTestModelzZThis defines the interface expected from all models used commonly for FSDP unit tests..returncCsdS)z+Returns an input for the model as as tuple.Nr)selfdevicer)r)r* get_input=szFSDPTestModel.get_inputcCsdS)z,Returns the loss given the input and output.Nr))r2inputoutputr)r)r*get_lossBszFSDPTestModel.get_lossNcCsdS)zr@r;r<r?rAr)r)r*initLs zFSDPTestModel.init)r$r%r&__doc__rrtorchTensorr4r7r: staticmethoddist ProcessGroupr#r r+r r strboolnnModulerBr)r)r)r*r.:s*r.)model process_group assert_fnc Csdd|D}|dd|D7}t|}ddt|D}tj|||d|d}|ddD]*}t||D]\\}} \}} || | qvqhdS) 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. cSs g|]\}}||fqSr)detachcpu).0 param_nameparamr)r)r* hsz)_assert_module_states..cSs g|]\}}||fqSr)rP)rSZ buffer_namebufferr)r)r*rVlscSsg|]}dqSNr))rS_r)r)r*rVqsr=rN)Znamed_parametersZ named_buffersrGZget_world_sizerangeZall_gather_objectzip) rMrNrOZnamed_module_states world_sizeZolistZ rank0_statesstaterYp1p2r)r)r*_assert_module_states\s  rbF)rM zero_buffersc Csrt|^|D] }t|W5QRXq|rd|D] }t|W5QRXqBW5QRXdS)zBZeros the parameters and optionally buffers of ``model`` in place.N)FSDPsummon_full_params parametersrDno_gradZzero_buffers)rMrcrUrWr)r)r* _zero_modelxs     ricCs |s |}|r||SrX)cudahalf state_dict)rM cpu_offloadrkr)r)r*_get_state_dicts rncsdfdd|DS)NrYcs$g|]}|dk rt|ndqS)Nnone)rI)rSstest_name_mappingr)r*rVsz subtest_name..)join)rrargsr)rqr* subtest_namesruTrMrecursec Cs6tj||dtt|W5QRSQRXdS)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``. )rwN)rdrerlistrfrvr)r)r*get_full_paramss ryrM move_to_cudacCs|r |S|SrX)rjrzr)r)r* _maybe_cudasr|)rM wrap_fsdpcOs|s|St|f||SrXrd)rMr}rtkwargsr)r)r*_maybe_wrap_fsdpsrc@s@eZdZeedddZedddZedddZd d Zd S) DummyProcessGroup)ranksizecCs||_||_dSrX)_rank_size)r2rrr)r)r*__init__szDummyProcessGroup.__init__r/cCs|jSrX)rr2r)r)r*rszDummyProcessGroup.rankcCs|jSrX)rrr)r)r*rszDummyProcessGroup.sizecOst}dd}||_|S)NcSstj}|d|S)Nr[)rDZfuturesZFutureZ set_result)futurer)r)r* get_futures  z/DummyProcessGroup.allreduce..get_future)rZMockr)r2rtrZ dist_waitrr)r)r* allreduceszDummyProcessGroup.allreduceN)r$r%r&intrrrrr)r)r)r*rsrcs.eZdZeddffdd ZddZZS)DeterministicModelF)offload_paramscsTttdtjdd|_|r      z$TransformerWithSharedParams.__init__cCsNtd|jtjd|dd|j}tj|jd|dd|j}||fS)Nr[ r3)rDrrZarangeviewr)r2r3srctgtr)r)r*r4sz%TransformerWithSharedParams.get_inputcCsJ||}||j|j|}||}||}|||}||SrX)rrrZtype_asrrr)r2Zsrc_idsZtgt_idsrrrr)r)r*rs     z#TransformerWithSharedParams.forwardcCs.|\}}tjj|d|d|dddS)Nsum)Z reduction)rKZ functionalZ cross_entropyrr)r2r5r6rYrr)r)r*r7s z$TransformerWithSharedParams.get_losscCs |dSrXbackwardr8r)r)r*r: sz(TransformerWithSharedParams.run_backwardNFT)r=r>r@r;r<rr0cCs|dkr i}|tjkr$t||||S|tjkrd|krJtjttthd}n | d}t t|||||fd|i|}|t j kr| }|Std|dS)a} 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 ``transformer_auto_wrap_policy()`` for encoder and decoder layers, but a different auto wrap policy may be specified via ``fsdp_kwargs``. cuda_init_mode (CUDAInitMode): Determines model movement to CUDA. 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. Nauto_wrap_policy)Ztransformer_layer_clsUnsupported FSDP init mode: )r#r'rr( functoolspartialrrrpoprdr+r-rj ValueError)r=r>r@r;r<rr fsdp_modelr)r)r*rBs2     z TransformerWithSharedParams.initcCs|jgSrX)rrr)r)r*get_ignored_modulesEsz/TransformerWithSharedParams.get_ignored_modules)NFT)r$r%r&rGrHr+rJrr4rr7r:rFr#r r rIr rrKrLrdrBrrr)r)rr*rs.* 4rc sxeZdZejeeedfdd ZddZddZ dd Z d d Z e deje eeeeefeejdddZZS)NestedWrappedModuler=r}r@r<c st|_|_|tjk}fdd}|rFtdt t t dd||t |t t dd|t t dd||t t dd|t t dd||_ dS)Ncsrt|fS|SrXr~)Zlayerr;r=r}r)r* _maybe_wrapWsz1NestedWrappedModule.__init__.._maybe_wraprrrr)rrrrr^r+r,rDrrK Sequentialr|rmodule)r2r=r}r@r<r;r{rrrr*rJs$     zNestedWrappedModule.__init__cCs"td|jtjdd|dfS)Nr[rrr)rDrrZrandr1r)r)r*r4jszNestedWrappedModule.get_inputcCs ||SrXrr2rr)r)r*rnszNestedWrappedModule.forwardcCs |}|SrX)rr2r5r6r9r)r)r*r7qszNestedWrappedModule.get_losscCs |dSrXrr8r)r)r*r:usz NestedWrappedModule.run_backwardNF)r=r>r@r;r<r0cCsp|dkr i}|tjkr&t|d||dS|tjkr^t|fd||d|}|tjkrZ|}|Std|dS)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. cuda_init_mode (CUDAInitMode): Determines model movement to CUDA. fsdp_kwargs (Optional[Dict[str, Any]]): Optional keyword arguments forwarded to the FSDP constructor. deterministic (bool): Whether to make the model deterministic across constructions. NF)r}r@r<Tr)r#r'rr(r+r-rjr)r=r>r@r;r<rr)r)r*rBxs.   zNestedWrappedModule.init)NF)r$r%r&rGrHrJr+rr4rr7r:rFr#r r rIr rKrLrBrr)r)rr*rIs( rc s>eZdZedejeeee e e fe dfdd Z ZS)AlwaysWrapNestedWrappedModuleNFr=r>r@r;r<csfttt}|j|tj|||d}|tjkr.|S|tjkrbt|fdti|}|tj kr^| }|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. rrN) rrrBr#r'r(rdrr+r-rj)r=r>r@r;r<Zsuper_rMrrr)r*rBs    z"AlwaysWrapNestedWrappedModule.init)NF)r$r%r&rFrGrHr#r+r r rIr rJrBrr)r)rr*rsrcsjeZdZdZejeedfdd ZddZddZ d d Z d d Z e e eeeeed 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||_||_||_dSrX)rrrrr)r2rrrrr)r*rs zModuleWithDelay.__init__cCs |j|SrX)rr4r1r)r)r*r4szModuleWithDelay.get_inputcCs ||SrXrrr)r)r*rszModuleWithDelay.forwardcCs4|j||}|jdkr0tjt|jt|SNr)rr7rrDrj_sleeprr"rr)r)r*r7s zModuleWithDelay.get_lossc s>tjjfdd}td|j|W5QRXdS)Ncs,jdkr"tjtjt||Sr)rrDrjrrr")rtrZorig_reduce_scatterr2r)r*_delayed_reduce_scatters  z=ModuleWithDelay.run_backward.._delayed_reduce_scatterz&torch.distributed._reduce_scatter_base)rD distributedZ_reduce_scatter_baserpatchrr:)r2r9rr)rr*r:szModuleWithDelay.run_backward) module_class model_argsrr model_kwargscOst|j||||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()``. )rrB)rrrrrr)r)r*rBs  zModuleWithDelay.init)r$r%r&rCrKrLrrr4rr7r:rFrr.r rBrr)r)rr*rs  rc sNeZdZeejddddfejeee e e e fe eedfdd ZZS)NestedWrappedModuleWithDelayNFrr=r>r@r;r<rrc s tttjt|||||||dS)Nr)rrrBrrrr)r*rB s z!NestedWrappedModuleWithDelay.init)r$r%r&rFr+r-rGrHr#r r rIr rJrrBrr)r)rr*r srcs$eZdZfddZddZZS)DummyDDPcst||_dSrX)rrr)r2rrr)r*r#s zDummyDDP.__init__cOs |j||SrXr)r2rtrr)r)r*r'szDummyDDP.forwardr$r%r&rrrr)r)rr*r"s rc sheZdZejeeeedfdd ZddZ ddZ e deje ee eeefeed d d ZZS)MixtureOfExperts)r=r}r@delay_before_free_msr<c stj||||d||_||_||_|tjk|_|rFt d|j d}d}d} t t |||j} tdd| D|_| D] } d| _q|rt d t t |||j} |rtj| g} t| | f|} t| |f|} t t t | ||j| | t t || |j|_dS) Nr*rrrcSsg|] }|qSr))ZnumelrSpr)r)r*rVGsz-MixtureOfExperts.__init__..Tr)rrr=rr}r+r,r{rDrrr|rKrrrfZnum_expert_paramsexpertrZ new_grouprdrr)r2r=r}r@rr<r;Zd_expertZd_sharedZd_inputrrZsharedZ expert_grouprr)r*r,sD    zMixtureOfExperts.__init__c sjdkrxjd}t|trxjdjfdd}t|dsJtdtj |d||W5QRSQRX|S)Nrrcs tjtjt|SrX)rDrjrrrr")rtZ orig_reshardr2r)r*_free_full_params_with_delayfsz>MixtureOfExperts.forward.._free_full_params_with_delay_reshardz4expert FSDP module should have a `_reshard()` method) rr isinstancerdrhasattrAssertionErrorrrobject)r2rrrr)rr*r`s$    zMixtureOfExperts.forwardc Csb||js^tB|D]2}t|dr0q |j|jtj j |j|j dq W5QRXdS)NrrZ) rr}rDrgrfrgradZdiv_r^r all_reducer=)r2r9rr)r)r*r:vs   zMixtureOfExperts.run_backwardNFr)r=r>r@r;r<rcCst|dkr i}|tjkr(t|d|||dS|tjkrbt|fd|||d|}|tjkr^|}|Std|dS)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. cuda_init_mode (CUDAInitMode): Determines model movement to CUDA. 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)r}r@rr<Tr)r#r'rr(r+r-rjr)r=r>r@r;r<rrr)r)r*rBs2   zMixtureOfExperts.init)NFr)r$r%r&rGrHrJr+rrrr:rFr#r r rIr rBrr)r)rr*r+s(4 rcs,eZdZfddZeddZeddZeddZd d Zd d Z d dZ e e e efeedddZeddZdejeeeeeeeeefeeeeed ddZdddeddddddddf eeee eeeeeee!ee"eeeeeeeeefee e efdddZ#Z$S) FSDPTestcstt||dSrX)rrsetUpZ_spawn_processesrrr)r*rszFSDPTest.setUpcCstjrtjSdS)Nr)rDrj is_available device_countrr)r)r*r^szFSDPTest.world_sizecCs tjSrX)rGZdistributed_c10dZ_get_default_grouprr)r)r*rNszFSDPTest.process_groupcCsdjt|jdS)Nz {}{file_name}) file_name)formatr!rrr)r)r* init_methodszFSDPTest.init_methodcCs|||jdSrX) assertEqualrm)r2rrmr)r)r*_check_cpu_offloadszFSDPTest._check_cpu_offloadcCs|||jdSrX)rbackward_prefetch)r2rrr)r)r*_check_backward_prefetchsz!FSDPTest._check_backward_prefetchcCs|||jdSrX)rforward_prefetch)r2rrr)r)r*_check_forward_prefetchsz FSDPTest._check_forward_prefetch)subtest_configtest_fn test_kwargsc Os~t|}dd|D}dd|D}tj|D]F}ddt||D} |jf| |||| W5QRXtq2dS)a Runs a test function given by ``test_fn`` as a subtest according to the configurations specified by ``subtest_config``. This amortizes the costly setup overhead (including process spawn and initializing the process group) over the subtests. Args: subtest_config (Dict[str, List[Any]]): A mapping from subtest keyword argument name to a list of its possible values. test_fn (Callable): A callable that runs the actual test. test_args: Positional arguments to pass to ``test_fn``. test_kwargs: Keyword arguments to pass to ``test_fn``. cSsg|] }|dqS)rr)rSitemr)r)r*rVsz)FSDPTest.run_subtests..cSsg|] }|dqS)r[r)rr)r)r*rVscSsi|]\}}||qSr)r))rSkwargvaluer)r)r* sz)FSDPTest.run_subtests..N)rxitems itertoolsproductr]ZsubTestrGbarrier) r2rrZ test_argsrZsubtest_config_itemsZsubtest_config_keysZsubtest_config_valuesvaluesZsubtest_kwargsr)r)r* run_subtestss zFSDPTest.run_subtestsc Cs||}||_||_td|jd|jtjr:dnd}z tj|j |t |j|jdWnBt k r}z$d|j dkrt tdjW5d}~XYnXtjrtjrtj|jtjt|||ttt ddS) Nz dist init r=z, world=ZncclZgloo)rbackendr^rZ recompilerZbackend_unavailable)rrprintr^rDrjrrGZinit_process_grouprr RuntimeErrorrtsysexitrZ exit_coderZ set_devicerZrun_testZdestroy_process_group)clsrZ test_namerpiper2rer)r)r*_runs.  z FSDPTest._run{Gz?NF) rM num_stepsautocastlrfsdp_cpu_offload norm_type save_modelmixed_precisionenable_sharded_grad_scaler use_pure_fp16c  Cs||o|j} t|j} t| d} tjj||dd}t|D]}| tj j j |d|j td}| s|rt|tst|tjr|}ntdd|D}||}| rt|tr|D]}||jtdq|j ||| }W5QRX| |}|s0| s0|jtjksvtdnF| rH||jtjn.t|trf||j|jn||jtj|j ||dk rd }t|tr|||t|||j }n tj!j"|||t#||}|$||k| rt|tr|D]}||jtdq| %|| &|r@d d |'(D}t)||*|q@t|trt|+t,j-|.S) N)Zenabledg?)rZmomentumrjcss|]}|VqdSrX)rk)rSrr)r)r* 5sz4FSDPTest._train_for_several_steps..rRzeloss data type should be float32, as the original parameter data type is float32.g333333?cSsi|]\}}||qSr))clone)rSkvr)r)r*rfsz5FSDPTest._train_for_several_steps..)/rnextrfr3rrDoptimZSGDr\Z zero_gradrjamprrr4rrdrErktuplerr7toZscalerfloat32rZfloat16Z param_dtyper:Zclip_grad_norm__collect_total_grad_norm_fsdprrKutils_collect_total_grad_norm_local assertTruestepupdaterlrriZload_state_dictZ _assert_staterZIDLErQ)r2rMr rrrrrrrrZcpu_offload_paramsZ model_deviceZsharded_grad_scalerrrYr5r6rr9Zmax_normZtotal_norm_after_cliprlr)r)r*_train_for_several_stepssp                  z!FSDPTest._train_for_several_stepsrT) model_classr>r@ ref_init_fn num_itersrrmrsharding_strategyrrrrrrAc! Ksj|tjkstd|dkri}d}|j}|j|jtjtjfddi|}|dkrft||g|d}n||}| rz| }|j ||| dk ||| || | d }t | }| ||| | | dz"|j|j|||fddi|}Wn<tk r}ztd |d t|W5d}~XYnXt|ts8t||jf|}| rF| }|tjkrZ|}|dk oh|j}|ox|tjk}|o|tjk}|rtd }| D]}||j|q|r|td nt}|$|j ||d |||| || | d }W5QRX|rdS|r4| D]}||j|q|}t|} tj||| dk rf|j|| ddddS)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 FSDPNr r<T)Z device_idsZ output_device)rrrrrrr)rmrr*rrz Initializing z raised error rRzExpected param to be on CPUF)rrrrrrrrzFSDP did not match DDP)Z exact_devicemsg)r#r'rrNrrBr+r,DDPrkr&rxrfr% ExceptionrrIrrdr-rjrrDr3rassertRaisesRegexrryZtestingZassert_allclose)!r2r'r>r@r(r)rrmrr*rrrrrrAr;rrrMZ ref_modelZref_lossZ ddp_paramsrr rZexpects_device_errorZexpects_cpu_deviceZ cpu_devicerUcontextZ fsdp_lossZfsdp_unsharded_paramsr)r)r*_test_fsdp_parityps!    *       zFSDPTest._test_fsdp_parity)r NNFNFF)%r$r%r&rpropertyr^rNrrrrr rIr r r r classmethodr rKrLrrJfloatr rrrr&rr.r#r+rrr0rr)r)rr*rs     ! , ]rcs$eZdZfddZddZZS) SkipModulecs ttjdddd|_dSN FZbias)rrrKrlinrrr)r*rs zSkipModule.__init__cCs ||SrX)r8rr)r)r*rszSkipModule.forwardrr)r)rr*r4s r4cs$eZdZfddZddZZS) NestedLinearcsDt|r*ttjdddd|_ntjdddd|_dSr5)rrrrKrrj nested_linear)r2 fsdp_wraprr)r*r s zNestedLinear.__init__cCs ||SrX)r:rr)r)r*rszNestedLinear.forwardrr)r)rr*r9 s r9cs$eZdZfddZddZZS) SkipModelcs@ttjdddd|_t|_tt |d|_ dS)Nr6Fr7)r;) rrrKrrjlinearr4 linear_skiprr9r:)r2Z double_nestrr)r*rs  zSkipModel.__init__cCs"||}||}||}|SrX)r=r>r:rr)r)r*rs   zSkipModel.forwardrr)r)rr*r<s r<cCsVt||}tjjj}|tkr*tjjj}d}tj|||d}tj ||d|d|S)N?r)op) r"rDrZReduceOpZSUMrMAXZtensorrGr)rMrr total_normr@Z return_normr)r)r*r %s   r cCsb|tkrtdd|DSd}|D]&}tjj|j|tjd}|||7}q*|d|SdS)Ncss|]}|jVqdSrX)rabsmaxrr)r)r*r2sz1_collect_total_grad_norm_local..grr?)rrDrfrDZlinalgZ vector_normrr)rMrrBrZ local_normr)r)r*r"0s r")F)FF)T)Vrrrabcrr contextlibrcopyrenumrrmathrtypingr r r r r rrrZunittestrrDZtorch.distributedrrGZtorch.nnrKZtorch.distributed.fsdprrrdZ2torch.distributed.fsdp.fully_sharded_data_parallelrrrrZ*torch.distributed.fsdp.sharded_grad_scalerrZtorch.distributed.fsdp.wraprrrrrZtorch.nn.parallel.distributedrr,Z*torch.testing._internal.common_distributedrr Z$torch.testing._internal.common_utilsr!r"r#r+rLr.rHrbrJrirnruryr|rrrrrrrrrrrr4r9r<r r"r)r)r)r*sp   (         #    {^ D  O